【学习笔记】RabbitMQ04：延迟队列的原理以及实现代码

参考资料

RabbitMQ官方网站
RabbitMQ官方文档
噼咔噼咔-动力节点教程

文章目录

- 七、延迟队列
- - 7.1 什么是延迟队列
  - 7.2 延迟队列的解决方案
  - - 7.2.1 定时任务
    - 7.2.2 **被动取消**
    - 7.2.3 JDK的延迟队列
    - 7.2.3 采用消息中间件（rabbitMQ
    - - 7.2.3.1 适用专门优化后的死信队列实现延迟队列
      - 7.2.3.2 :star:实例代码
      - 7.2.3.2 测试结果
    - 7.2.4 使用rabbitmq_delayed_message_exchange插件.
    - - 7.2.4.1 插件下载
      - 7.2.4.2 :star:如何在docker环境下安装插件
      - 7.2.4.3 :star: 代码示例：如何使用该插件
      - 7.2.4.4 测试结果
  - 7.3 问题：多个消息的延迟时间不同该如何解决？
  - - 7.3.1 解决方案一：用延迟队列区分
    - 7.3.2 使用延迟队列插件rabbitmq_delayed_message_exchange

七、延迟队列

7.1 什么是延迟队列

正常的MQ应用场景中，我们希望消息可以快速稳定的传递。但是有一些场景中，希望在指定的延迟后再消费信息，比如订单支付场景（订单15部分内未支付则关闭订单）。

这类实现延迟任务的场景，就可以采用延迟队列来实现。

以下介绍一下其他的一些方法。

7.2 延迟队列的解决方案

7.2.1 定时任务

每隔n秒扫描一次数据库，查询数据库装为过期的订单进行处理。

实现方式

spring schedule、quartz、xxljob等

优点

简单，容易实现；

缺点

存在延迟（受定时器延迟时间限制
性能较差，每次扫描数据库，如果订单量交大，会给数据库造成较大压力。

7.2.2 被动取消

当用户主动查询订单时，判断订单是否超时，超时则取消

优点：服务器压力小
缺点：如果用户长时间不查询，则会造成统计异常；而且用户打开订单页面会变慢，严重的话会影响用户体验

7.2.3 JDK的延迟队列

DelayedQueue：无界阻塞队列，该队列只有在延迟期满后，才能从中获取元素。

优点

实现简单，任务的延迟低。

缺点

服务器重启宕机，数据会丢失
只适用于单机版
订单量大时，可能会造成内存不足：OOM

7.2.3 采用消息中间件（rabbitMQ

RabbitMQ 本身不支持延迟队列，可以使用 TTL 结合 DLX 的方式来实现消息的延迟投递（前面提到的死信队列）。.

把 DLX 跟某个队列绑定，到了指定时间，消息过期后，就会从 DLX 路由到这个队列，消费者可以从DLX的队列中取走消息。

7.2.3.1 适用专门优化后的死信队列实现延迟队列

在上面的mq方案中，存在两个不同的交换机，我们可以利用直连交换机的特性，将交换机优化成一个交换机，同时通过不同的routingKey指定普通队列和死信队列。

思路解释

生产者发送消息到交换机X，并指定ttl的key
消息被交换机传递到ttl队列中（指定了消息过期时间的队列
同时，ttl队列还指定的死信交换机DLX为自身的交换机X，但是指定的routingKey为死信队列的key
这样，当消息在ttl队列中到期后，这条消息就会被传递到死信队列中，提供给消费者

7.2.3.2 ⭐️实例代码

为了便于测试，将发送和接收写在同一个服务中

配置信息

