RabbitMQ 死信队列

参考：https://blog.csdn.net/weixin_44688301/article/details/116237294

RabbitMQ是流行的开源消息队列系统，使用erlang语言开发，由于其社区活跃度高，维护更新较快，性能稳定，深得很多企业的欢心（当然，也包括我现在所在公司【手动滑稽】）。

为了保证订单业务的消息数据不丢失，需要使用到RabbitMQ的死信队列机制，当消息消费发生异常时，将消息投入死信队列中。但由于对死信队列的概念及配置不熟悉，导致曾一度陷入百度的汪洋大海，无法自拔，很多文章都看起来可行，但是实际上却并不能帮我解决实际问题。最终，在官网文档中找到了我想要的答案，通过官网文档的学习，才发现对于死信队列存在一些误解，导致配置死信队列之路困难重重。

于是本着记录和分享的精神，将死信队列的概念和配置完整的写下来，以便帮助遇到同样问题的朋友。

本文阅读前，需要对RabbitMQ有一个简单的了解，偏向实战配置讲解。

一、死信队列是什么

死信，在官网中对应的单词为“Dead Letter”，可以看出翻译确实非常的简单粗暴。那么死信是个什么东西呢？

“死信”是RabbitMQ中的一种消息机制，当你在消费消息时，如果队列里的消息出现以下情况，那么该消息将成为“死信”：

消息被否定确认，使用 channel.basicNack 或 channel.basicReject ，并且此时requeue 属性被设置为false。
消息在队列的存活时间超过设置的生存时间（TTL)时间。
消息队列的消息数量已经超过最大队列长度。

“死信”消息会被RabbitMQ进行特殊处理，如果配置了死信队列信息，那么该消息将会被丢进死信队列中，如果没有配置，则该消息将会被丢弃。

二、如何配置死信队列

这一部分将是本文的关键，如何配置死信队列呢？其实很简单，大概可以分为以下步骤：

配置业务队列，绑定到业务交换机上
为业务队列配置死信交换机和路由key
为死信交换机配置死信队列

注意，并不是直接声明一个公共的死信队列，然后所以死信消息就自己跑到死信队列里去了。而是为每个需要使用死信的业务队列配置一个死信交换机，这里同一个项目的死信交换机可以共用一个，然后为每个业务队列分配一个单独的路由key。

有了死信交换机和路由key后，接下来，就像配置业务队列一样，配置死信队列，然后绑定在死信交换机上。也就是说，死信队列并不是什么特殊的队列，只不过是绑定在死信交换机上的队列。死信交换机也不是什么特殊的交换机，只不过是用来接受死信的

交换机，所以可以为任何类型【Direct、Fanout、Topic】。一般来说，会为每个业务队列分配一个独有的路由key，并对应的配置一个死信队列进行监听，也就是说，一般会为每个重要的业务队列配置一个死信队列。

有了前文这些陈述后，接下来就是惊险刺激的实战环节，这里省略了RabbitMQ环境的部署和搭建环节。

消息生产者 Send:

/**
 * 类描述：
 * 测试死信队列，消息生产者
 * @author cfl
 * @version 1.0
 * @date 2022/10/14 16:06
 */
public class Send1 {
    public static final String QUEUE_NAME = "test.queue";

    public static void main(String[] args) throws IOException, TimeoutException {
        Connection connection = ConnectionUtil.getConnection();

        Channel channel = connection.createChannel();

        boolean durable = true;
        Map<String, Object> arguments = new HashMap<>();
        // 这里声明当前队列绑定的死信交换机
        arguments.put("x-dead-letter-exchange", "test.dead.letter.exchange");
        // 这里声明当前队列的死信路由key
        arguments.put("x-dead-letter-routing-key", "test.dead.letter.routing.key");
        // 这里设置队列中所有消息的有效期是5s
        arguments.put("x-message-ttl", 5000);
        // 申明队列
        channel.queueDeclare(QUEUE_NAME, durable, false, false, arguments);
        // 发送消息
        channel.basicPublish("", QUEUE_NAME, MessageProperties.PERSISTENT_TEXT_PLAIN, "send message".getBytes("utf-8"));

        channel.close();
        connection.close();
    }
}

消息消费者 DeadLetterRecv：

/**
 * 类描述：
 * 测试死信队列，死信队列消费者
 * @author cfl
 * @version 1.0
 * @date 2022/10/14 16:17
 */
public class DeadLetterRecv {

