Spring Cloud Stream RabbitMQ消费者消息重复投递问题排查

问题根源 #

你遇到的重复投递是RabbitMQ消费者超时机制导致的，和Spring Cloud Stream的自动确认模式、重试配置无关：

• 默认情况下，RabbitMQ的consumer_timeout为3分钟，若消费者在这个时间内未发送任何确认（ACK/NACK），RabbitMQ会判定消费者已失去响应，将未确认的消息重新投递给队列。
• Spring Cloud Stream的AUTO确认模式是在业务方法完全执行完毕后才发送ACK，断点暂停导致业务方法执行超时，触发了RabbitMQ的重投逻辑。
• maxAttempts=1是Spring层面的重试配置，只管控Spring内部的重试，对RabbitMQ原生的消息重投无效。

关键属性解释 #

• Redelivered=true：RabbitMQ标记该消息是重新投递的。
• deliveryAttempt=1：这是Spring的重试计数，当前重复投递是RabbitMQ触发的，不属于Spring重试流程，所以计数仍为1。
• 重复消息无x-death头：x-death是消息从死信交换机投递到Queue2时添加的元数据，而重复投递是RabbitMQ将Queue2中未确认的消息重新入队，并非死信流程触发，因此没有该头。

解决方案 #

方案1：延长RabbitMQ消费者超时时间 #

修改RabbitMQ的全局配置，延长consumer_timeout，避免断点暂停触发重投：

• 在RabbitMQ配置文件（rabbitmq.conf）中添加：

  consumer_timeout = 3600000 # 1小时，可根据需求调整
  

• 注意：这是全局配置，会影响所有消费者，需评估对其他业务的影响。

方案2：改用手动确认模式 #

手动控制ACK时机，在业务逻辑开始前就确认消息，即使后续暂停也不会触发重投：

1. 修改Spring Cloud Stream配置：

   spring:
     cloud:
       stream:
         rabbit:
           bindings:
             your-input-binding-name: # 替换为你的消费者绑定名称
               consumer:
                 acknowledge-mode: MANUAL
   

2. 在消费代码中手动发送ACK：

   @StreamListener("your-input-binding-name")
   public void processMessage(Message<String> message,
                             @Header(AmqpHeaders.CHANNEL) Channel channel,
                             @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag) throws IOException {
       // 先确认消息，告知RabbitMQ无需重投
       channel.basicAck(deliveryTag, false);
       // 再执行业务逻辑，即使断点暂停也不会触发重复投递
       // ... 你的业务处理代码
   }
   

   注意：此方式下如果业务逻辑执行失败，消息会丢失，需结合业务场景评估是否需要配套失败重试或死信机制。

方案3：调整消费者预取数（缓解方案） #

设置prefetchCount=1，确保同一时间消费者只处理一条消息，减少重投概率，但无法彻底解决超时重投问题：

spring:
  cloud:
    stream:
      rabbit:
        bindings:
          your-input-binding-name:
            consumer:
              prefetch: 1

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