RabbitMQ 常用模式：本地重试与死信队列

RabbitMQ 常用模式本地重试与死信队列推荐方案AUTO 确认 Spring 本地有限重试 RejectAndDontRequeueRecoverer RabbitMQ DLX。方案概览本方案适合秒级、少次数、可幂等的消费失败重试。各组件职责如下组件职责Spring Listener Retry在当前消费者线程内执行有限次数的本地重试RejectAndDontRequeueRecoverer重试耗尽后拒绝消息并设置requeuefalseRabbitMQ DLX将被拒绝的消息路由到死信交换机DLQ保存最终处理失败的消息供告警、排查或人工重放[!NOTE]Spring AMQP 的acknowledge-mode: auto表示由监听容器根据方法是否正常返回来发送 ACK/NACK不等同于 RabbitMQ 的autoAcktrue后者在 Spring AMQP 中对应AcknowledgeMode.NONE。执行链路如下监听方法抛异常 ↓ Spring 在当前消费线程内重试 ↓ 超过最大次数 ↓ RejectAndDontRequeueRecoverer ↓ requeuefalse RabbitMQ 将消息投递到 DLX ↓ DLQ 保存失败消息1. 配置消费者重试Spring Boot 4.x 使用max-retriesspring:rabbitmq:listener:simple:acknowledge-mode:autoprefetch:20concurrency:3max-concurrency:10default-requeue-rejected:falseretry:enabled:truemax-retries:3initial-interval:1smultiplier:2max-interval:10s这个例子的执行过程大致是首次消费失败 ↓ 等待1秒 第1次重试失败 ↓ 等待2秒 第2次重试失败 ↓ 等待4秒 第3次重试失败 ↓ 进入恢复逻辑最终投递 DLQ如果是 Spring Boot 2.x/3.x通常使用旧属性retry:enabled:truemax-attempts:4initial-interval:1smultiplier:2max-interval:10smax-attempts: 4包含第一次执行也就是“首次执行 3 次重试”。当前 Spring Boot 配置已经使用max-retries。参见 Spring Boot RabbitMQ 配置。注意不要配置错位置# 消费者监听重试 spring.rabbitmq.listener.simple.retry # 生产者发送重试 spring.rabbitmq.template.retry2. 声明主队列、DLX 和 DLQimportorg.springframework.amqp.core.Binding;importorg.springframework.amqp.core.BindingBuilder;importorg.springframework.amqp.core.DirectExchange;importorg.springframework.amqp.core.Queue;importorg.springframework.amqp.core.QueueBuilder;importorg.springframework.amqp.rabbit.retry.MessageRecoverer;importorg.springframework.amqp.rabbit.retry.RejectAndDontRequeueRecoverer;importorg.springframework.beans.factory.annotation.Qualifier;importorg.springframework.context.annotation.Bean;importorg.springframework.context.annotation.Configuration;/** * 功能: * p * 声明订单消息的主队列和死信拓扑。业务消息正常进入主队列 * 消费重试耗尽后消息被拒绝且不重新入队再由RabbitMQ转发到死信队列。 * /p * p * 该配置只负责消息基础设施不处理订单业务逻辑。 * /p */Configuration(proxyBeanMethodsfalse)publicclassOrderRabbitTopologyConfig{publicstaticfinalStringORDER_EXCHANGEorder.exchange;publicstaticfinalStringORDER_ROUTING_KEYorder.created;publicstaticfinalStringORDER_QUEUEorder.created.queue;publicstaticfinalStringORDER_DEAD_EXCHANGEorder.dead.exchange;publicstaticfinalStringORDER_DEAD_ROUTING_KEYorder.created.dead;publicstaticfinalStringORDER_DEAD_QUEUEorder.created.dlq;BeanDirectExchangeorderExchange(){returnnewDirectExchange(ORDER_EXCHANGE,true,false);}BeanQueueorderQueue(){returnQueueBuilder.durable(ORDER_QUEUE).deadLetterExchange(ORDER_DEAD_EXCHANGE).deadLetterRoutingKey(ORDER_DEAD_ROUTING_KEY).build();}BeanBindingorderBinding(Qualifier(orderQueue)Queuequeue,Qualifier(orderExchange)DirectExchangeexchange){returnBindingBuilder.bind(queue).to(exchange).with(ORDER_ROUTING_KEY);}BeanDirectExchangeorderDeadExchange(){returnnewDirectExchange(ORDER_DEAD_EXCHANGE,true,false);}BeanQueueorderDeadQueue(){returnQueueBuilder.durable(ORDER_DEAD_QUEUE).build();}BeanBindingorderDeadBinding(Qualifier(orderDeadQueue)Queuequeue,Qualifier(orderDeadExchange)DirectExchangeexchange){returnBindingBuilder.bind(queue).to(exchange).with(ORDER_DEAD_ROUTING_KEY);}/** * 功能: * p * 当监听器的有限重试全部失败后要求容器拒绝消息且不重新进入原队列。 * 源队列配置DLX后RabbitMQ会将该消息转发到订单死信队列。 * /p * * return 消费重试耗尽后的恢复策略 */BeanMessageRecovererorderMessageRecoverer(){returnnewRejectAndDontRequeueRecoverer(订单消息重试耗尽转入死信队列);}}Spring Boot 在开启监听器重试后默认也会使用RejectAndDontRequeueRecoverer这里显式声明是为了让失败语义更清楚。重试耗尽后消息会被拒绝配置了 DLX 就进入死信队列否则会被丢弃。参见 Spring Boot AMQP 文档。3. 编写消费者importorg.springframework.amqp.rabbit.annotation.RabbitListener;importorg.springframework.stereotype.Component;/** * 功能: * p * 消费订单创建消息并把具体业务处理交给订单消息服务。 * 监听方法正常返回时由Spring自动ACK处理失败时必须继续抛出异常 * 由监听器重试和死信机制统一处理。 * /p */ComponentpublicclassOrderCreatedMessageListener{privatefinalOrderMessageServiceorderMessageService;publicOrderCreatedMessageListener(OrderMessageServiceorderMessageService){this.orderMessageServiceorderMessageService;}/** * 功能: * p * 消费订单创建事件。业务服务执行完成后方法正常返回Spring发送ACK * 业务服务抛出异常时本方法不捕获交给Spring执行有限重试。 * /p * * param message 订单创建消息messageId用于消费幂等 */RabbitListener(queuesOrderRabbitTopologyConfig.ORDER_QUEUE)publicvoidconsume(OrderCreatedMessagemessage){orderMessageService.consume(message);}}这里最重要的是不要捕获异常后只打印日志。错误写法RabbitListener(queuesorder.created.queue)publicvoidconsume(OrderCreatedMessagemessage){try{orderMessageService.consume(message);}catch(Exceptionexception){log.error(订单消息处理失败,exception);}}异常被吞掉后监听方法正常返回Spring 认为处理成功 ↓ 发送 ACK ↓ 不会重试也不会进入 DLQ如果确实需要记录日志必须继续抛出RabbitListener(queuesorder.created.queue)publicvoidconsume(OrderCreatedMessagemessage){try{orderMessageService.consume(message);}catch(Exceptionexception){log.error(订单消息处理失败messageId{},message.getMessageId(),exception);throwexception;}}4. 业务服务必须实现幂等importorg.springframework.stereotype.Service;importorg.springframework.transaction.annotation.Transactional;/** * 功能: * p * 在数据库事务内完成订单消息幂等校验和业务处理。 * 消息重试或ACK丢失时可能发生重复投递因此通过messageId唯一记录 * 确保相同消息不会重复修改订单。 * /p */ServicepublicclassOrderMessageService{privatefinalConsumeLogRepositoryconsumeLogRepository;privatefinalOrderRepositoryorderRepository;publicOrderMessageService(ConsumeLogRepositoryconsumeLogRepository,OrderRepositoryorderRepository){this.consumeLogRepositoryconsumeLogRepository;this.orderRepositoryorderRepository;}/** * 功能: * p * 消费订单事件。首次消费时写入幂等记录并更新订单 * 重复消息直接返回使监听器可以安全ACK。 * /p * * param message 待处理的订单创建消息 */Transactional(rollbackForException.class)publicvoidconsume(OrderCreatedMessagemessage){booleanisFirstConsumptionconsumeLogRepository.tryInsert(message.getMessageId());if(!isFirstConsumption){return;}orderRepository.createOrder(message.getOrderId(),message.getUserId());}}message_id必须有数据库唯一索引CREATEUNIQUEINDEXuk_mq_consume_log_message_idONmq_consume_log(message_id);5. 消息什么时候进入 DLQ配置成功后监听器正常返回 → Spring ACK → 消息删除 监听器抛异常但重试成功 → Spring ACK → 消息删除 监听器一直失败 → 重试耗尽 → basic.reject / basic.nackrequeuefalse → order.dead.exchange → order.created.dlqRabbitMQ 会在死信消息的 Header 中增加x-death记录原队列、死信原因和次数。参见 RabbitMQ DLX 文档。注意这里的 Spring 重试是消费者进程内重试重试期间消息保持 Unacked重试会占用消费者线程每次本地重试不会增加x-death只有最终被 RabbitMQ 死信转发时才产生x-death记录。所以这种方案适合秒级、次数较少的重试。分钟级、小时级重试应改成主队列 ↓ 失败 短延时重试队列 ↓ 再失败 长延时重试队列 ↓ 超过上限 最终 DLQ6. 一个容易遇到的部署问题如果order.created.queue已经存在并且以前没有配置 DLX再用上面的代码声明可能报PRECONDITION_FAILED - inequivalent arg x-dead-letter-exchange因为 RabbitMQ 不允许直接修改已有队列的声明参数。处理方式测试环境删除旧队列后重新声明生产环境优先通过 RabbitMQ Policy 设置 DLX不要直接删除仍有消息的生产队列。RabbitMQ 官方也更推荐使用 Policy 配置 DLX因为 Policy 可以动态调整而硬编码的x-arguments通常需要重新创建队列。参见 RabbitMQ DLX Policy。7. 验收测试至少验证以下场景/** * 业务处理成功时验证监听器只执行一次且消息不会进入DLQ。 */TestvoidshouldAcknowledgeMessageWhenBusinessSucceeds(){}/** * 业务持续失败时验证达到重试上限后消息进入订单DLQ。 */TestvoidshouldMoveMessageToDeadQueueAfterRetriesExhausted(){}/** * 同一个messageId被重复投递时验证订单只创建一次。 */TestvoidshouldKeepBusinessIdempotentWhenMessageIsRedelivered(){}集成测试可以从 DLQ 读取结果MessagedeadMessagerabbitTemplate.receive(OrderRabbitTopologyConfig.ORDER_DEAD_QUEUE,15_000);assertThat(deadMessage).isNotNull();assertThat(deadMessage.getMessageProperties().getHeaders()).containsKey(x-death);8. 运维与重放建议为 DLQ 的消息数量、最老消息滞留时间和持续增长趋势配置监控告警。重放前先定位失败原因并修复消费者避免消息重新进入“主队列 → 重试 → DLQ”的循环。重放工具应保留原始messageId继续复用消费端幂等校验同时记录操作人、重放时间、批次和结果。DLQ 是失败消息的隔离区不等同于自动补偿机制是否自动重放应根据异常类型、业务风险和重试间隔单独设计。参考资料Spring Boot RabbitMQ 配置属性Spring Boot AMQP 文档Spring AMQP 监听容器配置RabbitMQ Dead Letter ExchangesRabbitMQ Reliability Guide