    public static void main(String[] args) throws IOException, TimeoutException {
        // 获取连接
        Connection connection = ConnectionUtil.getConnection();
        // 创建通道
        Channel channel = connection.createChannel();
        // 申明死信交换机
        channel.exchangeDeclare("test.dead.letter.exchange", "direct", true);
        // 申明死信队列（队列不存在时，会出现异常）
        channel.queueDeclare("test.dead.letter.queue", true, false, false, null);
        // 绑定队列交换机路由key
        channel.queueBind("test.dead.letter.queue", "test.dead.letter.exchange", "test.dead.letter.routing.key", null);

        // 接收消息
        channel.basicConsume("test.dead.letter.queue", false, new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("死信队列接收到消息：" + new String(body));
                // System.out.printf("死信消息properties：%s", properties);
                channel.basicAck(envelope.getDeliveryTag(), false);
            }
        });
    }
}

启动DeadLetterRecv后再启动Send1，当Send1发送的消息超时后（5s）就会发现DeadLetterRecv消费到了消息。

上面代码使用了过期消息来完成验证，也可以单使用Nack

// 不重新放入队列
boolean requeue = false;
channel.basicNack(envelope.getDeliveryTag(), false, requeue);

// 不重新放入队列
boolean requeue = false;
channel.basicReject(envelope.getDeliveryTag(), requeue);

三、死信消息的变化

那么“死信”被丢到死信队列中后，会发生什么变化呢？

如果队列配置了参数 x-dead-letter-routing-key 的话，“死信”的路由key将会被替换成该参数对应的值。如果没有设置，则保留该消息原有的路由key。

另外，由于被抛到了死信交换机，所以消息的Exchange Name也会被替换为死信交换机的名称。

消息的Header中，也会添加很多奇奇怪怪的字段，修改一下上面的代码，在死信队列的消费者中添加一行日志输出：

// 输出死信消息properties：#contentHeader<basic>(content-type=text/plain, content-encoding=null, headers={x-first-death-exchange=, x-death=[{reason=expired, count=1, exchange=, time=Fri Oct 14 20:17:24 CST 2022, routing-keys=[test.queue], queue=test.queue}], x-first-death-reason=expired, x-first-death-queue=test.queue}, delivery-mode=2, priority=0, correlation-id=null, reply-to=null, expiration=null, message-id=null, timestamp=null, type=null, user-id=null, app-id=null, cluster-id=null)
System.out.printf("死信消息properties：%s", properties);

Header中看起来有很多信息，实际上并不多，只是值比较长而已。下面就简单说明一下Header中的值：

字段名含义

x-first-death-exchange：第一次被抛入的死信交换机的名称
x-first-death-reason：第一次成为死信的原因（rejected：消息在重新进入队列时被队列拒绝，由于default-requeue-rejected 参数被设置为false。expired ：消息过期。maxlen ：队列内消息数量超过队列最大容量）
x-first-death-queue：第一次成为死信前所在队列名称
x-death：历次被投入死信交换机的信息列表，同一个消息每次进入一个死信交换机，这个数组的信息就会被更新

四、死信队列应用场景

通过上面的信息，我们已经知道如何使用死信队列了，那么死信队列一般在什么场景下使用呢？

一般用在较为重要的业务队列中，确保未被正确消费的消息不被丢弃，一般发生消费异常可能原因主要有由于消息信息本身存在错误导致处理异常，处理过程中参数校验异常，或者因网络波动导致的查询异常等等，当发生异常时，当然不能每次通过日志来获取原消息，然后让运维帮忙重新投递消息（没错，以前就是这么干的= =）。通过配置死信队列，可以让未正确处理的消息暂存到另一个队列中，待后续排查清楚问题后，编写相应的处理代码来处理死信消息，这样比手工恢复数据要好太多了。

五、总结

死信队列其实并没有什么神秘的地方，不过是绑定在死信交换机上的普通队列，而死信交换机也只是一个普通的交换机，不过是用来专门处理死信的交换机。

总结一下死信消息的生命周期：

业务消息被投入业务队列
消费者消费业务队列的消息，由于处理过程中发生异常，于是进行了nck或者reject操作
被nck或reject的消息由RabbitMQ投递到死信交换机中
死信交换机将消息投入相应的死信队列
死信队列的消费者消费死信消息

死信消息是RabbitMQ为我们做的一层保证，其实我们也可以不使用死信队列，而是在消息消费异常时，将消息主动投递到另一个交换机中，当你明白了这些之后，这些Exchange和Queue想怎样配合就能怎么配合。比如从死信队列拉取消息，然后发送邮件、短信、钉钉通知来通知开发人员关注。或者将消息重新投递到一个队列然后设置过期时间，来进行延时消费。

文章由极客之音整理，本文链接：https://www.bmabk.com/index.php/post/78228.html