面試官：談談RabbitMQ隊頭阻塞問題？
作者：磊哥
隊頭阻塞問題是發(fā)生在使用死信隊列加 TTL 實現(xiàn) RabbitMQ 延遲消息的場景中，造成的原因是隊列先進先出的特性，加上延遲消息的檢查機制導致的，我們可以使用 RabbitMQ 的延遲插件來避免此問題。
RabbitMQ 延遲消息的隊頭阻塞問題是指，在使用死信隊列（DLX）和 TTL（消息過期時間）實現(xià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)者先將設置了 TTL（過期時間）的消息發(fā)送到普通隊列。
- 普通隊列沒有消息者，所以一定會過期，消息過期之后就會發(fā)送到死信隊列。
- 消費者訂閱死信隊列獲取消息，并執(zhí)行延遲任務。
代碼實現(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");
}
// 定義普通隊列，設置死信交換器和路由鍵
@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（所有消息統(tǒng)一過期時間）
args.put("x-message-ttl", 10000); // 10秒
return new Queue("main.queue", true, false, false, args);
}
// 主隊列綁定到默認交換器（根據(jù)需要調(diào)整）
@Bean
public Binding mainBinding() {
return BindingBuilder.bind(mainQueue()).to(new DirectExchange("default.exchange")).with("main.routing.key");
}
}- 發(fā)送消息時設置 TTL（消息級別）：
// 發(fā)送延遲消息（消息級別 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;
});
}- 消費者監(jiān)聽死信隊列：
@RabbitListener(queues = "dlx.queue")
public void handleDelayedMessage(String message) {
System.out.println("處理延遲消息: " + message);
}所以說消息的過期時間 TTL 的設置方式有以下兩種：
- 隊列級別：通過設置隊列的 x-message-ttl 參數(shù)，設置隊列統(tǒng)一的過期時間。
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 分鐘），即使后續(xù)消息的 TTL 更短（例如 1 分鐘），這些消息也會被阻塞，直到隊頭消息過期或被移除。
如下圖所示：
圖片
解決方案
- 為不同延遲時間創(chuàng)建獨立隊列：將相同 TTL 的消息放入同一隊列，避免消息的過期時間不一致。
- 使用延遲插件：使用 RabbitMQ 的延遲插件 rabbitmq_delayed_message_exchange，直接通過延遲交換機實現(xiàn)延遲消息，繞過死信隊列的 FIFO 限制。延遲插件是通過將消息存儲到內(nèi)置數(shù)據(jù)庫 Mnesia，再通過不斷判斷過期消息，實現(xiàn)延遲消息的投遞和執(zhí)行的，因此它不存在隊列的先進先出和隊頭阻塞的問題。
小結
隊頭阻塞問題是發(fā)生在使用死信隊列加 TTL 實現(xiàn) RabbitMQ 延遲消息的場景中，造成的原因是隊列先進先出的特性，加上延遲消息的檢查機制導致的，我們可以使用 RabbitMQ 的延遲插件來避免此問題。
那么問題來了，使用延遲插件如何實現(xiàn)延遲任務？它和死信隊列的實現(xiàn)方式有哪些具體的區(qū)別呢？
責任編輯：武曉燕
來源：
磊哥和Java
