RabbitMQ之死信队列

淩亂°似流年 2024-03-31 15:56 68阅读 0赞

## **6. 死信队列** ##

### **6.1.** **死信的概念** ###

先从概念解释上搞清楚这个定义，死信，顾名思义就是无法被消费的消息，字面意思可以这样理

解，一般来说，producer 将消息投递到 broker 或者直接到queue 里了，consumer 从 queue 取出消息

进行消费，但某些时候由于特定的**原因导致** **queue** **中的某些消息无法被消费**，这样的消息如果没有

后续的处理，就变成了死信，有死信自然就有了死信队列。

**应用场景:**

为了保证订单业务的消息数据不丢失，需要使用到 RabbitMQ 的死信队列机制，当消息

消费发生异常时，将消息投入死信队列中.还有比如说: 用户在商城下单成功并点击去支付后在指定时

间未支付时自动失效

### **6.2.** **死信的来源** ###

消息 TTL 过期

队列达到最大长度(队列满了，无法再添加数据到 mq 中)

消息被拒绝(basic.reject 或 basic.nack)并且 requeue=false.

### **6.3.** **死信实战** ###

#### **6.3.1. 代码架构图** ####

![3d83da3b19573777812f27b5fc6da7eb.png][]

#### **6.3.2. 消息TTL 过期** ####

生产者代码

/*
     * 死信队列之生产者代码
     *
     * */
     public class Producer {
     
         //普通交换机的名称
         public static final String NORMAL_EXCHANGE = "normal_exchange";
         public static void main(String[] args) throws Exception{
             Channel channel = RabbitMqUtils.getChannel();
     
             //死信消息，设置TTL时间  单位是ms  10000ms是10s
             AMQP.BasicProperties properties = new AMQP.BasicProperties().builder().expiration("10000").build();
     
             for (int i = 0; i < 10; i++) {
                 String message = "info" + i;
                 channel.basicPublish(NORMAL_EXCHANGE,"zhangsan",properties,message.getBytes(StandardCharsets.UTF_8));
             }
         }
     }

消费者

/*
     * 死信队列实战
     * 消费者01
     * */
     public class Consumer01 {
     
         //普通交换机名称
         public static final String NORMAL_EXCHANGE = "normal_exchange";
     
         //死信交换机名称
         public static final String DEAD_EXCHANGE = "dead_exchange";
     
         //普通队列名称
         public static final String NORMAL_QUEUE = "normal_queue";
     
         //死信队列名称
         public static final String DEAD_QUEUE = "dead_queue";
     
         public static void main(String[] args) throws  Exception{
             Channel channel = RabbitMqUtils.getChannel();
     
             //声明死信和普通的交换机类型为direct
             channel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT);
             channel.exchangeDeclare(DEAD_EXCHANGE, BuiltinExchangeType.DIRECT);
     
             //声明普通队列
             HashMap<String, Object> arguments = new HashMap<>();
     
             //过期时间
             arguments.put("x-message-ttl",1000);
             //正常队列设置死信队列
             arguments.put("x-dead-letter-exchange",DEAD_EXCHANGE);
             //设置死信RoutingKey
             arguments.put("x-dead-letter-routing-key","lisi");
    
         //声明死信和普通队列
         channel.queueDeclare(NORMAL_QUEUE,false,false,false,arguments);
         channel.queueDeclare(DEAD_QUEUE,false,false,false,null);
    
         //绑定普通的交换机与普通的队列
         channel.queueBind(NORMAL_QUEUE,NORMAL_EXCHANGE,"zhangsan");
         //绑定死信的交换机与死信的队列
         channel.queueBind(DEAD_QUEUE,DEAD_EXCHANGE,"lisi");
         System.out.println("等待接收消息......");
    
         DeliverCallback deliverCallback = (consumerTag,message) -> {
             System.out.println("Consumer01接收的消息是：" + new String(message.getBody()));
         };
    
         channel.basicConsume(NORMAL_QUEUE,true,deliverCallback,consumerTag->{});
     }
    }

效果：启动生产者后，10条消息被传送到NORMAL\_QUEUE，然后被传送到DEAD\_QUEUE，此时启动消费者02，消息全被接收。

![1ff36c5931da670a5dc5c95d19b3a991.png][]

![cbfa4d91e0c6423a982b0ffed4b9db3f.png][]

#### **6.3.3. 队列达到最大长度** ####

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

