kafka学习笔记（三、消费者Consumer使用教程——从指定位置消费）

1.简介

Kafka的poll()方法消费无法精准的掌握其消费的起始位置，auto.offset.reset参数也只能在比较粗粒度的指定消费方式。更细粒度的消费方式kafka提供了seek()方法可以指定位移消费允许消费者从特定位置（如固定偏移量、时间戳或分区首尾）开始消费消息。

2.指定消费位置

2.1.从特定偏移量开始消费

使用seek(TopicPartition partition, long offset)指定具体偏移量。

源码分析：

• seek()方法更新消费者内部的subscriptions对象的position字段，记录目标偏移量。
• 后续poll()时，Fetcher类根据此位置向Broker发送拉取请求。

  代码示例：

  consumer.subscribe(Collections.singleton("test-topic"));
  Set assignment = new HashSet();
  // 确保分配到分区
  while (assignment.isEmpty()) {
      consumer.poll(Duration.ofMillis(100));
      assignment = consumer.assignment();
  }
  // 设置所有分区从offset=100开始消费
  assignment.forEach(tp -> consumer.seek(tp, 100));
  

  2.2.从时间戳开始消费

  使用offsetsForTimes()获取时间戳对应的偏移量，再调用seek()。

  源码分析：

  offsetsForTimes()向Broker发送ListOffsetRequest，查询满足时间戳条件的最早或最新偏移量。

  代码实例：

  Map timestamps = assignment.stream()
      .collect(Collectors.toMap(tp -> tp, tp -> System.currentTimeMillis() - 24 * 3600 * 1000L));
  // 获取24小时前的偏移量
  Map offsets = consumer.offsetsForTimes(timestamps);
  offsets.forEach((tp, offsetAndTs) -> {
      if (offsetAndTs != null) consumer.seek(tp, offsetAndTs.offset());
  });
  

  2.3.从分区首尾消费

  使用seekToBeginning()或seekToEnd()，或通过beginningOffsets()/endOffsets()获取首尾偏移量后手动设置。

  代码实例：

  // 从分区末尾开始消费（等效于auto.offset.reset=latest）
  Map endOffsets = consumer.endOffsets(assignment);
  assignment.forEach(tp -> consumer.seek(tp, endOffsets.get(tp)));
  

  2.4.注意事项

  1. 分区分配与poll()的依赖

     seek()必须在分区分配完成后调用，否则会抛出IllegalStateException。需通过循环poll()确保分配到分区。

  2. 数据过期问题

     若指定偏移量对应的消息已被删除（如日志清理导致），seek()将失效。此时需使用beginningOffsets()获取当前最小有效偏移量。

  3. 异步提交与位移覆盖风险

     异步提交（commitAsync()）失败时不会重试，可能因位移回滚导致重复消费。需结合同步提交（commitSync()）保证原子性。

  4. seek()方法提供了我们可以将消费者位移保存在外部的能力，还可以配合在均衡监听器来提供更加精准的消费能力。

  3.完整代码实例

  public class SeekToTimestampDemo {
      public static void main(String[] args) {
          Properties props = new Properties();
          props.put("bootstrap.servers", "localhost:9092");
          props.put("group.id", "seek-demo");
          props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
          props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
          props.put("enable.auto.commit", "false");
          KafkaConsumer consumer = new KafkaConsumer(props);
          consumer.subscribe(Collections.singleton("test-topic"));
          // 等待分区分配
          Set assignment = new HashSet();
          while (assignment.isEmpty()) {
              consumer.poll(Duration.ofMillis(100));
              assignment = consumer.assignment();
          }
          // 获取24小时前的时间戳对应偏移量
          Map timestamps = assignment.stream()
              .collect(Collectors.toMap(tp -> tp, tp -> System.currentTimeMillis() - 86400000L));
          Map offsets = consumer.offsetsForTimes(timestamps);
          // 指定位移
          offsets.forEach((tp, offsetAndTs) -> {
              if (offsetAndTs != null) {
                  consumer.seek(tp, offsetAndTs.offset());
              } else {
                  // 处理无有效偏移量的情况（如从头开始）
                  consumer.seekToBeginning(Collections.singleton(tp));
              }
          });
          while (true) {
              ConsumerRecords records = consumer.poll(Duration.ofMillis(1000));
              records.forEach(record -> System.out.printf("offset=%d, value=%s%n", record.offset(), record.value()));
          }
      }
  }