kafka 手动提交的同步提交和异步提交

71 阅读 0 评论 47 点赞

我是靠谱客的博主坚定小天鹅，最近开发中收集的这篇文章主要介绍kafka 手动提交的同步提交和异步提交，觉得挺不错的，现在分享给大家，希望可以做个参考。

概述

自动位移提交的方式在正常情况下不会发生消息丢失或重复消费的现象，但是在编程的世界里异常无可避免，与此同时，自动位移提交也无法做到精确的位移管理。

在Kafka 中还提供了手动位移提交的方式，这样可以使得开发人员对消费位移的管理控制更加灵活。很多时候并不是说拉取到消息就算消费完成，而是需要将消息写入数据库、写入本地缓存，或者是更加复杂的业务处理。在这些场景下，所有的业务处理完成才能认为消息被成功消费，手动的提交方式可以让开发人员根据程序的逻辑在合适的地方进行位移提交。开启手动提交功能的前提是消费者客户端参数enable.auto.commit 配置为false ，示例如下：

properties.put("enable.auto.commit", "false");

手动提交可以细分为同步提交和异步提交，对应于KafkaConsumer 中的commitSync（）和commitAsync（）两种类型的方法。我们这里先讲述同步提交的方式

commitSync（）方法

定义如下：

public void commitSync()

这个方法很简单，下面使用它演示同步提交的简单用法：

while(true) {
		ConsumerRecords<String , String> records= consumer.poll(1000);
		for (ConsumerRecord<String, String> record : records) {
		//do some logical processing .
		}
		consumer.commitSync() ;
	}

可以看到示例中先对拉取到的每一条消息做相应的逻辑处理，然后对整个消息集做同步提交。参考KafkaConsumer 源码中提供的示例，针对上面的示例还可以修改为批量处理＋批量提交的方式，关键代码如下：

上面的示例中将拉取到的消息存入缓存buffer，等到积累到足够多的时候，也就是示例中大于等于200 个的时候，再做相应的批量处理，之后再做批量提交。这两个示例都有重复消费的问题，如果在业务逻辑处理完之后，并且在同步位移提交前，程序出现了崩渍，那么待恢复之后又只能从上一次位移提交的地方拉取消息，由此在两次位移提交的窗口中出现了重复消费的现象。

commitSync （）方法会根据poll（）方法拉取的最新位移来进行提交（注意提交的值对应于图3-6中position 的位置〉，只要没有发生不可恢复的错误（ Unrecoverable Eηor ），它就会阻塞消费者线程直至位移提交完成。对于不可恢复的错误，比如CommitFailedException 、WakeupException 、InterruptException 、AuthenticationException 、AuthorizationException 等，我们可以将其捕获并做针对性的处理。对于采用commitSync（）的无参方法而言，它提交消费位移的频率和拉取批次消息、处理批
次消息的频率是一样的，如果想寻求更细粒度的、更精准的提交，那么就需要使用commitSync()的另一个含参方法，具体定义如下：

public void commitSync(final Map<TopicPartition , OffsetAndMetadata> offsets)

该方法提供了一个offsets 参数，用来提交指定分区的位移。无参的commitSync（）方法只能提交当前批次对应的position 值。如果需要提交一个中间值，比如业务每消费一条消息就提交一次位移，那么就可以使用这种方式，我们来看一下代码示例，如代码清单3-3 所示。

在实际应用中，很少会有这种每消费一条消息就提交一次消费位移的必要场景。commitSync（）方法本身是同步执行的，会耗费一定的性能，而示例中的这种提交方式会将性能拉到一个相当低的点。更多时候是按照分区的粒度划分提交位移的界限，这里我们就要用到了3.2.4 章中提及的ConsumerRecords 类的partitions（）方法和records(TopicPartition）方法，关键示例代码如代码清单3-4 所示（修改自KafkaConsumer 源码中的示例，注意代码中加粗的部分〉。

异步提交的方式（ commitAsync（））

在执行的时候消费者线程不会被阻塞，可能在提交消费位移的结果还未返回之前就开始了新一次的拉取操作。异步提交可以便消费者的性能得到一定的增强。commitAsync 方法有三个不同的重载方法，具体定义如下：

public void commitAsync()
public void commitAsync(OffsetCommitCallback callback)
public void commitAsync(final Map<TopicPartition , OffsetAndMetadata> offsets , OffsetCommitCallback callback)

第一个无参的方法和第三个方法中的offsets 都很好理解，对照commitSync（）方法即可。关键的是这里的第二个方法和第三个方法中的callback 参数，它提供了一个异步提交的回调方法，当位移提交完成后会回调OffsetCommitCallback 中的onComplete（）方法。

commitAsync（）提交的时候同样会有失败的情况发生，那么我们应该怎么处理呢？读者有可能想到的是重试，问题的关键也就在这里了。如果某一次异步提交的消费位移为x ，但是提交失败了，然后下一次又异步提交了消费位移为x+y，这次成功了。如果这里引入了重试机制，前一次的异步提交的消费位移在重试的时候提交成功了，那么此时的消费位移又变为了x 。如果此时发生异常（或者再均衡），那么恢复之后的消费者（或者新的消费者）就会从x 处开始消费消息，这样就发生了重复消费的问题。
为此我们可以设置一个递增的序号来维护异步提交的顺序，每次位移提交之后就增加序号相对应的值。在遇到位移提交失败需要重试的时候，可以检查所提交的位移和序号的值的大小，如果前者小于后者，则说明有更大的位移己经提交了，不需要再进行本次重试：如果两者相同，则说明可以进行重试提交。除非程序编码错误，否则不会出现前者大于后者的情况。如果位移提交失败的情况经常发生，那么说明系统肯定出现了故障，在－般情况下，位移提交失败的情况很少发生，不重试也没有关系，后面的提交也会有成功的。重试会增加代码逻辑的复杂度，不重试会增加重复消费的概率。如果消费者异常退出，那么这个重复消费的问题就很难避免，因为这种情况下无法及时提交消费位移；如果消费者正常退出或发生再均衡的情况，那么可以在退出或再均衡执行之前使用同步提交的方式做最后的把关。