RabbitMQ 實現延遲隊列的兩種方式詳解

Posted on 2021-12-09 by WalkonNet

定時任務各種各樣，常見的定時任務例如日志備份，我們可能在每天凌晨 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 三者組成。
配置消息隊列時，為消息隊列指定死信隊列。
配置隊列中的消息過期時間時，默認的時間單位時毫秒。

接下來我們為死信隊列配置一個消費者，如下：

@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 做延遲隊列的兩種思路～感興趣的小夥伴可以試試哦～

到此這篇關於RabbitMQ 實現延遲隊列的兩種方式詳解的文章就介紹到這瞭,更多相關RabbitMQ 延遲隊列內容請搜索WalkonNet以前的文章或繼續瀏覽下面的相關文章希望大傢以後多多支持WalkonNet！

RabbitMQ 實現延遲隊列的兩種方式詳解

目錄

1. 用插件

1.1 安裝插件

1.2 消息收發

2. DLX 實現延遲隊列

2.1 延遲隊列實現思路

2.2 案例

3. 小結

推薦閱讀：

發佈留言取消回覆

近期文章

目錄

1. 用插件

1.1 安裝插件

1.2 消息收發

2. DLX 實現延遲隊列

2.1 延遲隊列實現思路

2.2 案例

3. 小結

推薦閱讀：

發佈留言 取消回覆

近期文章

標籤

發佈留言取消回覆