1. 简介
在前面的高级特性中，我们介绍了重试机制和 TTL，那么产生下列问题：
- 在重试机制中，当消费者消费消息发生异常时，会触发消息重发机制，由于我们配置了最大的重发次数，那么当超过这个次数后，若消息依然没有被成功消费，就需要将消息进行保存，等待下一次消费，那么，这条消息应该保存到哪里去呢？
- 在 TTL 机制中，我们为队列和消息设置了过期时间，当超过这个时间后消息就会被删除，但是这条消息是需要被消费的，于是就需要将过期的消息保存下来，等待下次消费。但是，消息应该保存到哪里去呢？
在 RabbitMQ 中，将类似于结果的消息称为死信，那么就涉及到，变成死信的消息应该存储到哪里去？
2. 会产生死信的场景
- 消息重发后，次数到达指定重发次数依然未被消费，就会成为死信
- 消息到达过期时间依然没有被消费，就会成为死信
- 队列已经满了，却依然由消息入队列，就会产生溢出，溢出的这部分消息就会成为死信
3. 死信队列
在 RabbitMQ 中，可以声明一个队列，这个队列专门用来存放死信，于是就成为死信队列。
死信队列的工作流程如下：
- 首先需要声明一个死信交换机，与普通队列进行绑定；
- 其次需要声明一个死信队列，与死信交换机进行绑定；
- 当普通队列中的消息成为死信后，就会被发送给死信交换机，然后由死信交换机分配给与之绑定的死信队列；
- 存储在死信队列中的死信会等待被别的消费者再次消费。
4. 配置死信交换机与死信队列
声明一个正常交换机，正常队列，死信交换机，死信队列，并将正常交换机与正常队列进行绑定，将正常队列与死信交换机进行绑定，将死信交换机与死信队列进行绑定，代码如下：
@Configuration
public class DLConfig {
/**
* 正常
* @return
*/
@Bean("norQueue")
public Queue norQueue() {
return QueueBuilder.durable(Constants.NOR_QUEUE)
.ttl(10000) //过期时间 10s
.deadLetterExchange(Constants.DL_EXCHANGE) //绑定死信交换机
.deadLetterRoutingKey(Constants.DL_ROUTINGKEY)
.maxLength(10L) //队列长度为 10
.build();
}
@Bean("norExchange")
public DirectExchange norExchange() {
return ExchangeBuilder.directExchange(Constants.NOR_EXCHANGE).build();
}
@Bean("norBind")
public Binding norBind(@Qualifier("norExchange") DirectExchange directExchange,
@Qualifier("norQueue") Queue queue) {
return BindingBuilder.bind(queue).to(directExchange).with(Constants.NOR_ROUTINGKEY);
}
/**
* 死信
*/
@Bean("dlQueue")
public Queue dlQueue() {
return QueueBuilder.durable(Constants.DL_QUEUE).build();
}
@Bean("dlExchange")
public DirectExchange dlExchange() {
return ExchangeBuilder.directExchange(Constants.DL_EXCHANGE).build();
}
@Bean("dlBind")
public Binding dlBind(@Qualifier("dlExchange") DirectExchange directExchange,
@Qualifier("dlQueue") Queue queue) {
return BindingBuilder.bind(queue).to(directExchange).with(Constants.DL_ROUTINGKEY);
}
}在上面的代码中，我们指定了下面的条件：
- 队列的过期时间为 10s
- 队列长度为 10
重试机制的配置如下：
spring:
rabbitmq:
listener:
simple:
acknowledge-mode: auto
retry:
enabled: true # 开启消费者失败重试
initial-interval: 5000ms # 初始失败等待时⻓为5秒
max-attempts: 5 # 最⼤重试次数(包括自身消费的⼀次)在上面的配置中，设置的最大的重发次数为 5 次。
5. 验证消息重试超过规定次数是否进入死信队列
生产者代码如下：
@RequestMapping("/dl")
public String dl() {
String messageInfo = "dl... ";
rabbitTemplate.convertAndSend(Constants.NOR_EXCHANGE, Constants.NOR_ROUTINGKEY, messageInfo);
return "消息发送成功";
}消费者代码如下：
@Component
@Slf4j
public class DLListener {
@RabbitListener(queues = Constants.NOR_QUEUE)
public void listener(Message message) throws IOException {
String messageInfo = new String(message.getBody());
long deliveryTag = message.getMessageProperties().getDeliveryTag();
log.info("接收消息, message: {}, deliveryTag: {}", messageInfo, deliveryTag);
int num = 1 / 0;
log.info("消息消费完成");
}
}代码运行结果如下：
消息重发五次后抛出异常，观察 RabbitMQ 客户端后，这条消息已经存放进了死信队列中：
6. 验证 TTL 与死信队列
生产者代码如下：
@RequestMapping("/dl")
public String dl() {
String messageInfo = "dl... ";
rabbitTemplate.convertAndSend(Constants.NOR_EXCHANGE, Constants.NOR_ROUTINGKEY, messageInfo);
return "消息发送成功";
}将正常队列的 TTL 设置为 10s，运行代码后，观察 RabbitMQ 客户端：
10s 前：
10s 后：
7. 验证队列溢出与死信队列
将正常队列的长度设置为 10，生产者向 RabbitMQ 发送 20 条数据，生产者代码如下：
@RequestMapping("/dl")
public String dl() {
for (int i = 0; i < 20; i++) {
String messageInfo = "dl... " + i;
rabbitTemplate.convertAndSend(Constants.NOR_EXCHANGE, Constants.NOR_ROUTINGKEY, messageInfo);
}
return "消息发送成功";
}运行代码，观察 RabbitMQ 客户端：
可以看到，溢出的 10 条消息进入了死信队列。