点击上方蓝色“大数据实战演练”，选择“设为星标”或“置顶”

回复“资源”领取独家整理的学习资料！

每一个成功人士的背后，必定曾经做出过勇敢而又孤独的决定。

放弃不难，但坚持很酷~

kafka_2.11-1.1.0

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();

欢迎大家留言讨论

???? ???? ????

往期推荐

两种实现方式 | 如何查看消费者组的消费情况

Kafka消费者 之 指定位移消费

Kafka消费者 之 如何订阅主题或分区

Kafka消费者 之 如何进行消息消费

Kafka消费者 之 如何提交消息的偏移量

Kafka基础（二）：生产者相关知识汇总

Kafka基础（一）：基本概念及生产者、消费者示例

扫一扫，我们的故事就开始了。

如果这篇文章对你有所启发，点赞、转发都是一种支持！

让我知道你在看