Kafka Consumer 位移提交

位移的概念

每个 consumer 实例都会为它消费的分区维护属于自己的位置信息来记录当前消费了多少条消息。在 Kafka 中，这叫位移 Offset。消费位移记录了 Consumer 要消费的下一条消息的位移。
consumer group 使用一个长整型保存 offset。同时 Kafka consumer 还引入了检查点机制（ checkpointing）定期对 offset 进行持久化，从而简化了应答机制的实现。 Kafka consumer 在内部使用一个 map 来保存其订阅 topic 所属分区的 offset。

位移提交

consumer 客户端需要定期地向 Kafka 集群汇报自己消费数据的进度，这一过程被称为位移提交（Committing Offsets）。因为 Consumer 能够同时消费多个分区的数据，所以位移的提交实际上是在分区粒度上进行的，即 Consumer 需要为分配给它的每个分区提交各自的位移数据。consumer 把位移提交到 Kafka 的一个内部 topic([[__consumer_offsets]])上。
提交位移主要是为了表征 Consumer 的消费进度，这样当 Consumer 发生故障重启之后，就能够从 Kafka 中读取之前提交的位移值，然后从相应的位移处继续消费，从而避免整个消费过程重来一遍。

位移提交的语义保障

假设你的 Consumer 消费了 10 条消息，你提交的位移值却是 50，那么从理论上讲，位移介于 11～49 之间的消息是有可能丢失的；相反地，如果你提交的位移值是 5，那么位移介于 5～9 之间的消息就有可能被重复消费。Kafka 只会 “无脑” 地接受你提交的位移。你对位移提交的管理直接影响了你的 Consumer 所能提供的消息语义保障。

位移管理

consumer 会在 Kafka 集群的所有 broker 中选择一个 broker 作为 consumer group 的
coordinator，用于实现组成员管理、消费分配方案制定以及提交位移等。为每个组选择对应
coordinator 的依据就是 [[__consumer_offsets]]。和普通的 Kafka topic 相同，该 topic 配置有多个分区，每个分区有多个副本。它存在的唯一目的就是保存 consumer 提交的位移。
当消费者组首次启动时，由于没有初始的位移信息， coordinator 必须为其确定初始位移值，
这就是 consumer 参数 [[消费者参数#auto offset reset|auto.offset.reset]] 的作用。通常情况下， consumer 要么从最早的位移开始读取，要么从最新的位移开始读取。
当 consumer 运行了一段时间之后，它必须要提交自己的位移值。如果 consumer 崩溃或被
关闭，它负责的分区就会被分配给其他 consumer，因此一定要在其他 consumer 读取这些分区前就做好位移提交工作，否则会出现消息的重复消费。
consumer 提交位移的主要机制是通过向所属的 coordinator 发送位移提交请求来实现的。
每个位移提交请求都会往 [[__consumer_offsets]]) 对应分区上追加写入一条消息。消息的 key 是 group.id 、 topic 和分区的元组，而 value 就是位移值。如果 consumer 为同一个 group 的同一个 topic 分区提交了多次位移，那么[[__consumer_offsets]]) 对应的分区上就会有若干条 key 相同但 value 不同的消息，但显然我们只关心最新一次提交的那条消息。从某种程度来说，只有最新提交的位移值是有效的，其他消息包含的位移值其实都已经过期了。 Kafka 通过压实（compact）策略来处理这种消息使用模式。

位移提交

自动提交

默认情况下（ enable.auto.commit = true ），consumer 是自动提交位移的，自动提交间隔是 5 秒。通过设置 auto.commit.interval.ms 参数可以控制自动提交的间隔。

自动提交的缺点

假设现在自动提交间隔是 5 秒，并且在提交位移之后的 3 秒发生了 [[Rebalance]] 操作。在 [[Rebalance]] 之后，所有 Consumer 从上一次提交的位移处继续消费，但该位移已经是 3 秒前的位移数据了，故在 Rebalance 发生前 3 秒消费的所有数据都要重新再消费一次。虽然可以通过减少 auto.commit.interval.ms 的值来提高提交频率，但这么做只能缩小重复消费的时间窗口，不可能完全消除它。这是自动提交机制的一个缺陷。

