在很多場景下,我們都有延遲隊列的需求,然而默認情況下 RabbitMQ 并未提供延遲隊列相關的功能,不過,在前面的文章中,松哥和大家分享了 RabbitMQ 中的死信隊列,結合死信隊列以及消息過期時間的設定,我們也可以實現延遲隊列的功能,今天我們就一起來看下。
RabbitMQ 其實有提供專門的延遲隊列插件,這個咱們有空了再聊。
1. 什么時候需要延遲隊列?
先來說說什么時候我們需要延遲隊列。
舉幾個簡單的例子。
在電商項目中,當我們下單之后,一般需要 20 分鐘之內或者 30 分鐘之內付款,否則訂單就會進入異常處理邏輯中,被取消,那么進入到異常處理邏輯中,就可以當成是一個延遲隊列。
我買了一個智能砂鍋,可以用來煮粥,上班前把素材都放到鍋里,然后設置幾點幾分開始煮粥,這樣下班后就可以喝到香噴噴的粥了,那么這個煮粥的指令也可以看成是一個延遲任務,放到一個延遲隊列中,時間到了再執行。
公司的會議預定系統,在會議預定成功后,會在會議開始前半小時通知所有預定該會議的用戶。
安全工單超過 24 小時未處理,則自動拉企業微信群提醒相關責任人。
用戶下單外賣以后,距離超時時間還有 10 分鐘時提醒外賣小哥即將超時。
...
很多場景下我們都需要延遲隊列。
可能有小伙伴說至于嘛,直接搞個定時任務不更方便?如果項目中只有一個這樣的延遲隊列的場景,那么搞個定時任務似乎也可以,但是如果項目中有很多這樣的場景,那么定時任務很明顯就不是最佳方案了,我們可以通過延遲隊列來實現一個通用的解決方案。
2. 延遲隊列實現思路
延遲隊列實現的思路也很簡單,就是上篇文章我們所說的 DLX(死信交換機)+TTL(消息超時時間)。
我們可以把死信隊列就當成延遲隊列。
具體來說是這樣:
假如一條消息需要延遲 30 分鐘執行,我們就設置這條消息的有效期為 30 分鐘,同時為這條消息配置死信交換機和死信 routing_key,并且不為這個消息隊列設置消費者,那么 30 分鐘后,這條消息由于沒有被消費者消費而進入死信隊列,此時我們有一個消費者就在“蹲點”這個死信隊列,消息一進入死信隊列,就立馬被消費了。
這就是延遲隊列的實現思路,是不是很簡單?
3. 案例
接下來松哥通過一個簡單的案例,來和大家演示一下延遲隊列的具體實現。
首先準備好一個啟動的 RabbitMQ。
然后我們創建一個 Spring Boot 項目,引入 RabbitMQ 依賴:
然后在 application.properties 中配置一下 RabbitMQ 的基本連接信息:
- spring.rabbitmq.host=localhost
- spring.rabbitmq.username=guest
- spring.rabbitmq.password=guest
- spring.rabbitmq.port=5672
接下來我們來配置兩個消息隊列:一個普通隊列,一個死信隊列:
- @Configuration
- publicclassQueueConfig{
- publicstaticfinalStringJAVABOY_QUEUE_NAME="javaboy_queue_name";
- publicstaticfinalStringJAVABOY_EXCHANGE_NAME="javaboy_exchange_name";
- publicstaticfinalStringJAVABOY_ROUTING_KEY="javaboy_routing_key";
- publicstaticfinalStringDLX_QUEUE_NAME="dlx_queue_name";
- publicstaticfinalStringDLX_EXCHANGE_NAME="dlx_exchange_name";
- publicstaticfinalStringDLX_ROUTING_KEY="dlx_routing_key";
- /**
- *死信隊列
- *@return
- */
- @Bean
- QueuedlxQueue(){
- returnnewQueue(DLX_QUEUE_NAME,true,false,false);
- }
- /**
- *死信交換機
- *@return
- */
- @Bean
- DirectExchangedlxExchange(){
- returnnewDirectExchange(DLX_EXCHANGE_NAME,true,false);
- }
- /**
- *綁定死信隊列和死信交換機
- *@return
- */
- @Bean
- BindingdlxBinding(){
- returnBindingBuilder.bind(dlxQueue()).to(dlxExchange())
- .with(DLX_ROUTING_KEY);
- }
- /**
- *普通消息隊列
- *@return
- */
- @Bean
- QueuejavaboyQueue(){
-
Map
args=newHashMap<>(); - //設置消息過期時間
- 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);
- returnnewQueue(JAVABOY_QUEUE_NAME,true,false,false,args);
- }
- /**
- *普通交換機
- *@return
- */
- @Bean
- DirectExchangejavaboyExchange(){
- returnnewDirectExchange(JAVABOY_EXCHANGE_NAME,true,false);
- }
- /**
- *綁定普通隊列和與之對應的交換機
- *@return
- */
- @Bean
- BindingjavaboyBinding(){
- returnBindingBuilder.bind(javaboyQueue())
- .to(javaboyExchange())
- .with(JAVABOY_ROUTING_KEY);
- }
- }
這段配置代碼雖然略長,不過原理其實簡單。
配置可以分為兩組,第一組配置死信隊列,第二組配置普通隊列。每一組都由消息隊列、消息交換機以及 Binding 三者組成。
配置消息隊列時,為消息隊列指定死信隊列,不熟悉的小伙伴可以翻一下上篇文章,傳送門:RabbitMQ 中的消息會過期嗎?。
配置隊列中的消息過期時間時,默認的時間單位時毫秒。
接下來我們為死信隊列配置一個消費者,如下:
- @Component
- publicclassDlxConsumer{
- privatestaticfinalLoggerlogger=LoggerFactory.getLogger(DlxConsumer.class);
- @RabbitListener(queues=QueueConfig.DLX_QUEUE_NAME)
- publicvoidhandle(Stringmsg){
- logger.info(msg);
- }
- }
收到消息后就將之打印出來。
這就完事了。
啟動項目。
最后我們在單元測試中發送一條消息:
- @SpringBootTest
- classDelayQueueApplicationTests{
- @Autowired
- RabbitTemplaterabbitTemplate;
- @Test
- voidcontextLoads(){
- System.out.println(newDate());
- rabbitTemplate.convertAndSend(QueueConfig.JAVABOY_EXCHANGE_NAME,QueueConfig.JAVABOY_ROUTING_KEY,"hellojavaboy!");
- }
- }
這個就沒啥好說的了,就是普通的消息發送,10 秒之后這條消息會在死信隊列的消費者中被打印出來。
4. 小結
好啦,這就是我們用 RabbitMQ 做延遲隊列的思路~
原文鏈接:https://mp.weixin.qq.com/s/x53MSd7VNfA6OwMpFA-Bsg