面試官：談談RabbitMQ隊頭阻塞問題？

作者：磊哥 2025-02-19 00:00:00

開發前端

隊頭阻塞問題是發生在使用死信隊列加 TTL 實現 RabbitMQ 延遲消息的場景中，造成的原因是隊列先進先出的特性，加上延遲消息的檢查機制導致的，我們可以使用 RabbitMQ 的延遲插件來避免此問題。

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

實現原理

RabbitMQ 延遲消息的實現方式有以下兩種：

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

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

死信隊列 + TTL 的實現流程如下：

圖片

生產者先將設置了 TTL（過期時間）的消息發送到普通隊列。
普通隊列沒有消息者，所以一定會過期，消息過期之后就會發送到死信隊列。
消費者訂閱死信隊列獲取消息，并執行延遲任務。

代碼實現

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

定義死信交換器（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");
    }

    // 定義普通隊列，設置死信交換器和路由鍵
    @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");
        // 可選：設置隊列級別的 TTL（所有消息統一過期時間）
        args.put("x-message-ttl", 10000); // 10秒
        return new Queue("main.queue", true, false, false, args);
    }

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

發送消息時設置 TTL（消息級別）：

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

消費者監聽死信隊列：

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

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

隊列級別：通過設置隊列的 x-message-ttl 參數，設置隊列統一的過期時間。

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

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

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

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

造成隊頭阻塞的原因

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

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

如下圖所示：

圖片

解決方案

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

小結

那么問題來了，使用延遲插件如何實現延遲任務？它和死信隊列的實現方式有哪些具體的區別呢？

責任編輯：武曉燕來源：磊哥和Java