定时任务各种各样，常见的定时任务例如日志备份，我们可能在每天凌晨 3 点去备份，这种固定时间的定时任务我们一般采用 cron 表达式就能轻松的实现，还有一些比较特殊的定时任务，向大家看电影中的定时炸弹，3分钟后爆炸，这种定时任务就不太好用 cron 去描述，因为开始时间不确定，我们开发中有的时候也会遇到类似的需求，例如：

  • 在电商项目中，当我们下单之后，一般需要 20 分钟之内或者 30 分钟之内付款，否则订单就会进入异常处理逻辑中，被取消，那么进入到异常处理逻辑中，就可以当成是一个延迟队列。
  • 我买了一个智能砂锅，可以用来煮粥，上班前把素材都放到锅里，然后设置几点几分开始煮粥，这样下班后就可以喝到香喷喷的粥了，那么这个煮粥的指令也可以看成是一个延迟任务，放到一个延迟队列中，时间到了再执行。
  • 公司的会议预定系统，在会议预定成功后，会在会议开始前半小时通知所有预定该会议的用户。
  • 安全工单超过 24 小时未处理，则自动拉企业微信群提醒相关责任人。
  • 用户下单外卖以后，距离超时时间还有 10 分钟时提醒外卖小哥即将超时。
  • …

很多场景下我们都需要延迟队列。

本文以 RabbitMQ 为例来和大家聊一聊延迟队列的玩法。

整体上来说，在 RabbitMQ 上实现定时任务有两种方式：

  • 利用 RabbitMQ 自带的消息过期和死信队列机制，实现定时任务。
  • 使用 RabbitMQ 的 rabbitmq_delayed_message_exchange 插件来实现定时任务，这种方案较简单。

两种用法我们分别来看。

1. 用插件

1.1. 安装插件

首先我们需要下载 rabbitmq_delayed_message_exchange 插件，这是一个 GitHub 上的开源项目，我们直接下载即可：

注意：必须选择和你安装的RabbitMQ的版本兼容的插件，可以在RabbitMQ的web管理页面查看版本号，我这里选择最新的 3.8.x版。

07-RabbitMQ实现延迟队列 - 图1

在github找到对应的版本号，右键选择复制下载的地址，然后使用wget命令进行下载

07-RabbitMQ实现延迟队列 - 图2

然后执行如下命令下载

  1. $ wget https://github.com/rabbitmq/rabbitmq-delayed-message-exchange/releases/download/v3.8.0/rabbitmq_delayed_message_exchange-3.8.0.ez

下载完成后在命令行执行如下命令将下载文件拷贝到 Docker 容器中去：

  1. $ docker cp ./rabbitmq_delayed_message_exchange-3.8.0.ez rabbit01:/plugins

这里第一个参数是宿主机上的文件地址，第二个参数是拷贝到容器的位置。

接下来再执行如下命令进入到 RabbitMQ 容器中：

  1. $ docker exec -it rabbit01 /bin/bash

进入到容器之后，执行如下命令启用插件：

  1. $ rabbitmq-plugins enable rabbitmq_delayed_message_exchange

启用成功之后，还可以通过如下命令查看所有安装的插件，看看是否有我们刚刚安装过的插件，如下：

  1. $ rabbitmq-plugins list

07-RabbitMQ实现延迟队列 - 图3

OK，配置完成之后，接下来我们执行 exit 命令退出 RabbitMQ 容器。然后开始编码。

1.2. 消息收发

首先创建一个SpringBoot的工程，工程名叫做message_delay_plugin

07-RabbitMQ实现延迟队列 - 图4

07-RabbitMQ实现延迟队列 - 图5

选择依赖

07-RabbitMQ实现延迟队列 - 图6

在message_delay_plugin工程的application.properties配置文件中做如下的配置：

  1. spring.rabbitmq.host=localhost
  2. spring.rabbitmq.port=5672
  3. spring.rabbitmq.username=guest
  4. spring.rabbitmq.password=guest
  5. spring.rabbitmq.virtual-host=/

