(目录)
一、为什么需要带有 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 { ... @Override public 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 的响应（成功或失败），同时代码将继续执行。