RabbitMQ 学习（五）-- 死信队列

死信队列(Dead Letter Exchanges) : Dead Letter Exchanges — RabbitMQ

1、概念

DLX，全称为 Dead Letter Exchanges , 可以称之为死信交换机，也有人称之为死信邮箱。当消息在一个队列中变成死信(dead message)之后，它能被重新发送到另一个交换机中，这个交换机就是 DLX ，绑定 DLX 的队列就称之为死信队列。

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

2、死信的来源

消息变成死信的原因：

• 消费者使用 basic.reject 或 basic.nack 否定确认消息，并且将 requeue 参数设置为 false.。（消息被拒绝）

• 消息 TTL 过期。（消息过期）

• 消息被丢弃，因为它的队列达到最大长度（队列满了）

3、死信实战

1）代码架构图

生产者只关心我消息发给哪个交换机 ,不关心你转发给哪个队列 ,队列在消费者声明绑定就好了

生产者只负责生产消息和发送给交换机，其余消息有什么故障都是消费者所导致的

死信队列其实是绑定在死信交换机上的普通队列，而死信交换机也只是一个普通的交换机，不过是用来专门处理死信的交换机

消费者01 要声明 normal_exchange 和 dead_exchange 以及 normal_queue 和 dead_queue 对列

"queue" – 队列的名称
"durable"——如果我们声明一个持久队列，则为真（该队列将在服务器重启后继续存在）
"exclusive" – 如果我们声明一个独占队列，则为 true（仅限于此连接）
"autoDelete" – 如果我们声明一个自动删除队列，则为 true（服务器将在不再使用时将其删除）
"arguments" – 队列的其他属性（构造参数）

Queue.DeclareOk queueDeclare(String queue, 
                             	boolean durable, 
                             	boolean exclusive, 
                            	boolean autoDelete,
                                Map<String, Object> arguments) throws IOException;
1
2
3
4
5
6
7
8
9
10
11

2）消费者01 代码

/**
 * @desc
 * @auth llp
 * @date 2022年08月04日 22:48
 */
public class Consumer01 {
    /** 普通交换机名称 */
    private static final String NORMAL_EXCHANGE = "normal_exchange";
    /** 普通队列名称 */
    private static final String NORMAL_QUEUE = "normal_queue";
    /** 死信交换机名称 */
    private static final String DEAD_EXCHANGE = "dead_exchange";
    /** 死信队列名称 */
    private static final String DEAD_QUEUE = "dead_queue";

    public static void main(String[] args) throws IOException, TimeoutException {
        Channel channel = RabbitMQUtil.getChannel();
        // 声明交换机，类型为 direct
        channel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT);
        channel.exchangeDeclare(DEAD_EXCHANGE, BuiltinExchangeType.DIRECT);

        // 声明普通队列
        Map<String, Object> arguments = new HashMap<>();
        // 过期时间 单位 ms 注：也可以在生产者发送消息指定（比较灵活）
        // arguments.put("x-message-ttl", 5000);
        // 正常的队列要设置死信交换机    注意：key 是固定的
        arguments.put("x-dead-letter-exchange", DEAD_EXCHANGE);
        // 设置死信 RoutingKey
        arguments.put("x-dead-letter-routing-key", "dead");
        channel.queueDeclare(NORMAL_QUEUE, false, false, false, arguments);
        // 声明死信队列
        channel.queueDeclare(DEAD_QUEUE, false, false, false, null);

        // 交换机和队列绑定
        channel.queueBind(NORMAL_QUEUE, NORMAL_EXCHANGE, "normal");
        channel.queueBind(DEAD_QUEUE, DEAD_EXCHANGE, "dead");

