RabbitMQ 實作延遲佇列的兩種方式

• 在電商專案中，當我們下單之后，一般需要 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 做延遲佇列的兩種思路～感興趣的小伙伴可以試試哦～