Rabbitmq集成springboot 使用死信队列

一、何为死信队列

RabbitMQ的死信队列（Dead Letter Queue，DLQ）是一种特殊的队列机制，用于处理那些无法被正常消费的消息。这些消息可能由于各种原因无法被消费者正确处理，如果不加以处理，可能会导致队列堵塞，影响系统的正常运行。

1、死信队列的作用

隔离问题消息：将无法处理的消息转移到专门的死信队列中，避免影响其他正常消息的消费。
故障排查：通过分析死信队列中的消息，可以快速定位和解决消息消费失败的原因。
提高系统稳定性：防止因个别消息处理失败而导致整个队列阻塞，从而提升系统的健壮性和可靠性。

2、死信产生的原因

在 RabbitMQ 中，消息变成死信的情况主要有以下几种：

消息被拒绝（basic.reject 或 basic.nack）并且 requeue=false
如果消费者明确拒绝了某条消息，并且不希望重新入队，则该消息会被发送到死信队列(前提是配置了死信队列)。
消息过期（TTL 过期）
如果消息设置了生存时间（Time To Live, TTL），并且在队列中等待的时间超过了这个限制，消息也会成为死信。
队列达到最大长度限制
当队列已经满了（即达到了预设的最大长度），新进入的消息会根据策略被丢弃或放入死信队列。

3、死信队列的核心组件

死信交换机（Dead Letter Exchange, DLX）每个普通队列可以通过配置指定一个死信交换机。当消息变成死信时，RabbitMQ 会自动将该消息发布到对应的 DLX。
绑定键（Routing Key）可以为 DLX 指定一个绑定键，死信消息将会使用这个绑定键来路由到相应的死信队列。
死信队列（DLQ）实际上是一个普通的队列，只是它接收的是来自 DLX 的死信消息

4、私信队列消息示意图

+---------------------+
|                     |
|   生产者 Producer   |
|                     |
+----------+----------+|| 发送消息到业务交换机v
+-------------------------+
|                         |
| 业务交换机 BusinessExchange |
|                         |
+------------+------------+|| 根据路由键 routingKeyv
+----------------------------+
|                            |
| 业务队列 BusinessQueue     |
| (配置了 DLX 和 DLK)        |
|                            |
+-------------+--------------+|+---------+------------------+|         |                  |
消息被正常消费            消息达到最大重试次数
(channel.basicAck)       或被拒绝 requeue=false|         |                  |v         v                  v
+------------------+    +--------------------+
|                  |    |                    |
| 正常消费者       |    | 死信交换机 DLX     |
|                  |    |                    |
+------------------+    +---------+----------+|+-------v--------+|                || 死信队列 DLQ   ||                |+--------+-------+|+------v-------+|              || 死信消费者   ||              |+--------------+

二、准备基本环境

1、pom.xml引入的java包

<dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId><version>${springboot-version}</version></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-amqp</artifactId><version>${springboot-version}</version></dependency><dependency><groupId>org.projectlombok</groupId><artifactId>lombok</artifactId><version>1.18.24</version><scope>provided</scope></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-test</artifactId><version>${springboot-version}</version><scope>test</scope></dependency><dependency><groupId>com.alibaba</groupId><artifactId>fastjson</artifactId><version>2.0.57</version></dependency></dependencies>

2、yaml配置文件

