RabbitMQ 实现延迟队列的两种方式
文章目录1. 用插件1.1 安装插件1.2 消息收发2. DLX 实现延迟队列2.1 延迟队列实现思路2.2 案例3. 小结定时任务各种各样,常见的定时任务例如日志备份,我们可能在每天凌晨 3 点去备份,这种固定时间的定时任务我们一般采用 cron 表达式就能轻松的实现,还有一些比较特殊的定时任务,向大家看电影中的定时炸弹,3分钟后爆炸,这种定时任务就不太好用 cron 去描述,因为开始时间不确定
定时任务各种各样,常见的定时任务例如日志备份,我们可能在每天凌晨 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
命令退出 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<String, Object> 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 时有五个参数,含义分别如下:
- 交换机名称。
- 交换机类型,这个地方是固定的。
- 交换机是否持久化。
- 如果没有队列绑定到交换机,交换机是否删除。
- 其他参数。
最后一个 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=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<String, Object> 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 三者组成。
- 配置消息队列时,为消息队列指定死信队列,不熟悉的小伙伴可以翻一下上篇文章,传送门:RabbitMQ 中的消息会过期吗?。
- 配置队列中的消息过期时间时,默认的时间单位时毫秒。
接下来我们为死信队列配置一个消费者,如下:
@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 做延迟队列的两种思路~感兴趣的小伙伴可以试试哦~
![Logo](https://devpress.csdnimg.cn/79de2bf0b7994defa4242ef90d5513fa.jpg)
开放原子开发者工作坊旨在鼓励更多人参与开源活动,与志同道合的开发者们相互交流开发经验、分享开发心得、获取前沿技术趋势。工作坊有多种形式的开发者活动,如meetup、训练营等,主打技术交流,干货满满,真诚地邀请各位开发者共同参与!
更多推荐
所有评论(0)