手动提交

手动位移提交就是用户自行确定消息何时被真正处理完并可以提交位移。手动提交位移 API 进一步细分为同步手动提交和异步手动提交，即 commitSync 和 commitAsync 方法。

commitSync()

如果调用的是 commitSync，用户程序会等待位移提交结束才执行下一条语句命令。调用 commitSync() 时，Consumer 程序会处于阻塞状态，直到远端的 Broker 返回提交结果，这个状态才会结束。这可能会影响整个应用程序的 TPS。如果提交过程中出现异常，该方法会将异常信息抛出。

while (true) {ConsumerRecords<String, String> records =consumer.poll(Duration.ofSeconds(1));// 执行业务逻辑process(records);try {consumer.commitSync();} catch (CommitFailedException e) {handleAndLog(e);}
}

commitAsync()

commitAsync() 是一个异步非阻塞调用。调用 commitAsync() 之后，它会立即返回，不会阻塞，因此不会影响 Consumer 应用的 TPS。Kafka 给 commitAsync() 方法提供了回调函数。

while (true) {ConsumerRecords<String, String> records =consumer.poll(Duration.ofSeconds(1));// 执行业务逻辑process(records);consumer.commitAsync((offsets,e)->{if(e != null){handleAndLog(e);}});
}

consumer 在后续 poll 调用时轮询该位移提交的结果。特别注意的是，这里的异步提交位移不是指 consumer 使用单独的线程进行位移提交。实际上 consumer 依然会在用户主线程的 poll 方法中不断轮询这次异步提交的结果。只是该提交发起时此方法是不会阻塞的，因而被称为异步提交。commitAsync() 相较于 commitSync()，它不会自动重试。因为它是异步操作，倘若提交失败后自动重试，那么它重试时提交的位移值可能早已经「过期」或不是最新值了。因此，异步提交的重试其实没有意义，所以 commitAsync 是不会重试的。
当用户调用 consumer.commitSync() 或 consumer.commitAsync() 时， consumer 会为所有它订阅的分区提交位移。

结合 commitSync()和 commitAsync()

    try {while(true) {ConsumerRecords<String, String> records =consumer.poll(Duration.ofSeconds(1));process(records);// 异步提交commitAysnc();}}catch(Exception e) {handle(e);} finally {try {// 同步提交consumer.commitSync();} finally {consumer.close();}}

这段代码同时使用了 commitSync() 和 commitAsync()。对于常规性、阶段性的手动提交，我们调用 commitAsync() 避免程序阻塞，而在 Consumer 要关闭前，我们调用 commitSync() 方法执行同步阻塞式的位移提交，以确保 Consumer 关闭前能够保存正确的位移数据。将两者结合后，我们既实现了异步无阻塞式的位移管理，也确保了 Consumer 位移的正确性。

细粒度提交位移

commitSync() 和 commitAsync() 方法还有另外带参数的重载方法 commitSync(Map<TopicPartition, OffsetAndMetadata>) 和 commitAsync(Map<TopicPartition, OffsetAndMetadata>)。用户调用这个版本的方法时需要指定一个 Map 显式地告诉 Kafka 为哪些分区提交位移。它们的参数是一个 Map 对象，键就是 TopicPartition，即消费的分区，而值是一个 OffsetAndMetadata 对象，保存的主要是位移数据。

 private Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>();int count = 0;while (true) {ConsumerRecords<String, String> records =consumer.poll(Duration.ofSeconds(1));for (ConsumerRecord<String, String> record: records) {process(record);offsets.put(new TopicPartition(record.topic(), record.partition()),// 要提交下一条消息的位移，所以要 +1new OffsetAndMetadata(record.offset() + 1)；// 每 100 条记录提交一次if（count % 100 == 0）{consumer.commitAsync(offsets, null);}count++;}}