        // 接收消息
        System.out.println("Consumer01 等待接收消息...");
        // 消息消费成功的回调  consumerTag：与消费者关联的消费者标签
        DeliverCallback deliverCallback = (consumerTag, message) ->
                System.out.println("Consumer01 接收到的消息为 ==> " + new String(message.getBody(), StandardCharsets.UTF_8));
        // 消息消费失败的回调
        CancelCallback cancelCallback = consumerTag -> System.out.println(consumerTag+" 消息消费中断...");
        channel.basicConsume(NORMAL_QUEUE, true, deliverCallback, cancelCallback);
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47

3）生产者代码

/**
 * @desc
 * @auth llp
 * @date 2022年08月04日 23:16
 */
public class Producer {
    /** 普通交换机名称 */
    private static final String NORMAL_EXCHANGE = "normal_exchange";
    public static void main(String[] args) throws IOException, TimeoutException {
        Channel channel = RabbitMQUtil.getChannel();
        // 死信消息 设置 TTL 时间 time to live 单位 ms
        AMQP.BasicProperties properties =
                new AMQP.BasicProperties()
                        .builder().expiration("5000").build();
        for (int i = 1; i <= 10; i++) {
            String msg = "message" + i;
            channel.basicPublish(NORMAL_EXCHANGE, "normal", properties, msg.getBytes());
        }
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

4）消费者02 代码

/**
 * @desc
 * @auth llp
 * @date 2022年08月04日 22:48
 */
public class Consumer02 {
    /** 死信队列名称 */
    private static final String DEAD_QUEUE = "dead_queue";

    public static void main(String[] args) throws IOException, TimeoutException {
        Channel channel = RabbitMQUtil.getChannel();
        // 接收消息
        System.out.println("Consumer02 等待接收消息...");
        // 消息消费成功的回调  consumerTag：与消费者关联的消费者标签
        DeliverCallback deliverCallback = (consumerTag, message) ->
                System.out.println("Consumer02 接收到的消息为 ==> " + new String(message.getBody(), StandardCharsets.UTF_8));
        // 消息消费失败的回调
        CancelCallback cancelCallback = consumerTag -> System.out.println(consumerTag+" 消息消费中断...");
        channel.basicConsume(DEAD_QUEUE, true, deliverCallback, cancelCallback);
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

5）消息 TTL 过期

先启动 消费者01 的代码，声明对应的交换机和消息队列，然后停掉 消费者01 。再启动 生产者 ，发送的设置了 TTL 的消息给对列。

测试结果：

启动 消费者02 消费死信队列里的消息

​

6）队列达到最大长度

消费者01 添加以下代码：

// 设置普通队列的长度    注意：key 是固定的
arguments.put("x-max-length", 6);
1
2

生产者 将过期时间的代码去掉

// 第三个参数设置为 null
channel.basicPublish(NORMAL_EXCHANGE, "normal", null, msg.getBytes());
1
2

注意：此时需要先把原的队列删除，因为参数设置改变了

运行测试

重启 消费者01 之后需要将其关掉，不让消费消息，导致消息积压，启动 生产者 看是否有四条消息发到死信队列上。

7）消息被拒绝

将队列的最大长度参数设置注释掉，修改 消费者01 的代码为：

// 消息消费成功的回调  consumerTag：与消费者关联的消费者标签
DeliverCallback deliverCallback = (consumerTag, message) -> {
    String msg = new String(message.getBody(), StandardCharsets.UTF_8);
    if ("message5".equals(msg)){
        System.out.println("Consumer01 拒绝的消息为 ==> " + msg);
        channel.basicReject(message.getEnvelope().getDeliveryTag(), false);
    }else {
        System.out.println("Consumer01 接收到的消息为 ==> " + msg);
        channel.basicAck(message.getEnvelope().getDeliveryTag(), false);
    }
};
// 消息消费失败的回调
CancelCallback cancelCallback = consumerTag -> System.out.println(consumerTag+" 消息消费中断...");
// 手动应答 参数2：false
channel.basicConsume(NORMAL_QUEUE, false, deliverCallback, cancelCallback);
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

运行测试