Kafka：消费者位移api介绍

港控/mmm° 2024-03-24 13:32 33阅读 0赞

<table style="width:500px;"> 
 <tbody> 
  <tr> 
   <td>API</td> 
   <td>public void assign(Collection partitions)</td> 
  </tr> 
  <tr> 
   <td>说 明</td> 
   <td>给当前消费者手动分配一系列主题分区。 手动分配分区不支持增量分配，如果先前有分配分区，则该操作会覆盖之前的分配。 如果给出的主题分区是空的，则等价于调用unsubscribe方法。 手动分配主题分区的方法不使用消费组管理功能。当消费组成员变了，或者集群或主题的 元数据改变了，不会触发分区分配的再平衡。 手动分区分配assign(Collection)不能和自动分区分配subscribe(Collection, ConsumerRebalanceListener)一起使用。 如果启用了自动提交偏移量，则在新的分区分配替换旧的分区分配之前，会对旧的分区分 配中的消费偏移量进行异步提交。</td> 
  </tr> 
  <tr> 
   <td>API</td> 
   <td>public Set assignment()</td> 
  </tr> 
  <tr> 
   <td>说 明</td> 
   <td>获取给当前消费者分配的分区集合。如果订阅是通过调用assign方法直接分配主题分区， 则返回相同的集合。如果使用了主题订阅，该方法返回当前分配给该消费者的主题分区集 合。如果分区订阅还没开始进行分区分配，或者正在重新分配分区，则会返回none。</td> 
  </tr> 
  <tr> 
   <td>API</td> 
   <td>public Map&gt; listTopics()</td> 
  </tr> 
  <tr> 
   <td>说 明</td> 
   <td>获取对用户授权的所有主题分区元数据。该方法会对服务器发起远程调用。</td> 
  </tr> 
  <tr> 
   <td>API</td> 
   <td>public List partitionsFor(String topic)</td> 
  </tr> 
  <tr> 
   <td>说 明</td> 
   <td>获取指定主题的分区元数据。如果当前消费者没有关于该主题的元数据，就会对服务器发 起远程调用。</td> 
  </tr> 
  <tr> 
   <td>API</td> 
   <td>public Map beginningOffsets(Collection partitions)</td> 
  </tr> 
  <tr> 
   <td>说 明</td> 
   <td>对于给定的主题分区，列出它们第一个消息的偏移量。 注意，如果指定的分区不存在，该方法可能会永远阻塞。 该方法不改变分区的当前消费者偏移量。</td> 
  </tr> 
  <tr> 
   <td>API</td> 
   <td>public void seekToEnd(Collection partitions)</td> 
  </tr> 
  <tr> 
   <td>说 明</td> 
   <td>将偏移量移动到每个给定分区的最后一个。 该方法延迟执行，只有当调用过poll方法或position方法之后才可以使用。 如果没有指定分区，则将当前消费者分配的所有分区的消费者偏移量移动到最后。 如果设置了隔离级别为：isolation.level=read_committed，则会将分区的消费偏移量移动 到最后一个稳定的偏移量，即下一个要消费的消息现在还是未提交状态的事务消息。</td> 
  </tr> 
  <tr> 
   <td>API</td> 
   <td>public void seek(TopicPartition partition, long offset)</td> 
  </tr> 
  <tr> 
   <td>说 明</td> 
   <td>将给定主题分区的消费偏移量移动到指定的偏移量，即当前消费者下一条要消费的消息偏 移量。 若该方法多次调用，则最后一次的覆盖前面的。 如果在消费中间随意使用，可能会丢失数据。</td> 
  </tr> 
  <tr> 
   <td>API</td> 
   <td>public long position(TopicPartition partition)</td> 
  </tr> 
  <tr> 
   <td>说 明</td> 
   <td>检查指定主题分区的消费偏移量</td> 
  </tr> 
  <tr> 
   <td>API</td> 
   <td>public void seekToBeginning(Collection partitions)</td> 
  </tr> 
  <tr> 
   <td>说 明</td> 
   <td>将给定每个分区的消费者偏移量移动到它们的起始偏移量。该方法懒执行，只有当调用过 poll方法或position方法之后才会执行。如果没有提供分区，则将所有分配给当前消费者的 分区消费偏移量移动到起始偏移量。</td> 
  </tr> 
 </tbody> 
