[[429119]]

在很多場(chǎng)景下，我們都有延遲隊(duì)列的需求，然而默認(rèn)情況下 RabbitMQ 并未提供延遲隊(duì)列相關(guān)的功能，不過，在前面的文章中，松哥和大家分享了 RabbitMQ 中的死信隊(duì)列，結(jié)合死信隊(duì)列以及消息過期時(shí)間的設(shè)定，我們也可以實(shí)現(xiàn)延遲隊(duì)列的功能，今天我們就一起來看下。

RabbitMQ 其實(shí)有提供專門的延遲隊(duì)列插件，這個(gè)咱們有空了再聊。

1. 什么時(shí)候需要延遲隊(duì)列?

先來說說什么時(shí)候我們需要延遲隊(duì)列。

舉幾個(gè)簡(jiǎn)單的例子。

在電商項(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ì)列。

可能有小伙伴說至于嘛，直接搞個(gè)定時(shí)任務(wù)不更方便?如果項(xiàng)目中只有一個(gè)這樣的延遲隊(duì)列的場(chǎng)景，那么搞個(gè)定時(shí)任務(wù)似乎也可以，但是如果項(xiàng)目中有很多這樣的場(chǎng)景，那么定時(shí)任務(wù)很明顯就不是最佳方案了，我們可以通過延遲隊(duì)列來實(shí)現(xiàn)一個(gè)通用的解決方案。

2. 延遲隊(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)單?

3. 案例

接下來松哥通過一個(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)者中被打印出來。

4. 小結(jié)

好啦，這就是我們用 RabbitMQ 做延遲隊(duì)列的思路～