快捷導(dǎo)航

簡單聊聊RabbitMQ中的隊(duì)頭阻塞問題

更新時(shí)間：2025年02月19日 08:14:28 作者：Java中文社群

這篇文章主要為大家詳細(xì)介紹了RabbitMQ中的隊(duì)頭阻塞問題以及相關(guān)的解決方法,文中的示例代碼講解詳細(xì),感興趣的小伙伴可以跟隨小編一起學(xué)習(xí)一下

RabbitMQ 延遲消息的隊(duì)頭阻塞問題是指，在使用死信隊(duì)列（DLX）和 TTL（消息過期時(shí)間）實(shí)現(xiàn)延遲消息時(shí)，由于隊(duì)列的先進(jìn)先出（FIFO）特性，在隊(duì)列頭部消息未過期的情況下，即使后續(xù)消息已經(jīng)過期也不能及時(shí)處理的情況。

實(shí)現(xiàn)原理

RabbitMQ 延遲消息的實(shí)現(xiàn)方式有以下兩種：

死信隊(duì)列+TTL。
使用 rabbitmq-delayed-message-exchange 插件。

而我們本文要討論的“RabbitMQ 延遲消息的隊(duì)頭阻塞問題”只會(huì)發(fā)生在死信隊(duì)列+TTL 的實(shí)現(xiàn)方式中。

死信隊(duì)列 + TTL 的實(shí)現(xiàn)流程如下：

生產(chǎn)者先將設(shè)置了 TTL（過期時(shí)間）的消息發(fā)送到普通隊(duì)列。
普通隊(duì)列沒有消息者，所以一定會(huì)過期，消息過期之后就會(huì)發(fā)送到死信隊(duì)列。
消費(fèi)者訂閱死信隊(duì)列獲取消息，并執(zhí)行延遲任務(wù)。

代碼實(shí)現(xiàn)

死信隊(duì)列 + TTL 在 Spring Boot 項(xiàng)目中的實(shí)現(xiàn)代碼如下。

定義死信交換器（DLX）和死信隊(duì)列：

// Spring Boot 配置示例
@Configuration
public class RabbitConfig {
    // 定義死信交換器
    @Bean
    public DirectExchange dlxExchange() {
        return new DirectExchange("dlx.exchange");
    }

    // 定義死信隊(duì)列
    @Bean
    public Queue dlxQueue() {
        return new Queue("dlx.queue");
    }

    // 綁定死信隊(duì)列到 DLX
    @Bean
    public Binding dlxBinding() {
        return BindingBuilder.bind(dlxQueue()).to(dlxExchange()).with("dlx.routing.key");
    }

    // 定義普通隊(duì)列，設(shè)置死信交換器和路由鍵
    @Bean
    public Queue mainQueue() {
        Map<String, Object> args = new HashMap<>();
        args.put("x-dead-letter-exchange", "dlx.exchange");
        args.put("x-dead-letter-routing-key", "dlx.routing.key");
        // 可選：設(shè)置隊(duì)列級(jí)別的 TTL（所有消息統(tǒng)一過期時(shí)間）
        args.put("x-message-ttl", 10000); // 10秒
        return new Queue("main.queue", true, false, false, args);
    }

    // 主隊(duì)列綁定到默認(rèn)交換器（根據(jù)需要調(diào)整）
    @Bean
    public Binding mainBinding() {
        return BindingBuilder.bind(mainQueue()).to(new DirectExchange("default.exchange")).with("main.routing.key");
    }
}

發(fā)送消息時(shí)設(shè)置 TTL（消息級(jí)別）：

// 發(fā)送延遲消息（消息級(jí)別 TTL）
public void sendDelayedMessage(String message, int delayMs) {
    rabbitTemplate.convertAndSend("default.exchange", "main.routing.key", message, msg -> {
        // 設(shè)置消息過期時(shí)間（覆蓋隊(duì)列級(jí)別的 TTL）
        msg.getMessageProperties().setExpiration(String.valueOf(delayMs));
        return msg;
    });
}

消費(fèi)者監(jiān)聽死信隊(duì)列：

@RabbitListener(queues = "dlx.queue")
public void handleDelayedMessage(String message) {
System.out.println("處理延遲消息: " + message);
}

所以說消息的過期時(shí)間 TTL 的設(shè)置方式有以下兩種：

隊(duì)列級(jí)別：通過設(shè)置隊(duì)列的 x-message-ttl 參數(shù)，設(shè)置隊(duì)列統(tǒng)一的過期時(shí)間。

Map<String, Object> args = new HashMap<>();
args.put("x-message-ttl", 60000); // 設(shè)置隊(duì)列消息過期時(shí)間為 60 秒
channel.queueDeclare(queueName, true, false, false, args);

消息級(jí)別：通過給每個(gè)消息設(shè)置 expiration 屬性，為每個(gè)消息設(shè)置過期時(shí)間。

AMQP.BasicProperties properties = new AMQP.BasicProperties.Builder()
        .deliveryMode(2) // 消息持久化
        .expiration("60000") // 設(shè)置消息過期時(shí)間為 60 秒
        .build();
channel.basicPublish(exchangeName, routingKey, properties, message.getBytes());

如果同時(shí)設(shè)置了消息級(jí) TTL 和隊(duì)列級(jí) TTL，消息的實(shí)際過期時(shí)間會(huì)取兩者中的最小值。

造成隊(duì)頭阻塞的原因

造成隊(duì)頭阻塞的原因有以下兩個(gè)：

先進(jìn)先出的隊(duì)列特性：隊(duì)列中的消息必須按順序處理，即使后面的消息 TTL 較短且已過期，也必須等待隊(duì)頭的消息先被處理（或過期）。
TTL 檢查機(jī)制：RabbitMQ 默認(rèn)僅在處理隊(duì)頭消息時(shí)檢查其 TTL，如果隊(duì)頭消息的 TTL 較長（例如 10 分鐘），即使后續(xù)消息的 TTL 更短（例如 1 分鐘），這些消息也會(huì)被阻塞，直到隊(duì)頭消息過期或被移除。

如下圖所示：

解決方案

為不同延遲時(shí)間創(chuàng)建獨(dú)立隊(duì)列：將相同 TTL 的消息放入同一隊(duì)列，避免消息的過期時(shí)間不一致。
使用延遲插件：使用 RabbitMQ 的延遲插件 rabbitmq_delayed_message_exchange，直接通過延遲交換機(jī)實(shí)現(xiàn)延遲消息，繞過死信隊(duì)列的 FIFO 限制。延遲插件是通過將消息存儲(chǔ)到內(nèi)置數(shù)據(jù)庫 Mnesia，再通過不斷判斷過期消息，實(shí)現(xiàn)延遲消息的投遞和執(zhí)行的，因此它不存在隊(duì)列的先進(jìn)先出和隊(duì)頭阻塞的問題。