《RabbitMQ篇》死信队列和延时队列

死信队列

无法被正常消耗的消息就称为死信，存储这种消息的队列成为死信队列，便于开辟者分析和处理这些“死信”，以便找出问题地点，而不是直接丢弃这些消息。

• 消息 TTL 逾期
  

TTL是 Time To Live 的缩写, 也就是生存时间

• 队列到达最大长度
  

队列满了，无法再添加数据到 MQ 中

• 消息被拒绝
  

(basic.reject 或 basic.nack) 并且 requeue = false
实现方式

为正常的交换机配置一个死信交换机，当消息无法正常消耗时，正常交换机自动把消息转发给死信交换机。
验证TTL逾期

消耗者代码如下，启动之后马上停止运行（模拟消耗者的消耗消息失败）

1. import com.rabbitmq.client.BuiltinExchangeType;
2. import com.rabbitmq.client.Channel;
3. import com.rabbitmq.client.DeliverCallback;
5. import java.nio.charset.StandardCharsets;
6. import java.util.HashMap;
7. import java.util.Map;
9. public class Consumer01 {
11.     //普通交换机的名称
12.     public static final String NORMAL_EXCHANGE = "normal_exchange";
13.     //死信交换机的名称
14.     public static final String DEAD_EXCHANGE = "dead_exchange";
16.     //普通队列的名称
17.     public static final String NORMAL_QUEUE = "normal_queue";
18.     //死信队列的名称
19.     public static final String DEAD_QUEUE = "dead_queue";
21.     public static void main(String[] args) throws Exception {
22.         Channel channel = RabbitMQUtil.getChannel();
24.         // 声明死信和普通交换机，类型为direct
25.         channel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT);
26.         channel.exchangeDeclare(DEAD_EXCHANGE, BuiltinExchangeType.DIRECT);
28.         Map<String, Object> arguments = new HashMap<>();
29.         // 正常队列关联死信交换机，意味着正常队列消费异常的消息会转发到关联的这个死信交换机
30.         arguments.put("x-dead-letter-exchange", DEAD_EXCHANGE);
31.         arguments.put("x-dead-letter-routing-key", "routingKeyTwo");
32.         // 声明普通队列
33.         channel.queueDeclare(NORMAL_QUEUE, false, false, false, arguments);
35.         // 声明死信队列
36.         channel.queueDeclare(DEAD_QUEUE, false, false, false, null);
38.         // 绑定普通交换机及其队列
39.         channel.queueBind(NORMAL_QUEUE, NORMAL_EXCHANGE, "routingKeyOne");
41.         // 绑定死信交换机及其死信队列
42.         channel.queueBind(DEAD_QUEUE, DEAD_EXCHANGE, "routingKeyTwo");
43.         
44.         System.out.println("等待接收消息...");
45.         DeliverCallback deliverCallback = (consumerTag, message) -> {
46.             System.out.println("Consumer01接受的消息是：" + new String(message.getBody(), StandardCharsets.UTF_8));
47.         };
49.         channel.basicConsume(NORMAL_QUEUE, true, deliverCallback, consumerTag -> {}) ;
50.     }
51. }

复制代码

生产者代码如下，主要生产消息，并给消息设置TTL。

1. import com.rabbitmq.client.AMQP;
2. import com.rabbitmq.client.Channel;
4. public class DeadProducer {
5.     // 普通交换机的名称
6.     public static final String NORMAL_EXCHANGE = "normal_exchange";
8.     public static void main(String[] args) throws Exception {
9.         Channel channel = RabbitMQUtil.getChannel();
10.         // 死信消息 设置TTL(live to time)单位是ms
11.         AMQP.BasicProperties properties = new AMQP.BasicProperties().builder().expiration("10000").build();
12.         for (int i = 1; i <= 10; i++) {
13.             String message = "message" + i;
14.             channel.basicPublish(NORMAL_EXCHANGE, "routingKeyOne", properties, message.getBytes());
15.         }
16.     }
17. }

复制代码

测试结果如下

• 启动消耗者停止后，会创建出对应的交换机和队列，此时正常队列中没有消息。
  

• 启动生产者后，可以看到正常队列中有消息，在10s后，消息陆续转发到死信队列中
  

• 可以看到最终消息来到了死信队列，因为没有消耗者处理死信队列，全部消息会不停堆积在这。
  

验证队列到达最大长度

消耗者代码中，对普通队列参数加上以下代码，设置普通队列的最大长度。（启动之后关闭该消耗者，模拟其接收不到消息）

1. // 设置正常队列长度的限制，例如发送10个消息，6个为正常，4个为死信
2. arguments.put("x-max-length",6);

复制代码

把前面生产者代码中，设置消息TTL这段代码注释掉。（启动生产者）

1. // AMQP.BasicProperties properties = new AMQP.BasicProperties().builder().expiration("10000").build();

复制代码

测试结果如下
因为正常队列设置了最大长度为6，全部同时收到10个消息后，会把多余的4个当做死信转发到死信队列。

图：正常队列中消息数

图：死信队列中消息数

---

验证消息被拒

在前面代码的基础上，把消耗者代码中，普通队列长度设置的代码注释

1. // arguments.put("x-max-length", 6);

复制代码

同时关闭自动确认，使用手动应答。启动消耗者（这里不能马上停止），再启动生产者。

1. DeliverCallback deliverCallback = (consumerTag, message) -> {
2.     String msgContent = new String(message.getBody(), StandardCharsets.UTF_8);
3.     if (msgContent.equals("message5")) {
4.         System.out.println("Consumer01接受的消息是：" + msgContent + "： 此消息是被C1拒绝的");
5.         // requeue 设置为 false 代表拒绝重新入队 该队列如果配置了死信交换机将发送到死信队列中
6.         channel.basicReject(message.getEnvelope().getDeliveryTag(), false);
7.     } else {
8.         System.out.println("Consumer01接受的消息是：" + msgContent);
9.         channel.basicAck(message.getEnvelope().getDeliveryTag(), false);
10.     }
11. };
13. channel.basicConsume(NORMAL_QUEUE, false, deliverCallback, consumerTag -> {});

复制代码

测试结果如下，被拒绝的且不能重新入队的消息称为死信转发到死信队列。

延时队列

延时队列就是用来存放需要在指定时间被处理的元素的队列。好比：订单在非常钟之内未付出则自动取消。
队列和消息可以分别设置TTL

两者同时设置，取最小值。

• 设置队列的TTL属性，那么一旦消息逾期，就会被队列丢弃(如果配置了死信队列被丢到死信队列中)。
  
• 设置消息的TTL属性，消息纵然逾期，也不一定会被马上丢弃，因为消息是否逾期是在即将投递到消耗者之前判断的。在这种情况下需要注意，如果消息A的TTL大于消息B的TTL，他们先后放入到一个正常队列中，虽然消息B的TTL要小，但不会先放入到死信队列处理。
  

实现方式

给消息设置TTL，消息逾期后会转发到死信交换机，为死信队列配置消耗者处理这些消息，就到达延时队列的目的。
死信队列和延时队列区别

死信队列主要存储无法正常消耗的消息，便于处理分析。延时队列利用消息TTL产存亡信这一特点来控制消息的消耗时间。