在message_delay_plugin工程的org.javaboy.message_delay_plugin.config包下面创建一个消息队列的配置类

  1. @Configuration
  2. public class RabbitConfig {
  3.     public static final String DELAY_MSG_QUEUE_NAME = "delay_msg_queue_name";
  4.     public static final String DELAY_MSG_EXCHANGE_NAME = "delay_msg_exchange_name";
  6.     @Bean
  7.     Queue delayQueue() {
  8.         return new Queue(DELAY_MSG_QUEUE_NAME, true, false, false);
  9.     }
  11.     @Bean
  12.     CustomExchange customExchange() {
  13.         Map<String, Object> args = new HashMap<>();
  14.         //指定交换机的类型
  15.         args.put("x-delayed-type", "direct");
  16.         return new CustomExchange(DELAY_MSG_EXCHANGE_NAME, "x-delayed-message", true, false, args);
  17.     }
  19.     @Bean
  20.     Binding delayedBinding() {
  21.         return BindingBuilder.bind(delayQueue())
  22.                 .to(customExchange())
  23.                 .with(DELAY_MSG_QUEUE_NAME)
  24.                 .noargs();
  25.     }
  27. }

这里主要是交换机的定义有所不同，小伙伴们需要注意。

这里我们使用的交换机是 CustomExchange，这是一个 Spring 中提供的交换机，创建 CustomExchange 时有五个参数，含义分别如下：

  • 交换机名称。
  • 交换机类型，这个地方是固定的。
  • 交换机是否持久化。
  • 如果没有队列绑定到交换机，交换机是否删除。
  • 其他参数。

最后一个 args 参数中，指定了交换机消息分发的类型，这个类型就是大家熟知的 direct、fanout、topic 以及 header 几种，用了哪种类型，将来交换机分发消息就按哪种方式来。

在message_delay_plugin工程的org.javaboy.message_delay_plugin.receiver包下面创建一个消费者

  1. @Component
  2. public class MsgReceiver {
  3.     private static final Logger logger = LoggerFactory.getLogger(MsgReceiver.class);
  5.     @RabbitListener(queues = RabbitConfig.DELAY_MSG_QUEUE_NAME)
  6.     public void handleMsg(String msg) {
  7.         logger.info("handleMsg====== {}", msg);
  8.     }
  9. }

在message_delay_plugin工程的org.javaboy.message_delay_plugin.controller包下面创建一个控制器

  1. @RestController
  2. public class HelloController {
  3.     @Autowired
  4.     RabbitTemplate rabbitTemplate;
  6.     @GetMapping("/send")
  7.     public void hello() {
  8.         Message msg = MessageBuilder.withBody(("hello 江南一点雨" + new Date()).getBytes())
  9.                 //设置消息发送的延迟时间
  10.                 .setHeader("x-delay", 3000).build();
  11.         rabbitTemplate.send(RabbitConfig.DELAY_MSG_EXCHANGE_NAME, RabbitConfig.DELAY_MSG_QUEUE_NAME, msg);
  12.     }
  13. }

在消息头中设置消息的延迟时间。

接下来启动 Spring Boot 项目，在浏览器访问http://localhost:8080/send，然后查看控制台打印的日志，从日志中可以看到消息延迟已经实现了。

07-RabbitMQ实现延迟队列 - 图7

2. DLX实现延迟队列

2.1. 延迟队列实现思路

延迟队列实现的思路也很简单，就是上篇文章我们所说的 DLX（死信交换机）+TTL（消息超时时间）。

我们可以把死信队列就当成延迟队列。

具体来说是这样：

假如一条消息需要延迟 30 分钟执行，我们就设置这条消息的有效期为 30 分钟，同时为这条消息配置死信交换机和死信 routing_key，并且不为这个消息队列设置消费者，那么 30 分钟后，这条消息由于没有被消费者消费而进入死信队列，此时我们有一个消费者就在“蹲点”这个死信队列，消息一进入死信队列，就立马被消费了。

这就是延迟队列的实现思路，是不是很简单？

2.2. 案例

停止RabbitMQ的命令：

  1. #前提是需要进入到docker容器中来，然后才可以执行以下命令
  2. $ rabbitmqctl stop_app

重置RabbitMQ的命令：

  1. #前提是需要进入到docker容器中来，然后才可以执行以下命令
  2. $ rabbitmqctl reset

启动RabbitMQ的命令：

  1. #前提是需要进入到docker容器中来，然后才可以执行以下命令
  2. $ rabbitmqctl start_app

首先创建一个新的SpringBoot工程，工程名叫做message_delay_dlx

