RabbitMQ 之 死信队列

1.死信的概念

先从概念上搞清楚这个定义，死信，顾名思义就是无法被消费的消息，字面意思可以这样理解。一般来说，producer 将消息投递到 broker 或者直接到 queue 里了，consumer 从 queue 取出消息进行消费，但某些时候由于特定的原因导致 queue 中的某些消息无法被消费，这样的消息如果没有后续的处理，就变成了死信，有死信自然就有了死信队列。

应用场景：为了保证订单业务的消息数据不丢失，需要使用到 RabbitMQ 的死信队列机制，当消息消费发生异常时，将消息投入死信队列中。还有，比如说：用户在商城下单成功后并点击去支付后在指定时间未支付时自动失效。

2.死信的来源

消息 TTL 过期 队列达到最大长度（队列满了，无法再添加数据到 mq 中） 消息被拒绝（basic.reject 或 basic.nack）并且 requeue=false.

3.死信实战 3.1 代码结构图

3.2 消息 TTL 过期

生产者代码：

package com.wpp.test.rabbitmq_test.deadqueue.ttl;

import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.wpp.test.rabbitmq_test.utils.RabbitMQUtils;
import lombok.extern.slf4j.Slf4j;

/**
 * @Author wpp
 * @Date 2022/09/07/14:36
 * @Description
 */
@Slf4j
public class Producer {

    public static final String NORMAL_EXCHANGE = "normal_exchange";

    public static void main(String[] args) throws Exception {
        Channel channel = RabbitMQUtils.getChannel();
        channel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT);
        // 设置消息的 TTL 时间
        AMQP.BasicProperties properties = new AMQP.BasicProperties().builder().expiration("20000").build();
        // 该信息是用作演示队列个数限制
        for (int i = 1; i  {
            String message = new String(deliver.getBody());
            log.info("Consumer1 接收到消息:" + message);
        };
        channel.basicConsume(normalQueue,true,deliverCallback,consumerTag -> {

        });
    }

}

消费者2代码（以上步骤完成后，启动消费者2，它消费死信队列里面的消息）

package com.wpp.test.rabbitmq_test.deadqueue.ttl;

import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;
import com.wpp.test.rabbitmq_test.utils.RabbitMQUtils;
import lombok.extern.slf4j.Slf4j;

/**
 * @Author wpp
 * @Date 2022/09/14/8:58
 * @Description
 */
@Slf4j
public class Consumer2 {

    //死信交换机名称
    private static final String DEAD_EXCHANGE = "dead_exchange";

    public static void main(String[] args) throws Exception {
        Channel channel = RabbitMQUtils.getChannel();
        channel.exchangeDeclare(DEAD_EXCHANGE, BuiltinExchangeType.DIRECT);
        String deadQueue = "dead_queue";
        channel.queueDeclare(deadQueue,false,false,false,null);
        channel.queueBind(deadQueue,DEAD_EXCHANGE,"lisi");
        log.info("等待接受死信队列消息...");
        DeliverCallback deliverCallback = (consumerTag, deliver) -> {
            String message = new String(deliver.getBody());
            log.info("Consumer2 接收死信队列的消息:" + message);
        };
        channel.basicConsume(deadQueue,true,deliverCallback,consumerTag -> {

        });
    }

}

3.3 队列达到最大长度

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

package com.wpp.test.rabbitmq_test.deadqueue.maxlength;

import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.wpp.test.rabbitmq_test.utils.RabbitMQUtils;
import lombok.extern.slf4j.Slf4j;

/**
 * @Author wpp
 * @Date 2022/09/07/14:36
 * @Description
 */
@Slf4j
public class Producer {

    public static final String NORMAL_EXCHANGE = "normal_exchange";

    public static void main(String[] args) throws Exception {
        Channel channel = RabbitMQUtils.getChannel();
        channel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT);
        // 该信息是用作演示队列个数限制
        for (int i = 1; i  {
            String message = new String(deliver.getBody());
            log.info("Consumer1 接收到消息:" + message);
        };
        channel.basicConsume(normalQueue, true, deliverCallback, consumerTag -> {

        });
    }

}

3.消费者2代码不变（启动消费者2）

3.4 消息被拒

1.消息生产者代码不变 2.消费者1修改以下代码（启动之后关闭消费者1，模拟其接收不到消息）

package com.wpp.test.rabbitmq_test.deadqueue.reject;

import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;
import com.wpp.test.rabbitmq_test.utils.RabbitMQUtils;
import lombok.extern.slf4j.Slf4j;

import java.util.HashMap;
import java.util.Map;

/**
 * @Author wpp
 * @Date 2022/09/07/14:44
 * @Description
 */
@Slf4j
public class Consumer1 {

    //普通交换机名称
    public static final String NORMAL_EXCHANGE = "normal_exchange";
    //死信交换机名称
    private static final String DEAD_EXCHANGE = "dead_exchange";

    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);

        //声明死信队列
        String deadQueue = "dead_queue";
        channel.queueDeclare(deadQueue, false, false, false, null);
        //使用routingKey将死信队列和死信交换机进行绑定
        channel.queueBind(deadQueue, DEAD_EXCHANGE, "lisi");

        //正常队列绑定死信队列
        Map params = new HashMap();
        //正常队列设置死信交换机 参数是固定值
        params.put("x-dead-letter-exchange", DEAD_EXCHANGE);
        //正常队列设置设置死信 routingKey 参数 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");

        log.info("等待接收消息...");
        DeliverCallback deliverCallback = (consumerTag, deliver) -> {
            String message = new String(deliver.getBody());
            if (message.equals("info5")) {
                // requeue 设置为 false 代表拒绝重新入队，该队列如果配置了死信交换机将发送到死信队列中
                channel.basicReject(deliver.getEnvelope().getDeliveryTag(),false);
            } else {
                log.info("Consumer1 接收到消息:" + message);
                channel.basicAck(deliver.getEnvelope().getDeliveryTag(),false);
            }
        };
        channel.basicConsume(normalQueue, false, deliverCallback, consumerTag -> {

        });
    }

}

3.消费者2代码不变 启动生产者、消费者1、消费者2