RabbitMQ 實(shí)現(xiàn)延遲隊(duì)列的兩種方式！
定時(shí)任務(wù)各種各樣，常見的定時(shí)任務(wù)例如日志備份，我們可能在每天凌晨 3 點(diǎn)去備份，這種固定時(shí)間的定時(shí)任務(wù)我們一般采用 cron 表達(dá)式就能輕松的實(shí)現(xiàn)，還有一些比較特殊的定時(shí)任務(wù)，向大家看電影中的定時(shí)炸彈，3分鐘后爆炸，這種定時(shí)任務(wù)就不太好用 cron 去描述，因?yàn)殚_始時(shí)間不確定，我們開發(fā)中有的時(shí)候也會(huì)遇到類似的需求，例如：
- 在電商項(xiàng)目中，當(dāng)我們下單之后，一般需要 20 分鐘之內(nèi)或者 30 分鐘之內(nèi)付款，否則訂單就會(huì)進(jìn)入異常處理邏輯中，被取消，那么進(jìn)入到異常處理邏輯中，就可以當(dāng)成是一個(gè)延遲隊(duì)列。
- 我買了一個(gè)智能砂鍋，可以用來煮粥，上班前把素材都放到鍋里，然后設(shè)置幾點(diǎn)幾分開始煮粥，這樣下班后就可以喝到香噴噴的粥了，那么這個(gè)煮粥的指令也可以看成是一個(gè)延遲任務(wù)，放到一個(gè)延遲隊(duì)列中，時(shí)間到了再執(zhí)行。
- 公司的會(huì)議預(yù)定系統(tǒng)，在會(huì)議預(yù)定成功后，會(huì)在會(huì)議開始前半小時(shí)通知所有預(yù)定該會(huì)議的用戶。
- 安全工單超過 24 小時(shí)未處理，則自動(dòng)拉企業(yè)微信群提醒相關(guān)責(zé)任人。
- 用戶下單外賣以后，距離超時(shí)時(shí)間還有 10 分鐘時(shí)提醒外賣小哥即將超時(shí)。
- ...
很多場(chǎng)景下我們都需要延遲隊(duì)列。
本文以 RabbitMQ 為例來和大家聊一聊延遲隊(duì)列的玩法。
整體上來說，在 RabbitMQ 上實(shí)現(xiàn)定時(shí)任務(wù)有兩種方式：
利用 RabbitMQ 自帶的消息過期和私信隊(duì)列機(jī)制，實(shí)現(xiàn)定時(shí)任務(wù)。
使用 RabbitMQ 的 rabbitmq_delayed_message_exchange 插件來實(shí)現(xiàn)定時(shí)任務(wù)，這種方案較簡(jiǎn)單。
兩種用法我們分別來看。
1. 用插件
1.1 安裝插件
首先我們需要下載 rabbitmq_delayed_message_exchange 插件，這是一個(gè) GitHub 上的開源項(xiàng)目，我們直接下載即可：
https://github.com/rabbitmq/rabbitmq-delayed-message-exchange/releases
選擇適合自己的版本，我這里選擇最新的 3.9.0 版。
下載完成后在命令行執(zhí)行如下命令將下載文件拷貝到 Docker 容器中去：
- docker cp ./rabbitmq_delayed_message_exchange-3.9.0.ez some-rabbit:/plugins
這里第一個(gè)參數(shù)是宿主機(jī)上的文件地址，第二個(gè)參數(shù)是拷貝到容器的位置。
接下來再執(zhí)行如下命令進(jìn)入到 RabbitMQ 容器中：
- docker exec -it some-rabbit /bin/bash
進(jìn)入到容器之后，執(zhí)行如下命令啟用插件：
- rabbitmq-plugins enable rabbitmq_delayed_message_exchange
啟用成功之后，還可以通過如下命令查看所有安裝的插件，看看是否有我們剛剛安裝過的插件，如下：
- rabbitmq-plugins list
命令的完整執(zhí)行過程如下圖：
OK，配置完成之后，接下來我們執(zhí)行 exit 命令退出 RabbitMQ 容器。然后開始編碼。
1.2 消息收發(fā)
接下來開始消息收發(fā)。
首先我們創(chuàng)建一個(gè) Spring Boot 項(xiàng)目，引入 Web 和 RabbitMQ 依賴，如下：
項(xiàng)目創(chuàng)建成功后，在 application.properties 中配置 RabbitMQ 的基本信息，如下：
- spring.rabbitmq.host=localhost
- spring.rabbitmq.password=guest
- spring.rabbitmq.username=guest
- spring.rabbitmq.virtual-host=/
接下來提供一個(gè) 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();
- }
- }
這里主要是交換機(jī)的定義有所不同，小伙伴們需要注意。
這里我們使用的交換機(jī)是 CustomExchange，這是一個(gè) Spring 中提供的交換機(jī)，創(chuàng)建 CustomExchange 時(shí)有五個(gè)參數(shù)，含義分別如下：
- 交換機(jī)名稱。
- 交換機(jī)類型，這個(gè)地方是固定的。
- 交換機(jī)是否持久化。
- 如果沒有隊(duì)列綁定到交換機(jī)，交換機(jī)是否刪除。
- 其他參數(shù)。
最后一個(gè) args 參數(shù)中，指定了交換機(jī)消息分發(fā)的類型，這個(gè)類型就是大家熟知的 direct、fanout、topic 以及 header 幾種，用了哪種類型，將來交換機(jī)分發(fā)消息就按哪種方式來。
接下來我們?cè)賱?chuàng)建一個(gè)消息消費(fèi)者：
- @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);
- }
- }
打印一下消息內(nèi)容即可。
接下來再寫一個(gè)單元測(cè)試方法來發(fā)送消息：
- @SpringBootTest
- class MqDelayedMsgDemoApplicationTests {
- @Autowired
- RabbitTemplate rabbitTemplate;
- @Test
- void contextLoads() throws UnsupportedEncodingException {
- Message msg = MessageBuilder.withBody(("hello 江南一點(diǎn)雨"+new Date()).getBytes("UTF-8")).setHeader("x-delay", 3000).build();
- rabbitTemplate.convertAndSend(RabbitConfig.EXCHANGE_NAME, RabbitConfig.QUEUE_NAME, msg);
- }
- }
在消息頭中設(shè)置消息的延遲時(shí)間。
好啦，接下來啟動(dòng) Spring Boot 項(xiàng)目，然后運(yùn)行單元測(cè)試方法發(fā)送消息，最終的控制臺(tái)打印日志如下：
從日志中可以看到消息延遲已經(jīng)實(shí)現(xiàn)了。
2. DLX 實(shí)現(xiàn)延遲隊(duì)列
2.1 延遲隊(duì)列實(shí)現(xiàn)思路
延遲隊(duì)列實(shí)現(xiàn)的思路也很簡(jiǎn)單，就是上篇文章我們所說的 DLX(死信交換機(jī))+TTL(消息超時(shí)時(shí)間)。
我們可以把死信隊(duì)列就當(dāng)成延遲隊(duì)列。
具體來說是這樣：
假如一條消息需要延遲 30 分鐘執(zhí)行，我們就設(shè)置這條消息的有效期為 30 分鐘，同時(shí)為這條消息配置死信交換機(jī)和死信 routing_key，并且不為這個(gè)消息隊(duì)列設(shè)置消費(fèi)者，那么 30 分鐘后，這條消息由于沒有被消費(fèi)者消費(fèi)而進(jìn)入死信隊(duì)列，此時(shí)我們有一個(gè)消費(fèi)者就在“蹲點(diǎn)”這個(gè)死信隊(duì)列，消息一進(jìn)入死信隊(duì)列，就立馬被消費(fèi)了。
這就是延遲隊(duì)列的實(shí)現(xiàn)思路，是不是很簡(jiǎn)單?
2.2 案例
接下來松哥通過一個(gè)簡(jiǎn)單的案例，來和大家演示一下延遲隊(duì)列的具體實(shí)現(xiàn)。
首先準(zhǔn)備好一個(gè)啟動(dòng)的 RabbitMQ。
然后我們創(chuàng)建一個(gè) Spring Boot 項(xiàng)目，引入 RabbitMQ 依賴：
然后在 application.properties 中配置一下 RabbitMQ 的基本連接信息：
- spring.rabbitmq.host=localhost
- spring.rabbitmq.username=guest
- spring.rabbitmq.password=guest
- spring.rabbitmq.port=5672
接下來我們來配置兩個(gè)消息隊(duì)列：一個(gè)普通隊(duì)列，一個(gè)死信隊(duì)列：
- @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";
- /**
- * 死信隊(duì)列
- * @return
- */
- @Bean
- Queue dlxQueue() {
- return new Queue(DLX_QUEUE_NAME, true, false, false);
- }
- /**
- * 死信交換機(jī)
- * @return
- */
- @Bean
- DirectExchange dlxExchange() {
- return new DirectExchange(DLX_EXCHANGE_NAME, true, false);
- }
- /**
- * 綁定死信隊(duì)列和死信交換機(jī)
- * @return
- */
- @Bean
- Binding dlxBinding() {
- return BindingBuilder.bind(dlxQueue()).to(dlxExchange())
- .with(DLX_ROUTING_KEY);
- }
- /**
- * 普通消息隊(duì)列
- * @return
- */
- @Bean
- Queue javaboyQueue() {
- Map<String, Object> args = new HashMap<>();
- //設(shè)置消息過期時(shí)間
- args.put("x-message-ttl", 1000*10);
- //設(shè)置死信交換機(jī)
- args.put("x-dead-letter-exchange", DLX_EXCHANGE_NAME);
- //設(shè)置死信 routing_key
- args.put("x-dead-letter-routing-key", DLX_ROUTING_KEY);
- return new Queue(JAVABOY_QUEUE_NAME, true, false, false, args);
- }
- /**
- * 普通交換機(jī)
- * @return
- */
- @Bean
- DirectExchange javaboyExchange() {
- return new DirectExchange(JAVABOY_EXCHANGE_NAME, true, false);
- }
- /**
- * 綁定普通隊(duì)列和與之對(duì)應(yīng)的交換機(jī)
- * @return
- */
- @Bean
- Binding javaboyBinding() {
- return BindingBuilder.bind(javaboyQueue())
- .to(javaboyExchange())
- .with(JAVABOY_ROUTING_KEY);
- }
- }
這段配置代碼雖然略長(zhǎng)，不過原理其實(shí)簡(jiǎn)單。
配置可以分為兩組，第一組配置死信隊(duì)列，第二組配置普通隊(duì)列。每一組都由消息隊(duì)列、消息交換機(jī)以及 Binding 三者組成。
配置消息隊(duì)列時(shí)，為消息隊(duì)列指定死信隊(duì)列，不熟悉的小伙伴可以翻一下上篇文章，傳送門：RabbitMQ 中的消息會(huì)過期嗎?。
配置隊(duì)列中的消息過期時(shí)間時(shí)，默認(rèn)的時(shí)間單位時(shí)毫秒。
接下來我們?yōu)樗佬抨?duì)列配置一個(gè)消費(fèi)者，如下：
- @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);
- }
- }
收到消息后就將之打印出來。
這就完事了。
啟動(dòng)項(xiàng)目。
最后我們?cè)趩卧獪y(cè)試中發(fā)送一條消息：
- @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!");
- }
- }
這個(gè)就沒啥好說的了，就是普通的消息發(fā)送，10 秒之后這條消息會(huì)在死信隊(duì)列的消費(fèi)者中被打印出來。
3. 小結(jié)
好啦，這就是我們用 RabbitMQ 做延遲隊(duì)列的兩種思路～感興趣的小伙伴可以試試哦～
本文轉(zhuǎn)載自微信公眾號(hào)「江南一點(diǎn)雨」，可以通過以下二維碼關(guān)注。轉(zhuǎn)載本文請(qǐng)聯(lián)系江南一點(diǎn)雨公眾號(hào)。