# 8004是zookeeper服务器的支付服务提供者端口号
server:port: 8004
spring:application:name: cloud-mqrabbitmq:addresses: 192.168.96.133port: 5672username: guestpassword: guestvirtual-host: /#消费者配置listener:#todo 切记，设置了重拾机制,要抛出异常，不可try catch 后不抛异常,否则重试机制失效simple:#开启ack 手动确认消息是否被消费成功acknowledge-mode: manualretry:enabled: true# 消费失败后，继续消费，然后最多消费5次就不再消费。max-attempts: 5# 消费失败后 ，重试初始间隔时间 2秒initial-interval: 2000# 重试最大间隔时间5秒max-interval: 5000# 间隔时间乘子，间隔时间*乘子=下一次的间隔时间，最大不能超过设置的最大间隔时间multiplier: 2direct:#开启ack 手动确认消息是否被消费成功acknowledge-mode: manual#todo 切记，设置了重拾机制,要抛出异常，不可try catch 后不抛异常,否则重试机制失效retry:enabled: true# 消费失败后，继续消费，然后最多消费3次就不再消费。max-attempts: 3# 消费失败后 ，重试初始间隔时间 3秒initial-interval: 3000# 重试最大间隔时间max-interval: 7000# 间隔时间乘子，间隔时间*乘子=下一次的间隔时间，最大不能超过设置的最大间隔时间multiplier: 2# 生产者配置template:retry:# 开启消息发送失败重试机制enabled: true# 生产者 true-开启消息抵达队列的确认publisher-returns: false#simple 配置用于设置 RabbitMQ 消息生产者的消息确认类型为“简单确认”。这意味着当消息被发送到 RabbitMQ 之后，只有在消息成功投递到队列中后，RabbitMQ 才会向生产者发送一个确认（ack）通知。如果消息未能成功投递，则不会收到确认。#该配置通常与 publisher-returns: true 一起使用以启用消息返回机制，但在此配置中 publisher-returns 被设置为 false，表示不启用消息返回功能publisher-confirm-type: simple

3、主启动类


import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;/*** @author 10564*/
@SpringBootApplication
public class ApplicationRabbitmq {public static void main(String[] args) {SpringApplication.run(ApplicationRabbitmq.class, args);}
}

三、手动确认消息示例

1、定义消息队列Queue名称

package org.xwb.springcloud.constant;
/*** @author Administrator*/
public class MqDeadLetterConstant {/*** 手动确认 正常消息队列名称*/public static final String BUSINESS_QUEUE = "businessQueue";/*** 手动确认 正常交换机名称*/public static final String BUSINESS_EXCHANGE = "businessExchange";/*** 手动确认 路由key名称*/public static final String BUSINESS_ROUTING_KEY = "businessRoutingKey";/*** 死信队列名称DeadLetter*/public static final String DEAD_LETTER_QUEUE = "deadLetterQueue";/*** 死信交换机名称*/public static final String DEAD_LETTER_EXCHANGE = "deadLetterExchange";/*** 死信路由key名称*/public static final String DEAD_LETTER_ROUTING_KEY = "deadLetterRoutingKey";
}

2、配置类Configuration