@Configuration
public class DelayExchangeConfig {public static String exchangeName = "order.ttl.exchange";public static String orderQ = "order.ttl.queue";public static String dlxQ = "order.dlx.queue";@Beanpublic DirectExchange delayedExchange(){return ExchangeBuilder.directExchange(exchangeName).build();}@Beanpublic Queue orderQueue(){// 指定该队列的过期时间和死信队列Map<String , Object> properties = new HashMap<>();properties.put("x-message-ttl" , 15000);properties.put("x-dead-letter-exchange" , exchangeName);properties.put("x-dead-letter-routing-key" , "dead-letter");return QueueBuilder.durable(orderQ).withArguments(properties).build();}@Beanpublic Queue dlxQueue(){return QueueBuilder.durable(dlxQ).build();}@Beanpublic Binding dlxBinding1(){return BindingBuilder.bind(dlxQueue()).to(this.delayedExchange()).with("dead-letter");}@Beanpublic Binding ttlBinding1(){return BindingBuilder.bind(dlxQueue()).to(this.delayedExchange()).with("order");}}

测试代码

@RestController
@RequestMapping("/delay")
@Slf4j
public class DelayedController {@Resourceprivate RabbitTemplate rabbitTemplate;@GetMapping("/{msg}")public void sentErrorMsg(@PathVariable("msg") String msg) {log.info("(延迟队列)准备发送的信息：{} ， 路由键 :{}", msg, "order");// 发送到普通的延时列表中rabbitTemplate.convertAndSend(exchangeName, "order", msg.getBytes(StandardCharsets.UTF_8));log.info("(延迟队列)成功发送！发送时间{}" , LocalDateTimeUtil.now());}@RabbitListener(queues = "order.dlx.queue")public void receiveDelayedMsg(Message message){log.info("(延迟队列)接受到的消息是：{}" , new String(message.getBody()));}
}

7.2.3.2 测试结果

配置正确

控制台打印正确：15秒后接收到的了之前发送的信息

7.2.4 使用rabbitmq_delayed_message_exchange插件.

7.2.4.1 插件下载

插件下载地址

https://www.rabbitmq.com/community-plugins.html
https://github.com/rabbitmq/rabbitmq-delayed-message-exchange/releases
- 根据自己的rabbit版本，我这里用的是3.9

7.2.4.2 ⭐️如何在docker环境下安装插件

参考文章：https://juejin.cn/post/7138717546894589966

将下载到的文件，移动到容器内

docker cp rabbitmq_delayed_message_exchange-3.9.0.ez rabbitmq:/plugins

进入容器bash指令，并启动插件

docker exec -it rabbitmq bashroot@rabbit:/# rabbitmq-plugins enable rabbitmq_delayed_message_exchange# 使用下面的指令查看插件列表
rabbitmq-plugins list

进入控制台新建交换机，能查看到新的交换机类型

7.2.4.3 ⭐️ 代码示例：如何使用该插件

官方说明文档：https://github.com/rabbitmq/rabbitmq-delayed-message-exchange#usage

理解原理：delay exchange在接受到消息后，会先存在内部数据库中，检查x-delay延迟时间（头部

代码使用思路

要创建自定义的交换机类型，要使用CustomExchange()来创建。几个参数的解释如下：
- name：rabbit中交换机的名称
- type：交换机类型（x-delayed-message）
- durable：是否持久
- autoDelete：是否自动删除
- arguments：参数信息

arguments：参数信息从官方文档中获取

// ... elided code ...
Map<String, Object> args = new HashMap<String, Object>();
args.put("x-delayed-type", "direct");
channel.exchangeDeclare("my-exchange", "x-delayed-message", true, false, args);
// ... more code ...

交换机创建好后，只需要创建一条队列即可，并进行绑定
注意：消息发送需要在头部存放信息：headers.put("x-delay", 延迟时间)。不需要使用自带的expiration来控制延迟时间了

配置类

@Configuration
public class DelayPluginConfig {public static String exchangeName = "delay-x-plugin.x";public static String key = "demo";@Beanpublic CustomExchange customExchange(){// 参考官方文档，创建插件提供的自定义交换机Map<String, Object> args = new HashMap<>();args.put("x-delayed-type", "direct");// public CustomExchange(String name, String type, boolean durable, boolean autoDelete, Map<String, Object> arguments)return new CustomExchange(exchangeName, "x-delayed-message" , true , false , args);}@Beanpublic Queue delayDemoQueue(){return QueueBuilder.durable("delay-x-plugin.queue.demo").build();}@Beanpublic Binding delayPluginBinding(){return BindingBuilder.bind(delayDemoQueue()).to(customExchange()).with(key).noargs();}
}

生产者

@RestController
@RequestMapping("/delay/plugin")
@Slf4j
public class DelayedPluginController {@Resourceprivate RabbitTemplate rabbitTemplate;@GetMapping("/{delay}/{msg}")public void sentErrorMsg(@PathVariable("msg") String msg, @PathVariable("delay") Long delay) {log.info("(延迟插件队列)准备发送的信息：{} ，延迟时间：{} 路由键 :{}", msg, delay , "demo");// 在头部设置过期时间MessageProperties properties = new MessageProperties();properties.setHeader("x-delay", delay);Message message = MessageBuilder.withBody(msg.getBytes(StandardCharsets.UTF_8)).andProperties(properties).build();// 发送信息rabbitTemplate.convertAndSend(exchangeName, "demo", message);log.info("(延迟插件队列)成功发送！发送时间:{}", new SimpleDateFormat("yyyy-MM-dd hh:mm:ss").format(new Date()));}@RabbitListener(queues = "delay-x-plugin.queue.demo")public void receiveDelayedMsg(Message message) {log.info("(延迟插件队列)接受到的消息是：{}", new String(message.getBody()));}
}