KafkaConsumer断点续读：停止重启后从上次位置读取的实现方法

要实现重启后从上次停止位置继续消费，核心是持久化消费位移，下面是几种常用的实现方式：

1. 自动提交消费者位移（最简单的方式） #

Kafka默认支持自动提交位移，只需确保以下配置正确：

• 开启自动提交：enable.auto.commit=true（默认值）
• 设置提交间隔：auto.commit.interval.ms=5000（默认5秒，可根据业务调整）

这种方式下，消费者会定期自动把当前拉取到的消息位移提交到Kafka内置的__consumer_offsets主题中。重启时，消费者会从该主题拉取所属消费者组的最新位移，继续消费。

注意：自动提交的是拉取到的消息位移，而非处理完成的消息位移。如果在提交后、处理前重启，会出现重复消费；如果处理完但还没提交就重启，会丢失部分消息。适合对数据一致性要求不高的场景。

2. 手动提交位移（精确控制，推荐） #

如果需要严格保证数据不丢不重，建议关闭自动提交，手动控制位移提交时机：

• 关闭自动提交：enable.auto.commit=false

同步提交 #

处理完一批消息后，调用commitSync()方法同步提交位移，该方法会阻塞直到提交成功：

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    for (ConsumerRecord<String, String> record : records) {
        // 处理消息逻辑
        processRecord(record);
    }
    // 全部处理完成后提交位移
    consumer.commitSync();
}

异步提交 #

如果不想阻塞，可使用commitAsync()异步提交，还能添加回调处理提交失败的情况：

consumer.commitAsync((offsets, exception) -> {
    if (exception != null) {
        // 处理提交失败逻辑，比如记录日志、重试
        log.error("提交位移失败: {}", offsets, exception);
    }
});

手动提交能保证只有消息处理完成后才提交位移，重启时不会丢失已处理的消息，也能最大程度减少重复消费的概率。

3. 自定义位移存储（灵活扩展） #

如果不想依赖Kafka内置的位移存储，或者需要和业务数据一起持久化，可以自己实现位移存储，比如存到MySQL、Redis等：

1. 启动时定位位移：消费者初始化后，从自定义存储中读取上次消费的位移（按分区存储），调用seek(TopicPartition partition, long offset)方法定位到指定位置：

// 获取主题的所有分区
List<TopicPartition> partitions = consumer.partitionsFor("topic-name").stream()
    .map(p -> new TopicPartition(p.topic(), p.partition()))
    .collect(Collectors.toList());
// 从Redis读取每个分区的上次位移
for (TopicPartition partition : partitions) {
    long lastOffset = redisTemplate.opsForValue().get("kafka-offset:" + partition.topic() + ":" + partition.partition());
    consumer.seek(partition, lastOffset);
}

2. 消费后持久化位移：处理完消息后，把当前分区的最新位移写入自定义存储，建议和消息处理放在同一个事务中，保证数据和位移的一致性。

注意：自定义存储需要自己处理位移的原子性和并发问题，比如多个消费者实例消费同一个分区时的位移冲突。

关键注意事项 #

• 必须保证消费者组ID（group.id）一致，Kafka是通过消费者组ID来关联位移的，不同组ID会被视为新的消费者，从头开始消费。
• 注意位移过期时间：Kafka默认会保留位移offsets.retention.minutes=10080（7天），如果消费者长时间不启动，位移可能被清理，导致重启后从头消费，可根据业务调整该参数。

内容的提问来源于stack exchange，提问作者Mohit Kumar