import org.springframework.amqp.core.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.xwb.springcloud.constant.MqDeadLetterConstant;/*** @author Administrator*/
@Configuration
public class RabbitmqDeadLetterConfig {/*** 死信队列*/@Beanpublic Queue deadLetterQueue() {return new Queue(MqDeadLetterConstant.DEAD_LETTER_QUEUE, true);}/*** 死信交换机*/@Beanpublic DirectExchange deadLetterExchange() {return new DirectExchange(MqDeadLetterConstant.DEAD_LETTER_EXCHANGE);}// 将死信队列绑定到死信交换机，并设置路由键与死信队列的路由键一致@Beanpublic Binding bindingDeadLetter() {// 绑定死信队列到死信交换机和路由键一致return BindingBuilder.bind(deadLetterQueue()).to(deadLetterExchange()).with(MqDeadLetterConstant.DEAD_LETTER_ROUTING_KEY);}//todo =======================业务队列、交换机配置========================/*** 定义业务队列* todo  配置业务队列，并设置死信交换机、死信队列、TTL、路由键等信息*/@Beanpublic Queue businessQueue() {//业务的队列return QueueBuilder.durable(MqDeadLetterConstant.BUSINESS_QUEUE)//设置死信交换机（DLX），当消息无法被业务消费时（例如过期、拒绝等），消息会被转发到指定的死信交换机.deadLetterExchange(MqDeadLetterConstant.DEAD_LETTER_EXCHANGE)//设置死信队列的路由键，用于将消息正确地路由到死信队列中.deadLetterRoutingKey(MqDeadLetterConstant.DEAD_LETTER_ROUTING_KEY)// 可选：设置消息过期时间.ttl(10000).build();}/*** 业务队列 交换机*/@Beanpublic DirectExchange ackBusinessExchange() {return new DirectExchange(MqDeadLetterConstant.BUSINESS_EXCHANGE);}/*** 将业务队列绑定到业务交换机上，指定交换机的路由*/@Beanpublic Binding bindingBusiness() {return BindingBuilder.bind(businessQueue()).to(ackBusinessExchange()).with(MqDeadLetterConstant.BUSINESS_ROUTING_KEY);}}

3、生产者Producer

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.stereotype.Component;
import org.xwb.springcloud.constant.MqDeadLetterConstant;import javax.annotation.Resource;/*** @author Administrator*/
@Component
public class DeadLetterProducer {private static final Logger log = LoggerFactory.getLogger(DeadLetterProducer.class);@Resourceprivate RabbitTemplate rabbitTemplate;public void senderDeadLetterMessage(String message) {log.info("\n生产者DeadLetter发送消息:【{}】\n", message);//参数1：交换机名称//参数2：路由key//参数3：消息//topic_exchange交换机  需要指定路由key  绑定到该交换机且符合路由key的队列都会收到消息rabbitTemplate.convertAndSend(MqDeadLetterConstant.BUSINESS_EXCHANGE, MqDeadLetterConstant.BUSINESS_ROUTING_KEY, message);}
}

4、消费者Consumer

1、直接进入死信队列测试结果


import com.alibaba.fastjson2.util.DateUtils;
import com.rabbitmq.client.Channel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Component;
import org.xwb.springcloud.constant.MqDeadLetterConstant;import java.io.IOException;
import java.util.Date;/*** @author 10564*/
@Component
public class DeadLetterBusinessAndConsumer {private static final Logger log = LoggerFactory.getLogger(DeadLetterBusinessAndConsumer.class);/*** @param msg     消息内容* @param channel 通道对象* @param tag     消息的tag*/@RabbitListener(queues = MqDeadLetterConstant.BUSINESS_QUEUE)public void receiveAckBusinessQueueMessage(String msg, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long tag) throws Exception {try {log.info("\n消费者eadLetter接收消息:【{}】\n", msg);if ("normal".equals(msg)) {log.info("\n已被正常消费 normal 【{}】,【{}】\n", msg, DateUtils.format(new Date()));//todo 消息正常处理，从队列中删除掉当前消息channel.basicAck(tag, false);} else if ("deadLetter".equals(msg)) {log.info("\n模拟消费异常,需要进入死信队列中 deadLetter 【{}】,【{}】\n", msg, DateUtils.format(new Date()));//channel.basicReject(tag, false);也可以使用channel.basicNack(tag, false,false);} else {throw new Exception(msg);}} catch (Exception e) {log.info("\n消费者处理消息异常:【{}】\n", e.getMessage());//todo true 允许加入队列， false 拒绝加入队列 ，进入私信队列中channel.basicReject(tag, false);}}/*** 死信队列消费者** @param msg     消息* @param channel 通道对象* @param tag     消息的tag*/@RabbitListener(queues = MqDeadLetterConstant.DEAD_LETTER_QUEUE)public void receiveAckDlqQueue(String msg, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long tag) {try {// 处理死信队列消息log.error("\n死信消费者接收消息:【{}】\n", msg);//todo true 允许加入队列， false 拒绝加入队列channel.basicAck(tag, false);} catch (Exception e) {log.info("\n死信消费者消息异常:【{}】\n", e.getMessage());try {channel.basicNack(tag, false, true);} catch (IOException ex) {throw new RuntimeException(ex);}}}
}

用到的basicNack,basicAck,basicReject的具体参数意义参考手动确认消息basicAck、basicNack、basicReject的使用

2、触发重试机制最终进入私信队列测试结果

package org.xwb.springcloud.messagetype.deadletter;import com.alibaba.fastjson2.util.DateUtils;
import com.rabbitmq.client.Channel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Component;
import org.xwb.springcloud.constant.MqDeadLetterConstant;import java.io.IOException;
import java.util.Date;
import java.util.Map;/*** @author 10564*/
@Component
public class DeadLetterBusinessAndRetryCountConsumer {private static final Logger log = LoggerFactory.getLogger(DeadLetterBusinessAndRetryCountConsumer.class);Map<String, Integer> retryCountMap = new java.util.HashMap<>();@Value("${spring.rabbitmq.listener.simple.retry.max-attempts}")private Integer retryCount;/*** @param msg     消息内容* @param channel 通道对象* @param tag     消息的tag*/@RabbitListener(queues = MqDeadLetterConstant.BUSINESS_QUEUE)public void receiveAckBusinessQueueMessage(String msg, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long tag) throws Exception {try {log.info("\n消费者eadLetter接收消息:【{}】\n", msg);if ("normal".equals(msg)) {log.info("\n已被正常消费 normal 【{}】,【{}】\n", msg, DateUtils.format(new Date()));//todo 消息正常处理，从队列中删除掉当前消息channel.basicAck(tag, false);}  else {//todo 此处模拟消息消费异常，计数器累加，超过5次则拒绝加入队列 正常情况下使用redis 实现计数器，此处为了演示直接使用map代替Integer timesObj = retryCountMap.get("msg:" + tag);int times = timesObj == null ? 0 : timesObj;if (times >= retryCount - 1) {channel.basicNack(tag, false, false);} else {times = times == 0 ? 1 : times + 1;retryCountMap.put("msg:" + tag, times);log.error("\n已被消费异常，开始重试【{}】,第【{}】次\n", msg, times);throw new Exception(msg);}}} catch (Exception e) {//抛出异常，触发重试机制throw e;}}/*** 死信队列消费者** @param msg     消息* @param channel 通道对象* @param tag     消息的tag*/@RabbitListener(queues = MqDeadLetterConstant.DEAD_LETTER_QUEUE)public void receiveAckDlqQueue(String msg, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long tag) {try {// 处理死信队列消息log.error("\n死信消费者接收消息:【{}】\n", msg);//todo true 允许加入队列， false 拒绝加入队列channel.basicAck(tag, false);} catch (Exception e) {log.error("\n死信消费者消息异常:【{}】\n", e.getMessage());try {channel.basicNack(tag, false, true);} catch (IOException ex) {throw new RuntimeException(ex);}}}
}

5、测试Test

package org.xwb.springcloud.controller;import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import org.xwb.springcloud.messagetype.deadletter.DeadLetterProducer;
import javax.annotation.Resource;/*** @author 10564*/
@RestController
@RequestMapping("/mq")
public class MqMessageController {@Resourceprivate DeadLetterProducer deadLetterProducer;@GetMapping("/deadLetter")public void deadLetter(String message) {deadLetterProducer.senderDeadLetterMessage(message);}
}

6、测试结果

1、直接进入私信队列


### deadLetter
GET http://localhost:8004/mq/deadLetter?message=normal
###结果
2025-06-22 09:04:47.926  INFO 11116 --- [nio-8004-exec-5] o.x.s.m.deadletter.DeadLetterProducer    : 
生产者DeadLetter发送消息:【normal】2025-06-22 09:04:47.929  INFO 11116 --- [ntContainer#2-1] o.x.s.m.d.DeadLetterBusinessAndConsumer  : 
d消费者eadLetter接收消息:【normal】2025-06-22 09:04:47.929  INFO 11116 --- [ntContainer#2-1] o.x.s.m.d.DeadLetterBusinessAndConsumer  : 
已被正常消费 normal 【normal】,【2025-06-22 09:04:47】### deadLetter
GET http://localhost:8004/mq/deadLetter?message=deadLetter
###结果
2025-06-22 09:05:11.311  INFO 11116 --- [nio-8004-exec-6] o.x.s.m.deadletter.DeadLetterProducer    : 
生产者DeadLetter发送消息:【deadLetter】2025-06-22 09:05:11.315  INFO 11116 --- [ntContainer#2-1] o.x.s.m.d.DeadLetterBusinessAndConsumer  : 
d消费者eadLetter接收消息:【deadLetter】2025-06-22 09:05:11.315  INFO 11116 --- [ntContainer#2-1] o.x.s.m.d.DeadLetterBusinessAndConsumer  : 
模拟消费异常,需要进入死信队列中 deadLetter 【deadLetter】,【2025-06-22 09:05:11】2025-06-22 09:05:11.319 ERROR 11116 --- [ntContainer#1-1] o.x.s.m.d.DeadLetterBusinessAndConsumer  : 
死信消费者接收消息:【deadLetter】### deadLetter
GET http://localhost:8004/mq/deadLetter?message=deadLetter1
###结果
2025-06-22 09:05:29.494  INFO 11116 --- [nio-8004-exec-7] o.x.s.m.deadletter.DeadLetterProducer    : 
生产者DeadLetter发送消息:【deadLetter1】2025-06-22 09:05:29.497  INFO 11116 --- [ntContainer#2-1] o.x.s.m.d.DeadLetterBusinessAndConsumer  : 
d消费者eadLetter接收消息:【deadLetter1】2025-06-22 09:05:29.497  INFO 11116 --- [ntContainer#2-1] o.x.s.m.d.DeadLetterBusinessAndConsumer  : 
消费者处理消息异常:【deadLetter1】2025-06-22 09:05:29.499 ERROR 11116 --- [ntContainer#1-1] o.x.s.m.d.DeadLetterBusinessAndConsumer  : 
死信消费者接收消息:【deadLetter1】

2、触发重试机制最终进入私信队列-测试结果

### deadLetter
GET http://localhost:8004/mq/deadLetter?message=deadLetter1
###结果
2025-06-22 11:10:06.909  INFO 2280 --- [nio-8004-exec-1] o.x.s.m.deadletter.DeadLetterProducer    : 
生产者DeadLetter发送消息:【deadLetter1】2025-06-22 11:10:06.918  INFO 2280 --- [ntContainer#2-1] o.x.s.m.d.DeadLetterBusinessAndConsumer  : 
消费者eadLetter接收消息:【deadLetter1】2025-06-22 11:10:06.918 ERROR 2280 --- [ntContainer#2-1] o.x.s.m.d.DeadLetterBusinessAndConsumer  : 
已被消费异常，开始重试【deadLetter1】,第【1】次2025-06-22 11:10:08.921  INFO 2280 --- [ntContainer#2-1] o.x.s.m.d.DeadLetterBusinessAndConsumer  : 
消费者eadLetter接收消息:【deadLetter1】2025-06-22 11:10:08.921 ERROR 2280 --- [ntContainer#2-1] o.x.s.m.d.DeadLetterBusinessAndConsumer  : 
已被消费异常，开始重试【deadLetter1】,第【2】次2025-06-22 11:10:12.928  INFO 2280 --- [ntContainer#2-1] o.x.s.m.d.DeadLetterBusinessAndConsumer  : 
消费者eadLetter接收消息:【deadLetter1】2025-06-22 11:10:12.929 ERROR 2280 --- [ntContainer#2-1] o.x.s.m.d.DeadLetterBusinessAndConsumer  : 
已被消费异常，开始重试【deadLetter1】,第【3】次2025-06-22 11:10:17.935  INFO 2280 --- [ntContainer#2-1] o.x.s.m.d.DeadLetterBusinessAndConsumer  : 
消费者eadLetter接收消息:【deadLetter1】2025-06-22 11:10:17.935 ERROR 2280 --- [ntContainer#2-1] o.x.s.m.d.DeadLetterBusinessAndConsumer  : 
已被消费异常，开始重试【deadLetter1】,第【4】次2025-06-22 11:10:22.948  INFO 2280 --- [ntContainer#2-1] o.x.s.m.d.DeadLetterBusinessAndConsumer  : 
消费者eadLetter接收消息:【deadLetter1】2025-06-22 11:10:22.960 ERROR 2280 --- [ntContainer#1-1] o.x.s.m.d.DeadLetterBusinessAndConsumer  : 
死信消费者接收消息:【deadLetter1】