RabbitMQ 实现延迟队列的两种方式详解

RabbitMQ 实现延迟队列的两种方式详解

目录1. 用插件1.1 安装插件1.2 消息收发2. DLX 实现延迟队列2.1 延迟队列实现思路2.2 案例3. 小结

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

在电商项目中，当我们下单之后，一般需要 20 分钟之内或者 30 分钟之内付款，否则订单就会进入异常处理逻辑中，被取消，那么进入到异常处理逻辑中，就可以当成是一个延迟队列。

我买了一个智能砂锅，可以用来煮粥，上班前把素材都放到锅里，然后设置几点几分开始煮粥，这样下班后就可以喝到香喷喷的粥了，那么这个煮粥的指令也可以看成是一个延迟任务，放到一个延迟队列中，时间到了再执行。

公司的会议预定系统，在会议预定成功后，会在会议开始前半小时通知所有预定该会议的用户。

安全工单超过 24 小时未处理，则自动拉企业微信群提醒相关责任人。

用户下单外卖以后，距离超时时间还有 10 分钟时提醒外卖小哥即将超时。

…

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

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

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

利用 RabbitMQ 自带的消息过期和私信队列机制，实现定时任务。

使用 RabbitMQ 的 rabbitmq_delayed_message_exchange 插件来实现定时任务，这种方案较简单。

两种用法我们分别来看。

1. 用插件

1.1 安装插件

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

https://github.com/rabbitmq/rabbitmq-delayed-message-exchange/releases

选择适合自己的版本，我这里选择最新的 3.9.0 版。

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

docker cp ./rabbitmq_delayed_message_exchange-3.9.0.ez some-rabbit:/plugins

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

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

docker exec -it some-rabbit /bin/bash

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

rabbitmq-plugins enable rabbitmq_delayed_message_exchange

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

rabbitmq-plugins list

命令的完整执行过程如下图：

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

1.2 消息收发

接下来开始消息收发。

首先我们创建一个 Spring Boot 项目，引入 Web 和 RabbitMQ 依赖，如下：

项目创建成功后，在 application.properties 中配置 RabbitMQ 的基本信息，如下：

spring.rabbitmq.host=localhost

spring.rabbitmq.password=guest

spring.rabbitmq.username=guest

spring.rabbitmq.virtual-host=/

接下来提供一个 RabbitMQ 的配置类：

@Configuration

public class RabbitConfig {

public static final String QUEUE_NAME = "javaboy_delay_queue";

public static final String EXCHANGE_NAME = "javaboy_delay_exchange";

public static final String EXCHANGE_TYPE = "x-delayed-message";

@Bean

Queue queue() {

return new Queue(QUEUE_NAME, true, false, false);

}

@Bean

CustomExchange customExchange() {

Map args = new HashMap<>();

args.put("x-delayed-type", "direct");

return new CustomExchange(EXCHANGE_NAME, EXCHANGE_TYPE, true, false,args);

}

@Bean

Binding binding() {

return BindingBuilder.bind(queue())

.to(customExchange()).with(QUEUE_NAME).noargs();

}

}

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

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

交换机名称。

交换机类型，这个地方是固定的。

交换机是否持久化。

如果没有队列绑定到交换EHEJY机，交换机是否删除。

其他参数。

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

接下来我们再创建一个消息消费者：

@Component

public class MsgReceiver {

private static final Logger logger = LoggerFactory.getLogger(MsgReceiver.class);

@RabbitListener(queues = RabbitConfig.QUEUE_NAME)

public void handleMsg(String msg) {

logger.info("handleMsg,{}",msg);

}

}

打印一下消息内容即可。

接下来再写一个单元测试方法来发送消息：

@SpringBootTest

class MqDelayedMsgDemoApplicationTests {

@Autowired

RabbitTemplate rabbitTemplate;

@Test

void contextLoads() throws UnsupportedEncodingException {

Message msg = MessageBuilder.withBody(("hello 江南一点雨"+new Date()).getBytes("UTF-8")).setHeader("x-delay", 3000).build();

rabbitTemplate.convertAndSend(RabbitConfig.EXCHANGE_NAME, RabbitConfig.QUEUE_NAME, msg);

}

}

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

好啦，接下来启动 Spring Boot 项目，然后运行单元测试方法发送消息，最终的控制台打印日志如下：

