Spring Kafka如何消费服务宕机期间Topic堆积的未消费消息

问题根因 #

当前代码无法消费宕机期间堆积的消息，核心是Spring Kafka默认消费者配置不符合需求：

• 默认auto-offset-reset参数值为latest：消费者启动时，仅会拉取连接建立后新写入Topic的消息，分区内已存在的历史堆积消息会被跳过
• 若开启自动偏移量提交，会存在消息未完成处理就提交偏移量的风险，重启后无法回溯到未处理的消息位置

具体配置步骤 #

1. 修改Spring Kafka消费者配置 #

在项目的application.yml（或application.properties）中添加/修改如下配置：

spring:
  kafka:
    consumer:
      bootstrap-servers: 替换为实际Kafka集群地址
      group-id: 替换为你实际使用的消费组ID
      # 关闭自动偏移量提交，由Spring监听容器控制提交时机
      enable-auto-commit: false
      # 无已提交偏移量时，从分区最早的可用消息开始消费
      auto-offset-reset: earliest
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
    listener:
      # 单条消息业务处理逻辑正常执行完成后，再提交对应偏移量
      ack-mode: RECORD

如果用properties格式配置，对应核心参数为：

spring.kafka.consumer.enable-auto-commit=false
spring.kafka.consumer.auto-offset-reset=earliest
spring.kafka.listener.ack-mode=RECORD
# 其余序列化、服务地址、消费组参数按实际环境补全即可

2. 配置生效注意事项 #

• 如果你之前已经用当前配置的group-id启动过服务，且之前的auto-offset-reset为latest，Kafka中已经留存了该消费组的已提交偏移量，此时仅改配置不会触发从最早位置消费。解决方式两种：要么换一个全新的未被使用过的消费组ID，要么通过Kafka命令行工具手动重置该消费组的偏移量到对应Topic分区的最早位置。
• 消费逻辑中如果处理消息抛出异常，对应消息的偏移量不会提交，服务重启后会重新拉取该条消息重试，避免消息丢失。
• 不要开启enable-auto-commit=true的自动提交模式，该模式下消费者会按固定时间间隔提交偏移量，和业务逻辑是否处理完成无关，服务宕机时大概率出现消息未处理但偏移量已提交的问题，导致消息丢失。

3. 代码适配 #

你当前写的@KafkaListener标注的消费方法不需要做逻辑改动，只要配置正确，服务重启后就会自动拉取宕机期间堆积的全部消息，按顺序触发消费逻辑。

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