RabbitMQ--死信队列/延迟队列--使用/原理

Posted 2021-12-14 IT利刃出鞘

原文网址：RabbitMQ--死信队列/延迟队列--使用/原理_IT利刃出鞘的博客-CSDN博客

简介

        本文介绍RabbitMQ的死信队列和延迟队列。

        本内容也是Java后端面试中常见的问题。

死信队列

简介

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

以下几种情况会导致消息变成死信：

1. 消息被拒绝（Basic.Reject/Basic.Nack)，并且设置requeue参数为false;
2. 消息过期；
3. 队列达到最大长度。

        DLX是一个正常的交换器，和一般的交换器没有区别，它能在任何的队列上被指定，实际上就是设置某个队列的属性。当这个队列中存在死信时，RabbitMQ就会自动地将这个消息重新发布到设置的DLX上去，进而被路由到另一个队列，即死信队列。可以监听这个队列中的消息以进行相应的处理，这个特性与将消息的TTL设置为0配合使用可以弥补immediate参数的功能。

为队列添加DLX的方法

法1：代码方式

//创建 DLX: dlx_exchange
channel.exchangeDeclare("dlx_exchange", "direct" );
Map<String, Object> args = new HashMap<String, Object>;
args.put("x-dead-letter-exchange", "dlx_exchange");
//为队列myqueue添加DLX
channel.queueDeclare("myqueue", false, false, false, args);

也可以为这个DLX指定路由键。（如果没有特殊指定，则使用原队列的路由键）

args.put("x-dead-letter-routing-key","dlx-routing-key"); 

法2：命令方式

rabbitmqctl set_policy DLX ".*" '"dead-letter-exchange":"dlx_exchange"' --apply-to queues

示例

代码

channel.exchangeDeclare("exchange.dlx", "direct", true);
channel.exchangeDeclare("exchange.normal", "fanout", true);
Map<String, Object> args = new HashMap<String, Object>();
args.put("x-message-ttl", 10000);
args.put("x-dead-letter-exchange" , "exchange.dlx");
args.put("x-dead-letter-routing-key" , "routingkey");
channel.queueDeclare("queue.normal" , true, false, false, args);
channel.queueBind("queue.normal", "exchange.normal", "");
channel.queueDeclare("queue.dlx", true, false, false, null);
channel.queueBind("queue.dlx", "exchange.dlx" , "routingkey");
channel.basicPublish("exchange.normal" , "rk",
MessageProperties.PERSISTENT_TEXT_PLAIN, "dlx".getBytes());

        这里创建了两个交换器exchange.normal和exchange.dlx，分别绑定两个队列queue.normal和queue.dlx。

Web管理页面结果

        由下图（图1-1）的Web管理页面可以看出，两个队列都被标记了“D”，这个是durable的缩写，即设置了队列持久化。queue.normal这个队列还配置了TTL、DLX和DLK，其中DLX指的是
x-dead-letter-routing-key这个属性。 

图1-1

案例分析

        参考下图（图1-2），生产者首先发送一条携带路由键为“rk”的消息，然后经过交换器exchange.normal顺利地存储到队列queue.normal中。由于队列queue.normal设置了过期时间为10s，在这10s内没有消费者消费这条消息，那么判定这条消息为过期。由于设置了DLX，过期之时，消息被丢给交换器exchange.dlx中，这时找到与exchange.dlx匹配的队列queue.dlx，最后消息被存储在queue.dk这个死信队列中。 

图1-2

        对于RabbitMQ来说，DLX是一个非常有用的特性。它可以处理异常情况下，消息不能够被消费者正确消费（消费者调用了Basic.Nack或者Basic.Reject)而被置入死信队列中
的情况，后续分析程序可以通过消费这个死信队列中的内容来分析当时所遇到的异常情况，进而可以改善和优化系

统。DLX配合TTL使用还可以实现延迟队列的功能，详细请看下一节。

延迟队列

简介

        延迟队列用来存放延迟消息。延迟消息：指当消息被发送以后，不想让消费者立刻拿到消息，而是等待特定时间后，消费者才能拿到这个消息进行消费。

        在AMQP协议中，或者RabbitMQ本身没有直接支持延迟队列的功能，但是有两种方案来间接实现：

• 方案1：采用rabbitmq-delayed-message-exchange 插件实现。（RabbitMQ 3.6.x开始支持）
• 方案2：通过前面所介绍的DLX和TTL模拟出延迟队列的功能。

        在图1-2中，不仅展示的是死信队列的用法，也是延迟队列的用法，对于queue.dlx这个死信队列来说，同样可以看作延迟队列。假设一个应用中需要将每条消息都设置为10秒的延迟，
生产者通过exchange.normal这个交换器将发送的消息存储在queue.normal这个队列中。消费者订阅的并非是queue.normal这个队列，而是queue.dlx这个队列。当消息从queue.normal这个队列中过期之后被存入queue.dlx这个队列中，消费者就恰巧消费到了延迟10秒的这条消息。

        在真实应用中，对于延迟队列可以根据延迟时间的长短分为多个等级，一般分为5秒、10秒、30秒、1分钟、5分钟、10分钟、30分钟、1小时这几个维度，当然也可以再细化一下。

        以下图（图2-1）为例进行说明。为简化，只设置5秒、10秒、30秒、1分钟这四个等级。根据需求的不同，生产者发送消息的时候通过设置不同的路由键，将消息发送到与交换器绑定的不同的队列中。这里队列也分别配置了DLX和相应的死信队列，当相应的消息过期时，就会转存到相应的死信队列（即延迟队列）中，这样消费者根据业务自身的情况，分别选择不同延迟等级的延迟队列进行消费。

图2-1

使用场景

延迟队列的使用场景有很多，比如：

• 用户下订单场景：用户下单后有30分钟的时间支付，若30分钟内没有支付，则将这个订单取消。
  • 方案：用户下单后将取消订单的消息发送到延迟队列，延迟时间设置为30分钟。取消订单这个消息的订阅者程序在30分钟后收到消息，判断该订单的状态是否为已支付，若还没支付，则将该订单状态设置为：已取消。
• 定时遥控场景：用户想用手机远程遥控家里的智能设备在指定的时间工作。
  • 方案：假设用户想要的操作是：开启热水器。首先，将开启热水器这个消息发送到延迟队列，延迟时间设置到用户想要的时间到现在时间的差值。开启热水器这个消息的订阅者程序在指定时间收到消息，再将指令推送到智能设备。

        需要注意的是，延迟队列的消息是不能取消的，解决方案是：在消费消息的时候判断这个消息对应的业务的当前状态。例如：对于取消订单来说，收到消息时，读取这个消息所对应的数据库信息，如果已经是已付款状态了，就不进行任何操作了，如果是未支付状态，则改为已取消。

        

其他网址

《RabbitMQ实战指南》=> 4.3 死信队列

《RabbitMQ实战指南》=> 4.4 延迟队列