从日志中可以看到消息延迟已经实现了。

2. DLX 实现延迟队列

2.1 延迟队列实现思路

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

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

具体来说是这样：

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

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

2.2 案例

接下来松哥通过一个简单的案例，来和大家演示一下延迟队列的具体实现。

首先准备好一个启动的 RabbitMQ。

然后我们创建一个 Spring Boot 项目，引入 RabbitMQ 依赖：

然后在 application.properties 中配置一下 RabbitMQ 的基本连接信息：

spring.rabbitmq.host=localhost

spring.rabbitmq.username=http://guest

spring.rabbitmq.password=guest

spring.rabbitmq.port=5672

接下来我们来配置两个消息队列：一个普通队列，一个死信队列：

@Configuration

public class QueueConfig {

public static final String JAVABOY_QUEUE_NAME = "javaboy_queue_name";

public static final String JAVABOY_EXCHANGE_NAME = "javaboy_exchange_name";

public static final String JAVABOY_ROUTING_KEY = "javaboy_routing_key";

public static final String DLX_QUEUE_NAME = "dlx_queue_name";

public static final String DLX_EXCHANGE_NAME = "dlx_exchange_name";

public static final String DLX_ROUTING_KEY = "dlx_routing_key";

/**

* 死信队列

* @return

*/

@Bean

Queue dlxQueue() {

return new Queue(DLX_QUEUE_NAME, true, false, false);

}

/**

* 死信交换机

* @return

*/

@Bean

DirectExchange dlxExchange() {

return new DirectExchange(DLX_EXCHANGE_NAME, true, false);

}

/**

* 绑定死信队列和死信交换机

* @return

*/

@Bean

Binding dlxBinding() {

return BindingBuilder.bind(dlxQueue()).to(dlxExchange())

.with(DLX_ROUTING_KEY);

}

/**

* 普通消息队列

* @return

*/

@Bean

Queue javaboyQueue() {

Map args = new HashMap<>();

//设置消息过期时间

args.put("x-message-ttl", 1000*10);

//设置死信交换机

args.put("x-dead-letter-exchange", DLX_EXCHANGE_NAME);

//设置死信 routing_key

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

return new Queue(JAVABOY_QUEUE_NAME, true, false, false, args);

}

/**

* 普通交换机

* @return

*/

@Bean

DirectExchange javaboyExchange() {

return new DirectExchange(JAVABOY_EXCHANGE_NAME, true, false);

}

/**

* 绑定普通队列和与之对应的交换机

* @return

*/

@Bean

Binding javaboyBinding() {

return BindingBuilder.bind(javaboyQueue())

.to(javaboyExchange())

.with(JAVABOY_ROUTING_KEY);

}

}

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

配置可以分为两组，第一组配置死信队列，第二组配置普通队列。每一组都由消息队列、消息交换机以及 Binding 三者组成。

配置消息队列时，为消息队列指定死信队列。

配置队列中的消息过期时间时，默认的时间单位时毫秒。

接下来我们为死信队列配置一个消费者，如下：

@Component

public class DlxConsumer {

private static final Logger logger = LoggerFactory.getLogger(DlxConsumer.class);

@RabbitListener(queues = QueueConfig.DLX_QUEUE_NAME)

public void handle(String msg) {

logger.info(msg);

}

}

收到消息后就将之打印出来。

这就完事了。

启动项目。

最后我们在单元测试中发送一条消息：

@SpringBootTest

class DelayQueueApplicationTests {

@Autowired

RabbitTemplate rabbitTemplate;

@Test

void contextLoads() {

System.out.println(new Date());

rabbitTemplate.convertAndSend(QueueConfig.JAVABOY_EXCHANGE_NAME, QueueConfig.JAVABOY_ROUTING_KEY, "hello javaboy!");

}

}

这个就没啥好说的了，就是普通的消息发送，10 秒之后这条消息会在死信队列的消费者中被打印出来。

3. 小结

好啦，这就是我们用 RabbitMQ 做延迟队列的两种思路～感兴趣的小伙伴可以试试哦～

版权声明：本文内容由网络用户投稿，版权归原作者所有，本站不拥有其著作权，亦不承担相应法律责任。如果您发现本站中有涉嫌抄袭或描述失实的内容，请联系我们jiasou666@gmail.com 处理，核实后本网站将在24小时内删除侵权内容。