07-RabbitMQ实现延迟队列 - 图8

07-RabbitMQ实现延迟队列 - 图9

选择依赖

07-RabbitMQ实现延迟队列 - 图10

在message_delay_dlx工程的application.properties配置文件中做如下配置：

  1. spring.rabbitmq.host=localhost
  2. spring.rabbitmq.port=5672
  3. spring.rabbitmq.username=guest
  4. spring.rabbitmq.password=guest
  5. spring.rabbitmq.virtual-host=/

在message_delay_dlx工程的org.javaboy.message_delay_dlx.config包下面创建配置类配置两个消息队列：一个普通队列，一个死信队列：

  1. @Configuration
  2. public class RabbitConfig {
  3.     public static final String JAVABOY_QUEUE_NAME = "javaboy_queue_name";
  4.     public static final String JAVABOY_EXCHANGE_NAME = "javaboy_exchange_name";
  5.     public static final String DELAY_EXCHANGE_NAME = "delay_exchange_name";
  6.     public static final String DELAY_QUEUE_NAME = "delay_queue_name";
  8.     @Bean
  9.     Binding msgBinding() {
  10.         return BindingBuilder.bind(msgQueue())
  11.                 .to(msgExchange())
  12.                 .with(JAVABOY_QUEUE_NAME);
  13.     }
  15.     @Bean
  16.     DirectExchange msgExchange() {
  17.         return new DirectExchange(JAVABOY_EXCHANGE_NAME, true, false);
  18.     }
  20.     @Bean
  21.     Queue msgQueue() {
  22.         Map<String, Object> args = new HashMap<>();
  23.         args.put("x-message-ttl", 10000);
  24.         args.put("x-dead-letter-exchange", DELAY_EXCHANGE_NAME);
  25.         args.put("x-dead-letter-routing-key", DELAY_QUEUE_NAME);
  26.         return new Queue(JAVABOY_QUEUE_NAME, true, false, false, args);
  27.     }
  29.     @Bean
  30.     Binding dlxBinding() {
  31.         return BindingBuilder.bind(dlxQueue())
  32.                 .to(dlxExchange())
  33.                 .with(DELAY_QUEUE_NAME);
  34.     }
  36.     @Bean
  37.     DirectExchange dlxExchange() {
  38.         return new DirectExchange(DELAY_EXCHANGE_NAME, true, false);
  39.     }
  41.     @Bean
  42.     Queue dlxQueue() {
  43.         return new Queue(DELAY_QUEUE_NAME, true, false, false);
  44.     }
  46. }

这段配置代码虽然略长，不过原理其实简单。

  • 配置可以分为两组，第一组配置死信队列，第二组配置普通队列。每一组都由消息队列、消息交换机以及 Binding 三者组成。
  • 配置消息队列时，为消息队列指定死信队列，不熟悉的小伙伴可以翻一下上篇文章，传送门：RabbitMQ 中的消息会过期吗？。
  • 配置队列中的消息过期时间时，默认的时间单位时毫秒。

在message_delay_dlx工程的org.javaboy.message_delay_dlx.receiver包下面创建消费者

  1. @Component
  2. public class MsgReceiver {
  3.     private static final Logger logger = LoggerFactory.getLogger(MsgReceiver.class);
  5.     @RabbitListener(queues = RabbitConfig.DELAY_QUEUE_NAME)
  6.     public void handleMsg(String msg) {
  7.         logger.info("handleMsg,{}", msg);
  8.     }
  9. }

在message_delay_dlx工程的org.javaboy.message_delay_dlx.controller包下面创建控制器

  1. @RestController
  2. public class HelloController {
  3.     @Autowired
  4.     RabbitTemplate rabbitTemplate;
  6.     @GetMapping("/send")
  7.     public void hello() {
  8.         Message msg = MessageBuilder.withBody(("hello javaboy" + new Date()).getBytes()).build();
  9.         rabbitTemplate.send(RabbitConfig.JAVABOY_EXCHANGE_NAME, RabbitConfig.JAVABOY_QUEUE_NAME, msg);
  10.     }
  11. }

接下来启动 Spring Boot 项目，在浏览器访问http://localhost:8080/send，然后查看控制台打印的日志，从日志中可以看到消息延迟已经实现了。

07-RabbitMQ实现延迟队列 - 图11