为什么kafka 需要 subscribe 的 group.id？我们是否需要使用 commitSync 手动提交偏移量？

一、为什么需要带有 subscribe 的 group.id

• 消费概念：
  Kafka 使用消费者组的概念来实现主题的并行消费 - 每条消息都将在每个消费者组中传递一次，无论该组中实际有多少个消费者。所以 group 参数是强制性的，如果没有组，Kafka 将不知道如何对待订阅同一主题的其他消费者。
• 偏移量：
  每当我们启动一个消费者时，它都会加入一个消费者组，然后根据该消费者组中的其他消费者数量，为其分配要读取的分区。对于这些分区，它会检查列表读取偏移量是否已知，如果找到，它将从这一点开始读取消息。如果没有找到偏移量，则参数 auto.offset.reset 控制是从分区中最早的消息还是从最新的消息开始读取。

二、我们需要使用commitSync手动提交偏移量吗？

• 是否需要手动提交偏移？
  是否需要提交偏移量取决于作为参数 enable.auto.commit 选择的值。默认情况下，此设置为 true，这意味着消费者将定期自动提交其偏移量（由auto.commit.interval.ms 决定提交的频率）。如果将其设置为 false，那么将需要自己提交偏移量。这种默认行为可能也是导致很多发现 kafka 总是从最新的开始消费的原因，由于偏移量是自动提交的，因此它将使用该偏移量。

• 有没有办法从头开始重播消息？
  如果想每次都从头开始读取，可以调用seekToBeginning，如果不带参数调用，它将重置为所有订阅分区中的第一条消息，或者仅重置您传入的那些分区。

• seekToBeginning
  查找每个给定分区的第一个偏移量。poll(long) 该函数延迟计算，仅在调用或时才查找所有分区中的第一个偏移量position(TopicPartition)。如果未提供分区，则查找所有当前分配的分区的第一个偏移量。

  public class MyListener implements ConsumerSeekAware {...@Overridepublic void onPartitionsAssigned(Map<TopicPartition, Long> assignments, ConsumerSeekCallback callback) {callback.seekToBeginning(assignments.keySet());}}
  

• 有没有办法从最后开始重播消息？
  有的，可以使用 seekToEnd() 查找所有分配的分区到最后。或者使用 seekToTimestamp(long time)- 查找所有分配的分区到该时间戳表示的偏移量。

  public class MyListener extends AbstractConsumerSeekAware {@KafkaListener(...)void listn(...) {...}
  }public class SomeOtherBean {MyListener listener;...void someMethod() {this.listener.seekToTimestamp(System.currentTimeMillis - 60_000);}}
  

三、如果我想手动提交偏移量，该怎么做？

• 1、禁用自动提交

  props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
  

• 提交方法
  对于手动提交，KafkaConsumers提供了两种方法，即 commitSync() 和 commitAsync()。commitSync()是一个阻塞调用，在偏移量成功提交后返回，commitAsync()则立即返回。如果想知道提交是否成功，可以为回调处理程序 ( OffsetCommitCallback) 提供一个方法参数。请注意，在两次提交调用中，消费者都会提交最新poll()调用的偏移量。
  举个例子：假设一个分区主题有一个消费者并且最后一次调用poll()返回偏移量为 4、5、6 的消息。提交时，偏移量 6 将被提交，因为这是消费者客户端跟踪的最新偏移量。
  同时，commitSync() 和 commitAsync() 都允许更多地控制我们想要提交的偏移量：如果你使用允许你指定的相应重载，那么Map<TopicPartition, OffsetAndMetadata>消费者将仅提交指定的偏移量（即，映射可以包含分配的分区的任何子集） ，并且指定的偏移量可以为任意值）。

• 同步提交：
  阻塞线程，直到提交成功或遇到不可恢复的错误（在这种情况下，它被抛出给调用者）

  while (true) {ConsumerRecords<String, String> records = consumer.poll(100);for (ConsumerRecord<String, String> record : records) {System.out.printf("offset = %d, key = %s, value = %s", record.offset(), record.key(), record.value());consumer.commitSync();}
  }
  

  对于 for 循环中的每次迭代，只有在consumer.commitSync()成功返回或因抛出异常而中断后，代码才会移至下一次迭代。

• 异步提交：
  是一种非阻塞方法。调用它不会阻塞线程。相反，它将继续处理以下指令，无论最终是成功还是失败。

  while (true) {ConsumerRecords<String, String> records = consumer.poll(100);for (ConsumerRecord<String, String> record : records) {System.out.printf("offset = %d, key = %s, value = %s", record.offset(), record.key(), record.value());consumer.commitAsync(callback);}
  }
  

  对于 for 循环中的每次迭代，无论consumer.commitAsync()最终会发生什么，代码都会移至下一次迭代。并且，提交的结果将由定义的回调函数处理。

• 权衡：延迟与数据一致性
  1、如果必须确保数据一致性，请选择commitSync()，因为它将确保在执行任何进一步操作之前，你将知道偏移量提交是成功还是失败。但由于它是同步和阻塞的，你将花费更多的时间来等待提交完成，这会导致高延迟。
  2、如果可以接受某些数据不一致并希望具有低延迟，请选择commitAsync()，因为它不会等待完成。相反，它只会发出提交请求并稍后处理来自 Kafka 的响应（成功或失败），同时代码将继续执行。