概念

先从概念解释上搞清楚这个定义，死信，顾名思义就是无法被消费的消息，字面意思可以这样理解，一般来说，producer 将消息投递到 broker 或者直接到queue 里了，consumer 从 queue 取出消息进行消费，但某些时候由于特定的原因导致 queue 中的某些消息无法被消费，这样的消息如果没有后续的处理，就变成了死信，有死信自然就有了死信队列。
应用场景: 为了保证订单业务的消息数据不丢失，需要使用到 RabbitMQ 的死信队列机制，当消息消费发生异常时，将消息投入死信队列中.还有比如说: 用户在商城下单成功并点击去支付后在指定时间未支付时自动失效

死信的来源

1.消息 TTL 过期
2.队列达到最大长度(队列满了，无法再添加数据到 mq 中)
3.消息被拒绝(basic.reject 或 basic.nack)并且 requeue=false(不放回队列中).

死信实战

代码架构图

07 死信队列 - 图1

消息TTL 过期

消费者 C1 代码

  1. public class Consumer01 {
  3.     public static void main(String[] argv) throws Exception {
  4.         // 普通交换机名称
  5.         String normalExchange = "normal_exchange";
  6.         // 死信交换机名称
  7.         String deadExchange = "dead_exchange";
  8.         // 普通队列名称
  9.         String normalQueue = "normal-queue";
  10.         // 死信队列名称
  11.         String deadQueue = "dead-queue";
  13.         Channel channel = RabbitMqUtils.getChannel();
  15.         // 声明死信和普通交换机类型为 direct
  16.         channel.exchangeDeclare(normalExchange, BuiltinExchangeType.DIRECT);
  17.         channel.exchangeDeclare(deadExchange, BuiltinExchangeType.DIRECT);
  19.         // 声明死信队列
  20.         channel.queueDeclare(deadQueue, false, false, false, null);
  21.         // 死信队列绑定死信交换机与 routingkey
  22.         channel.queueBind(deadQueue, deadExchange, "lisi");
  24.         // 正常队列绑定死信队列信息
  25.         Dict params = Dict.create()
  26.                 // 正常队列设置死信交换机 参数 key 是固定值
  27.                 .set("x-dead-letter-exchange", deadExchange)
  28.                 // 正常队列设置死信 routing-key 参数 key 是固定值
  29.                 .set("x-dead-letter-routing-key", "lisi");
  31.         channel.queueDeclare(normalQueue, false, false, false, params);
  32.         channel.queueBind(normalQueue, normalExchange, "zhangsan");
  34.         System.out.println("等待接收消息    ");
  35.         DeliverCallback deliverCallback = (consumerTag, delivery) -> {
  36.             String message = new String(delivery.getBody(), StandardCharsets.UTF_8);
  37.             System.out.println("Consumer01 接收到消息" + message);
  38.         };
  39.         channel.basicConsume(normalQueue, true, deliverCallback, consumerTag -> {
  40.         });
  41.     }
  42. }

生产者代码

  1. public class Producer {
  3.     public static void main(String[] argv) throws Exception {
  4.         // 普通交换机名称
  5.         String normalExchange = "normal_exchange";
  6.         Channel channel = RabbitMqUtils.getChannel();
  8.         // 设置消息的TTL时间 time to live 10000ms=10s
  9.         AMQP.BasicProperties properties = new AMQP.BasicProperties().builder().expiration("10000").build();
  11.         // 该信息是用作演示队列个数限制
  12.         for (int i = 1; i <= 10; i++) {
  13.             String message = "info:" + i;
  14.             channel.basicPublish(normalExchange, "zhangsan", properties, message.getBytes());
  15.             System.out.println("生产者发送消息:" + message);
  16.         }
  17.     }
  18. }

运行结果

启动c1消费者之后关闭(目的为创建交互机,防止报错)，模拟其接收不到消息； 然后启动生成者发送消息

07 死信队列 - 图2

消费者 C2 代码

