RabbitMQ 默認(rèn)并不支持“真正意義上的延遲隊(duì)列”。實(shí)現(xiàn)延時(shí)消息最常用的方式就是安裝 rabbitmq_delayed_message_exchange 插件。
本文將從插件安裝、啟用、配置、使用全流程帶你掌握 RabbitMQ 延時(shí)隊(duì)列的正確姿勢(shì)。

?? 一、什么是 RabbitMQ 延時(shí)隊(duì)列？

延時(shí)隊(duì)列的核心功能就是：讓消息延遲一定時(shí)間后再投遞到消費(fèi)者。

典型應(yīng)用場(chǎng)景包括：

• 訂單支付超時(shí)取消
• 秒殺未支付自動(dòng)釋放庫(kù)存
• 用戶注冊(cè)后延時(shí)發(fā)送歡迎郵件
• 自動(dòng)關(guān)閉未操作的工單等

?? 二、安裝前準(zhǔn)備

? RabbitMQ 環(huán)境要求

• RabbitMQ ≥ 3.6.x
• Erlang ≥ 19.x（越新越好）
• 插件版本需與 RabbitMQ 版本匹配（注意版本兼容）

Docker 搭建 RabbitMQ

?? 三、安裝延時(shí)隊(duì)列插件

?? 插件名稱：

rabbitmq_delayed_message_exchange

1?? 下載插件

你可以從 GitHub 或 RabbitMQ 官網(wǎng)下載預(yù)編譯的 .ez 文件。

?? 插件發(fā)布地址：

• https://www.rabbitmq.com/community-plugins.html

• GitHub 地址：https://github.com/rabbitmq/rabbitmq-delayed-message-exchange

進(jìn)入web端后左上角有顯示當(dāng)前安裝的RabbitMQ版本

需要保證RabbitMQ插件的大版本與RabbitMQ保持一致，否則會(huì)啟動(dòng)失敗

例如（RabbitMQ 4.0.*）：

wget https://github.com/rabbitmq/rabbitmq-delayed-message-exchange/releases/download/v4.0.7/rabbitmq_delayed_message_exchange-v4.0.7.ez

2?? 復(fù)制插件到插件目錄

將插件拷貝到容器內(nèi)plugins目錄下

docker cp /root/rabbitmq_delayed_message_exchange-v4.0.7.ez rabbitmq:/plugins

將 .ez 文件復(fù)制進(jìn)去：

// 進(jìn)入容器 我這里使用容器名字 也可以用容器id進(jìn)入
docker exec -it rabbitmq /bin/bash
// 移動(dòng)到plugins目錄下
cd plugins
// 查看是否上傳成功
ls

3?? 啟用插件

rabbitmq-plugins enable rabbitmq_delayed_message_exchange

輸出類似：

檢查插件是否存在于鏡像中
容器內(nèi)執(zhí)行：

rabbitmq-plugins list

這表明：

? 插件 rabbitmq_delayed_message_exchange 已經(jīng)啟用并正在運(yùn)行。

4?? 重啟 RabbitMQ 容器（建議）

docker restart rabbitmq

5?? 驗(yàn)證插件是否啟用成功

容器啟動(dòng)成功之后，登錄RabbitMQ的管理界面（ip:15672 訪問web界面），找到ExchangesTab頁(yè)。點(diǎn)擊Add a new exchange，在Type里面查看是否有x-delayed-message選項(xiàng)，如果存在就代表插件安裝成功。

6?? 配置容器重啟時(shí)自動(dòng)加載插件

為了確保 RabbitMQ 容器每次重啟時(shí)自動(dòng)啟用指定插件，需要在容器內(nèi) /etc/rabbitmq/enabled_plugins 文件中配置插件列表。

進(jìn)入正在運(yùn)行的容器：

docker exec -it <容器名稱或ID> /bin/bash

編輯或創(chuàng)建插件配置文件 /etc/rabbitmq/enabled_plugins，寫入如下內(nèi)容：

[rabbitmq_management, rabbitmq_delayed_message_exchange].

保存文件并退出容器。

重新啟動(dòng)容器，確認(rèn)插件自動(dòng)啟用：

docker restart <容器名稱或ID>

通過(guò) RabbitMQ 管理界面或命令行確認(rèn)插件狀態(tài)：

docker exec <容器名稱或ID> rabbitmq-plugins list

備注：

• 確保掛載的配置目錄（包含 enabled_plugins 文件）權(quán)限正確，容器內(nèi) RabbitMQ 進(jìn)程能正常讀取。
• enabled_plugins 文件必須以 Erlang 列表格式并以句點(diǎn) . 結(jié)尾。
• 也可通過(guò)環(huán)境變量或啟動(dòng)命令方式啟用插件，但寫入 enabled_plugins 文件更持久、可靠。

