Kafka消费者重启后如何从指定偏移量（第6条消息）开始消费？

要实现消费者宕机恢复后从第6条消息（上次消费完成的下一条）继续消费，核心是精准管理消费者组的偏移量提交逻辑，结合你提供的Apache Camel代码，具体实现步骤如下：

1. 完善生产者代码（补充必要配置） #

原生产者代码缺少定时器触发间隔，补充后添加中文注释：
Producer.java

public class KafkaProducerRoute extends RouteBuilder {
    @Override
    public void configure() throws Exception {
        // 每秒触发一次，持续生产消息
        from("timer:time?period=1000")
                .setBody().constant("Hello, World!") // 设置消息内容
                .to("kafka:hello1?brokers=localhost:9092") // 发送到hello1主题
                .log("消息已生产");
    }
}

2. 改造消费者代码（关键配置+手动提交偏移量） #

核心是关闭自动提交，改为消息消费成功后手动提交偏移量，确保偏移量只在消息处理完成后更新：
Consumer.java

from("kafka:hello1?brokers=localhost:9092&groupId=c1&autoCommitEnable=false&autoOffsetReset=earliest")
        .process(exchange -> {
            // 获取消息内容和偏移量
            String msgContent = exchange.getIn().getBody(String.class);
            long offset = exchange.getIn().getHeader(KafkaConstants.OFFSET, Long.class);
            System.out.printf("收到消息，偏移量：%d - 内容：%s%n", offset, msgContent);

            // 消息处理完成后，手动提交偏移量
            KafkaManualCommit manualCommit = exchange.getIn().getHeader(KafkaConstants.MANUAL_COMMIT, KafkaManualCommit.class);
            if (manualCommit != null) {
                manualCommit.commitSync();
            }
        });

3. 关键配置说明 #

• groupId=c1：必须指定消费者组ID，Kafka通过消费者组维护该组的消费偏移量，同一组的消费者共享偏移量记录。
• autoCommitEnable=false：关闭Kafka自动提交偏移量的功能，避免消息还没处理完就被自动提交偏移量，导致宕机后重复消费或跳过消息。
• autoOffsetReset=earliest：当消费者组没有历史偏移量记录时（比如首次启动），从最早的消息开始消费；如果已有偏移量记录，会优先使用已提交的偏移量继续消费。

4. 偏移量丢失后的补救方案 #

如果之前未配置手动提交导致偏移量丢失，可手动重置消费者组的偏移量到第6条消息对应的偏移值（假设第5条消息偏移量为4，第6条则为5），用Kafka命令行工具执行：

kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group c1 --topic hello1 --reset-offsets --to-offset 5 --execute

内容的提问来源于stack exchange，提问作者Aayush Saini