标签：异步 offsets partition kafka 提交 offset entry commitOffset

转：https://blog.csdn.net/CREATE_17/article/details/108722808

kafka手动异步提交 offset 的步骤大概分为以下几步，如下图所示：

1、配置手动提交

enable.auto.commit 修改为 false 。

2、订阅 topic

consumer.subscribe(Arrays.asList("topic name"));

3、获取 topic 各分区当前读取到的最后一条记录的offset
首先定义一个全局变量：

//用来记录当前消费的偏移
private static Map<TopicPartition, Long> offsets = new HashMap<>();
for (TopicPartition partition : records.partitions()) {
List<ConsumerRecord<String, String>> partitionRecords = records.records(partition);
// 获取当前读取到的最后一条记录的offset
long lastOffset = partitionRecords.get(partitionRecords.size() - 1).offset();
// 提交offset
offsets.put(partition, lastOffset + 1);
}
至于为什么消费者提交 offsets 时要 +1，在《Kafka消费者 之 如何提交消息的偏移量》中的概述章节里面也给出了答案。

4、手动异步提交 offset
首先定义一个全局变量：

//用来记录当需要提交的偏移
private static Map<TopicPartition, OffsetAndMetadata> commitOffset = new HashMap<>();
//
for (Map.Entry<TopicPartition, Long> entry : offsets.entrySet()) {
commitOffset.put(entry.getKey(), new OffsetAndMetadata(offsets.get(entry.getKey())));
logger.info("partition[{}], 当前待提交kafka偏移:[{}]", entry.getKey().partition(), offsets.get(entry.getKey()));
}
// 异步提交offset
consumer.commitAsync(commitOffset, (offsets, exception) -> {
if (exception != null) {
logger.error("fail to commit offsets {}, {}", offsets, exception);
// 同步提交，来做offset提交最后的保证。
consumer.commitSync();
}
});
清空：

commitOffset.clear();
offsets.clear();

标签：异步,offsets,partition,kafka,提交,offset,entry,commitOffset
来源： https://www.cnblogs.com/qsds/p/16688979.html

本站声明： 1. iCode9 技术分享网（下文简称本站）提供的所有内容，仅供技术学习、探讨和分享；
2. 关于本站的所有留言、评论、转载及引用，纯属内容发起人的个人观点，与本站观点和立场无关；
3. 关于本站的所有言论和文字，纯属内容发起人的个人观点，与本站观点和立场无关；
4. 本站文章均是网友提供，不完全保证技术分享内容的完整性、准确性、时效性、风险性和版权归属；如您发现该文章侵犯了您的权益，可联系我们第一时间进行删除；
5. 本站为非盈利性的个人网站，所有内容不会用来进行牟利，也不会利用任何形式的广告来间接获益，纯粹是为了广大技术爱好者提供技术内容和技术思想的分享性交流网站。