?? 四、使用延時(shí)隊(duì)列功能（Java 示例）

延時(shí)消息的核心是：通過(guò)一個(gè)特殊的 Exchange 類型 x-delayed-message + 自定義 header 設(shè)置延遲時(shí)間。

?? 創(chuàng)建延時(shí)交換機(jī)與隊(duì)列

/**
     * 聲明延遲隊(duì)列（Queue 本身無(wú) TTL，延遲由消息屬性 x-delay 控制）
     * 注意：需要安裝 RabbitMQ 的延遲消息插件（rabbitmq_delayed_message_exchange）
     */
    @Bean
    public Queue delayQueue() {
        return QueueBuilder.durable(getRetry().getQueue()).build();
    }
    /**
     * 聲明延遲交換機(jī)：必須使用插件類型 x-delayed-message，并聲明延遲基礎(chǔ)類型為 direct
     */
    @Bean
    public CustomExchange delayExchange() {
        Map<String, Object> args = new HashMap<>();
        args.put("x-delayed-type", "direct");
        return new CustomExchange(
                getRetry().getExchange(),      // 名稱
                "x-delayed-message",           // 類型（延遲消息插件）
                true,                          // durable
                false,                         // auto-delete
                args                           // 參數(shù)：聲明 delayed 類型為 direct
        );
    }
    /**
     * 綁定延遲隊(duì)列到延遲交換機(jī)
     */
    @Bean
    public Binding delayBinding() {
        return BindingBuilder.bind(delayQueue())
                .to(delayExchange())
                .with(getRetry().getRoutingKey())
                .noargs();
    }

?? 發(fā)送延時(shí)消息

/**
     * 默認(rèn)延遲 30 分鐘（1800000 毫秒）
     */
    public void sendDelayedTask(String message) {
        this.sendDelayedTask(message, 30 * 60 * 1000);
    }
    /**
     * 發(fā)送延遲消息
     */
    public void sendDelayedTask(String message, long delayMillis) {
        MessageProperties properties = new MessageProperties();
        properties.setContentType(MessageProperties.CONTENT_TYPE_JSON);
        properties.setHeader("x-delay", delayMillis); // 設(shè)置延遲時(shí)間（單位：毫秒）
        Message amqpMessage = new Message(message.getBytes(StandardCharsets.UTF_8), properties);
        rabbitTemplate.send(
                rabbitMqProperties.getConfig().getRetry().getExchange(),
                rabbitMqProperties.getConfig().getRetry().getRoutingKey(),
                amqpMessage
        );
        log.info("延遲任務(wù)已發(fā)送，延遲 {} s 后投遞: {}", delayMillis/1000, message);
    }

?? 五、注意事項(xiàng)

• 插件啟用后，RabbitMQ 會(huì)新增一種交換機(jī)類型 x-delayed-message
• 不兼容 RabbitMQ 自帶的 TTL + DLX 延遲實(shí)現(xiàn)方案
• 插件僅支持設(shè)置 發(fā)送時(shí)延遲，不支持基于隊(duì)列統(tǒng)一延遲
• 插件不支持所有 AMQP 客戶端（需顯式支持 header 設(shè)置的客戶端）

?? 六、Web 管理界面配置示例

你也可以通過(guò) RabbitMQ 的 Web 控制臺(tái)：

1. 創(chuàng)建交換機(jī)類型為 x-delayed-message
2. 設(shè)置參數(shù) x-delayed-type = direct
3. 綁定隊(duì)列
4. 消息發(fā)布時(shí)設(shè)置 header：x-delay: 10000（單位 ms）

?? 七、延時(shí)隊(duì)列 VS RabbitMQ 原生 TTL 實(shí)現(xiàn)

? 八、常見問題 FAQ

1. 插件安裝后沒有效果？

• 檢查是否重啟 RabbitMQ
• 檢查交換機(jī)類型是否正確設(shè)置為 x-delayed-message
• 檢查消息是否設(shè)置了 header：x-delay

2. 插件是否與集群兼容？

支持，只需確保每個(gè)節(jié)點(diǎn)都安裝啟用了插件。

?? 總結(jié)

通過(guò)安裝 rabbitmq_delayed_message_exchange 插件，RabbitMQ 擁有了真正意義上的“消息級(jí)”延遲投遞功能，無(wú)需依賴死信隊(duì)列和 TTL，極大簡(jiǎn)化了延時(shí)場(chǎng)景的開發(fā)與維護(hù)工作。

到此這篇關(guān)于RabbitMQ 延時(shí)隊(duì)列插件安裝與使用示例詳解(基于 Delayed Message Plugin)的文章就介紹到這了,更多相關(guān)RabbitMQ 延時(shí)隊(duì)列插件內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

最新評(píng)論