以上步骤完成后 启动 C2 消费者 让它消费死信队列里面的消息

  1. public class Consumer02 {
  3.     public static void main(String[] argv) throws Exception {
  4.         // 死信队列名称
  5.         String deadQueue = "dead-queue";
  7.         Channel channel = RabbitMqUtils.getChannel();
  9.         System.out.println("等待接收消息    ");
  10.         DeliverCallback deliverCallback = (consumerTag, delivery) -> {
  11.             String message = new String(delivery.getBody(), StandardCharsets.UTF_8);
  12.             System.out.println("Consumer02 接收到消息" + message);
  13.         };
  14.         channel.basicConsume(deadQueue, true, deliverCallback, consumerTag -> {
  15.         });
  16.     }
  17. }

07 死信队列 - 图3

队列达到最大长度

消息生产者代码去掉 TTL 属性

  1. public class Producer {
  3.     public static void main(String[] argv) throws Exception {
  4.         // 普通交换机名称
  5.         String normalExchange = "normal_exchange";
  6.         Channel channel = RabbitMqUtils.getChannel();
  8.         // 设置消息的TTL时间 time to live 10000ms=10s
  9.         // AMQP.BasicProperties properties = new AMQP.BasicProperties().builder().expiration("10000").build();
  11.         // 该信息是用作演示队列个数限制
  12.         for (int i = 1; i <= 10; i++) {
  13.             String message = "info:" + i;
  14.             // channel.basicPublish(normalExchange, "zhangsan", properties, message.getBytes());
  15.             channel.basicPublish(normalExchange, "zhangsan", null, message.getBytes());
  16.             System.out.println("生产者发送消息:" + message);
  17.         }
  18.     }
  19. }

C1 消费者修改以下代码

启动之后关闭该消费者 模拟其接收不到消息
image.png

运行结果

C2 消费者代码不变(启动 C2 消费者)
运行生产者
07 死信队列 - 图5

消息被拒

修改消费者01

  1. public class Consumer01 {
  3.     public static void main(String[] argv) throws Exception {
  4.         // 普通交换机名称
  5.         String normalExchange = "normal_exchange";
  6.         // 死信交换机名称
  7.         String deadExchange = "dead_exchange";
  8.         // 普通队列名称
  9.         String normalQueue = "normal-queue";
  10.         // 死信队列名称
  11.         String deadQueue = "dead-queue";
  13.         Channel channel = RabbitMqUtils.getChannel();
  15.         // 声明死信和普通交换机类型为 direct
  16.         channel.exchangeDeclare(normalExchange, BuiltinExchangeType.DIRECT);
  17.         channel.exchangeDeclare(deadExchange, BuiltinExchangeType.DIRECT);
  19.         // 声明死信队列
  20.         channel.queueDeclare(deadQueue, false, false, false, null);
  21.         // 死信队列绑定死信交换机与 routingkey
  22.         channel.queueBind(deadQueue, deadExchange, "lisi");
  24.         // 正常队列绑定死信队列信息
  25.         Dict params = Dict.create()
  26.                 // 正常队列设置死信交换机 参数 key 是固定值
  27.                 .set("x-dead-letter-exchange", deadExchange)
  28.                 // 正常队列设置死信 routing-key 参数 key 是固定值
  29.                 .set("x-dead-letter-routing-key", "lisi")
  30.                 // 设置正常队列长度
  31.                 .set("x-max-length",6);
  33.         channel.queueDeclare(normalQueue, false, false, false, params);
  34.         channel.queueBind(normalQueue, normalExchange, "zhangsan");
  36.         System.out.println("等待接收消息    ");
  37.         DeliverCallback deliverCallback = (consumerTag, delivery) -> {
  38.             String message = new String(delivery.getBody(), StandardCharsets.UTF_8);
  39.             if(message.equals("info5")){
  40.                 // requeue 设置为 false 代表拒绝重新入队 该队列如果配置了死信交换机将发送到死信队列中
  41.                 channel.basicReject(delivery.getEnvelope().getDeliveryTag(), false);
  42.                 System.out.println("Consumer01 接收到消息" + message + "并拒绝签收该消息");
  43.             }else{
  44.                 System.out.println("Consumer01 接收到消息" + message);
  45.             }
  46.         };
  47.         boolean autoAck = false; // 取消自动应答
  48.         channel.basicConsume(normalQueue, autoAck, deliverCallback, consumerTag -> {
  49.         });
  50.     }
  51. }

运行结果

07 死信队列 - 图6
启动消费者01,02
07 死信队列 - 图7