public class Producer {
        private static final String NORMAL_EXCHANGE = "normal_exchange";
        public static void main(String[] argv) throws Exception {
            try (Channel channel = RabbitMqUtils.getChannel())
            { channel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT);
    //该信息是用作演示队列个数限制
                for (int i = 1; i <11 ; i++) {
                    String message="info"+i;
                    channel.basicPublish(NORMAL_EXCHANGE,"zhangsan",null, message.getBytes());
                    System.out.println("生产者发送消息:"+message);
                }
            }
        }
    }

2. C1 消费者修改以下代码(**启动之后关闭该消费者 模拟其接收不到消息**)

![a2df5259b06f1264c88f54583dfd31f7.png][]

**注意此时需要把原先队列删除 因为参数改变了**

3. C2 消费者代码不变(启动 C2 消费者)

![ba3e9759d91646c14eae385f298864dc.png][]

#### **6.3.4. 消息被拒** ####

1.消息生产者代码同上生产者一致

2.C1 消费者代码(**启动之后关闭该消费者模拟其接收不到消息**)

![ec3c10980c3427a41d0e596317a3805b.png][]

public class Consumer01 {
        //普通交换机名称
        private static final String NORMAL_EXCHANGE = "normal_exchange";
        //死信交换机名称
        private static final String DEAD_EXCHANGE = "dead_exchange";
        public static void main(String[] argv) throws Exception {
            Channel channel = RabbitUtils.getChannel();
    //声明死信和普通交换机 类型为 direct
            channel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT);
            channel.exchangeDeclare(DEAD_EXCHANGE, BuiltinExchangeType.DIRECT);
    //声明死信队列
            String deadQueue = "dead-queue";
            channel.queueDeclare(deadQueue, false, false, false, null);
    //死信队列绑定死信交换机与 routingkey
            channel.queueBind(deadQueue, DEAD_EXCHANGE, "lisi");
            //正常队列绑定死信队列信息
            Map<String, Object> params = new HashMap<>();
    //正常队列设置死信交换机 参数 key 是固定值
            params.put("x-dead-letter-exchange", DEAD_EXCHANGE);
    //正常队列设置死信 routing-key 参数 key 是固定值
            params.put("x-dead-letter-routing-key", "lisi");
            String normalQueue = "normal-queue";
            channel.queueDeclare(normalQueue, false, false, false, params);
            channel.queueBind(normalQueue, NORMAL_EXCHANGE, "zhangsan");
            System.out.println("等待接收消息........... ");
            DeliverCallback deliverCallback = (consumerTag, delivery) ->
            {String message = new String(delivery.getBody(), "UTF-8");
                if(message.equals("info5")){
                    System.out.println("Consumer01 接收到消息" + message + "并拒绝签收该消息");
    //requeue 设置为 false 代表拒绝重新入队 该队列如果配置了死信交换机将发送到死信队列中
                    channel.basicReject(delivery.getEnvelope().getDeliveryTag(), false);
                }else {
                    System.out.println("Consumer01 接收到消息"+message);
                    channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
                }
            };
            boolean autoAck = false;
            channel.basicConsume(normalQueue, autoAck, deliverCallback, consumerTag -> {
            });
        }
    }

![6e1717eebac5fa0be53a14aa819fdc0c.png][]

3.C2 消费者代码不变

**启动消费者 1 然后再启动消费者 2**

![fb545dbf89cc79e7dcc6a9a64e64912e.png][]

[3d83da3b19573777812f27b5fc6da7eb.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/31/960bca4145da4b579e242c23958299be.png
[1ff36c5931da670a5dc5c95d19b3a991.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/31/593fc6756c424da29dac73778f571e09.png
[cbfa4d91e0c6423a982b0ffed4b9db3f.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/31/dff7b8064c814d53b55fcac3ae16617e.png
[a2df5259b06f1264c88f54583dfd31f7.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/31/c2be582fa6fc4ec58a6fd00ca4bf3066.png
[ba3e9759d91646c14eae385f298864dc.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/31/b41aea1079c24b518747ecaa9f50c9a5.png
[ec3c10980c3427a41d0e596317a3805b.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/31/107957310f0f472f9ee6f3c12b8587bd.png
[6e1717eebac5fa0be53a14aa819fdc0c.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/31/46f90661e9444e57b472555e1c0c6f01.png
[fb545dbf89cc79e7dcc6a9a64e64912e.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/31/161174c521f543fa9dbdacf260188cca.png