快捷導(dǎo)航

簡單聊聊RabbitMQ中的隊頭阻塞問題

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

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

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

實現(xiàn)原理

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

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

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

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

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

代碼實現(xiàn)

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

定義死信交換器（DLX）和死信隊列：

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

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

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

    // 定義普通隊列，設(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è)置隊列級別的 TTL（所有消息統(tǒng)一過期時間）
        args.put("x-message-ttl", 10000); // 10秒
        return new Queue("main.queue", true, false, false, args);
    }

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

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

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

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

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

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

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

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

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

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

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

造成隊頭阻塞的原因

造成隊頭阻塞的原因有以下兩個：

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

如下圖所示：

解決方案

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