</table>

import org.apache.kafka.clients.consumer.ConsumerConfig;
    import org.apache.kafka.clients.consumer.KafkaConsumer;
    import org.apache.kafka.common.PartitionInfo;
    import org.apache.kafka.common.TopicPartition;
    import org.apache.kafka.common.serialization.StringDeserializer;
    
    import java.util.function.BiConsumer;
    
    public class MyOffsetManager {
        public static void main(String[] args) {
    
            Map<String, Object> configs = new HashMap<>();
            configs.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.25.129:9092");
            configs.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
            configs.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
            // group.id很重要
            configs.put(ConsumerConfig.GROUP_ID_CONFIG, "mygrp1");
    
            KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(configs);
    
    //        consumer.subscribe(Collections.singleton("tp_demo_01"));
    
            // 如何手动给消费者分配分区？
            // 1、需要知道有哪些主题可以访问，和消费
    
            // 获取当前消费者可以访问和消费的主题以及它们的分区信息
            final Map<String, List<PartitionInfo>> stringListMap = consumer.listTopics();
    
            stringListMap.forEach(new BiConsumer<String, List<PartitionInfo>>() {
                @Override
                public void accept(String topicName, List<PartitionInfo> partitionInfos) {
                    System.out.println("主题名称：" + topicName);
                    for (PartitionInfo partitionInfo : partitionInfos) {
                        System.out.println(partitionInfo);
                    }
                }
            });
    
            final Set<TopicPartition> assignment1 = consumer.assignment();
    
            for (TopicPartition partition : assignment1) {
                System.out.println(partition);
            }
            System.out.println("----------------------------");
    
            // 给当前消费者分配指定的主题分区
            consumer.assign(Arrays.asList(
                    new TopicPartition("tp_demo_01", 0),
                    new TopicPartition("tp_demo_01", 1),
                    new TopicPartition("tp_demo_01", 2)
            ));
    
            // 获取给当前消费者分配的主题分区信息
            final Set<TopicPartition> assignment = consumer.assignment();
    
            for (TopicPartition partition : assignment) {
                System.out.println(partition);
            }
    
            // 查看当前消费者在指定主题的分区上的消费者偏移量
            final long offset0 = consumer.position(new TopicPartition("tp_demo_01", 0));
    
            System.out.println("当前主题在0号分区上的位移：" + offset0);
    
            consumer.seekToBeginning(Arrays.asList(
                    new TopicPartition("tp_demo_01", 0),
                    new TopicPartition("tp_demo_01", 2)
            ));
            long offset0 = consumer.position(new TopicPartition("tp_demo_01", 0));
            long offset1 = consumer.position(new TopicPartition("tp_demo_01", 1));
            long offset2 = consumer.position(new TopicPartition("tp_demo_01", 2));
    
            System.out.println(offset0);
            System.out.println(offset1);
            System.out.println(offset2);
    
    //        consumer.seekToEnd(Arrays.asList(new TopicPartition("tp_demo_01", 2)));
    
            consumer.seek(new TopicPartition("tp_demo_01", 2), 14);
    
            offset0 = consumer.position(new TopicPartition("tp_demo_01", 0));
            offset1 = consumer.position(new TopicPartition("tp_demo_01", 1));
            offset2 = consumer.position(new TopicPartition("tp_demo_01", 2));
    
            System.out.println(offset0);
            System.out.println(offset1);
            System.out.println(offset2);
    
            consumer.close();
        }
    
    
    }