---

本文是博主在記錄使用 RabbitMQ 在執行業務時遇到的問題和解決辦法，因此查閱了相關資料并做了以下記載，記錄了優先級隊列的機制和使用要點。

本文為長文，詳細介紹了相關的知識，可作為學習資料看。

---

一、優先級隊列介紹

首先我們要知道，RabbitMQ 是一個消息中間件，支持多種消息隊列模式，而優先級隊列就是其中一種，優先級隊列允許不同優先級的消息被存儲，高優先級 的消息會被 先消費。

• 配置隊列的優先級范圍比較常見的是例如 0 ~ 255 之間的數值，數值越大優先級越高 。
• 優先級的配置時機是在生產者在發送消息時，消費者在處理時，隊列會按優先級順序傳遞消息。

上面標黃的這句話會引出一個新的問題，比如，優先級隊列如何影響消息的排序？如果隊列已經有消息了，新來的高優先級消息會插到前面嗎？或者只在消費者獲取消息的時候才按優先級排序？

1、消息是在入優先級隊列時排序，還是等消費者拉取消息時排序？

我們知道優先級隊列和普通隊列不同，普通隊列是先進先出（FIFO），而優先級隊列會根據消息的優先級來決定處理順序。

假設隊列中已經有三個消息，優先級分別為1、2、3（假設數字越大優先級越高），順序是按照到達時間排列的。現在來了一條優先級為4的新消息，會引發先排序還是后排序的問題。

（1）先排序：消息入隊列立即排序

消息到達隊列先排序，立即將這條優先級為 4 的新消息插入到最前面，消費者直接獲取優先級最高的消息。

• 能做到消費者每次取消息時不需要額外的排序開銷。
• 隊列內部的結構需要支持快速插入和刪除，以保持消息的有序性，如 二叉堆結構 。

（2）后排序：消費者拉消息時排序

消息先入隊列，消費者來取的時候，隊列會按照優先級順序將消息提供給消費者。

• 消費者拉取消息時排序，需要一個能快速查找最高優先級消息的結構，如堆（heap） 。

上面兩種做法大家可以思考一下優缺點。

1.1 消息入隊列立即排序，即先排序

優先級隊列通常會在消息入隊時根據其優先級插入到正確的位置，而不是等到消費者獲取時才進行排序。

該方式可以確保消費者每次獲取到的都是當前隊列中優先級最高的消息，而不需要每次消費時都重新排序整個隊列。

因此，當一個新的高優先級消息到達時，它會被立即插入到隊列中比它優先級低的消息前面，這樣在后續的消費過程中，消費者會優先處理這條高優先級的消息。

1.2 優先級隊列通常遵循以下原則

• 入隊時排序：大多數消息隊列系統（如RabbitMQ）在消息入隊時即根據優先級調整順序。新到達的高優先級消息會立即插入到隊列中合適的位置（如前面），而非等待消費者拉取時才排序。這意味著隊列內部始終按優先級維護有序狀態。
• 數據結構支持：隊列通常使用 堆（Heap） 或類似結構，確保插入和提取最高優先級消息的時間復雜度為O(log n) 。例如，新消息D（優先級4）到達后，會被插入到比現有消息更高優先級的位置。
• 消費者行為透明：消費者獲取消息時，無需感知排序過程，直接按隊列現有順序取出最高優先級的消息。隊列在入隊時已完成排序，因此消費時無額外開銷。

示例場景:
現有隊列：消息A（1）、B（2）、C（3）→ 順序為 A → B → C（假設數字越大優先級越高）。  
新消息D（4）到達 → 隊列調整為 D（4）→ C（3）→ B（2）→ A（1）。  
消費者下次獲取時，直接取出D。

• 優勢：提前排序確保消費效率，避免每次拉取時計算。
• 限制：高頻插入高優先級消息可能導致調整成本增加，但通常影響可控。

注意：RabbitMQ 的優先級隊列有一個最大優先級的限制（0到255），并且如果隊列中已經有大量消息，插入一個高優先級消息可能會導致性能問題，因為它需要調整內部結構。

---

2、高優先級消息源源不斷，優先級隊列如何處理低優先級消息？

在了解了消息如何入隊列之后，就會緊接著遇到下一個問題，高優先級的消息一直進來，隊列可能會一直優先處理這些高優先級的，導致低優先級的消息被 餓死 ，也就是永遠得不到處理。這時候我們需要考慮怎么讓低優先級的消息也能有機會被處理到。

2.1 時間衰減機制（Aging）

• 原理：動態提升長時間未處理消息的優先級。
• 實現：為每個消息記錄等待時間（如時間戳）。
• 定義優先級增長函數（如線性增長 新優先級 = 原優先級 + k*等待時間 ）。
• 定期掃描隊列，更新低優先級消息的權重。
• 效果：避免低優先級任務無限期等待，平衡公平性。

2.2 分層隊列 + 比例調度

• 原理：通過多級隊列和固定比例分配資源。
• 實現：將隊列分為高、中、低三級，按比例分配處理機會，如 3:2:1 。
• 使用加權輪詢（WRR）或赤字輪詢（DRR）算法調度。
• 示例：網絡 QoS 中為不同流量類型分配帶寬。（自行查閱）

2.3 動態優先級調整

• 原理：根據系統負載動態調整優先級策略。
• 實現：監控隊列深度和處理延遲，當高優先級隊列超過閾值時，臨時提升低優先級任務的權重。
• 結合反饋控制（如PID控制器）自動調整參數。
• 場景：實時系統在過載時降級非關鍵任務。

2.4 容量限制與降級

• 原理：限制高優先級隊列容量，強制資源釋放。
• 實現：設置高優先級隊列的最大長度（如1000條），超出容量后，新到的高優先級消息降級為中等優先級。
• 結合斷路器模式拒絕過量請求。
• 優勢：防止高優先級洪泛導致系統崩潰。

2.5 空閑時段處理

• 原理：利用系統空閑時間處理積壓任務。
• 實現：定義空閑檢測機制，如 CPU利用率 < 20% 持續 5 秒 ，觸發后臺線程批量處理低優先級隊列。
• 結合預取和緩存優化處理效率。
• 案例：數據庫在低峰期執行維護任務。

常見的策略都在上面展示出來了，如果可以跟據業務需求進行調整，選擇其中的一種或者組合使用。在使用的過程中要注意好如何權衡與優化。

• 公平性 vs 吞吐量：嚴格按優先級排序最大化吞吐量，但需犧牲公平性；引入Aging或分層隊列可改善公平性。
• 實時性要求：硬實時系統可能需要絕對優先級，而軟實時系統可結合超時重試機制。
• 監控與調優：跟蹤低優先級任務的平均等待時間（如 Prometheus 監控），動態調整Aging因子和隊列比例（如基于強化學習）。

常見的一些應用場景，可以根據示例搜索如何使用。

• 操作系統調度：Linux CFS 調度器通過虛擬運行時間（vruntime）實現公平性。
• 消息隊列系統：RabbitMQ 通過 x-max-priority 和插件支持優先級衰減。
• 微服務架構：Envoy 網關通過流量優先級分級保障關鍵 API 的 SLA。

通過上述策略的組合，系統能夠在高負載下兼顧高優先級任務的及時處理和低優先級任務的最終可達性。

---

3、優先級參數是聲明隊列的時候設置，還是在發送消息時設置？

在 RabbitMQ 中，優先級隊列的配置需要同時在聲明隊列時和發送消息時設置，二者缺一不可。

3.1 聲明隊列時：啟用優先級支持

必須在隊列聲明階段通過參數 x-max-priority 顯式啟用優先級功能，并定義優先級范圍。

Map<String, Object> args = new HashMap<>();
args.put("x-max-priority", 10);  // 定義優先級范圍：0-10（共11級）
channel.queueDeclare("priority_queue", true,   // 持久化隊列false, false, args    // 傳入參數
);

注意事項

• 必須顯式聲明：如果隊列未設置 x-max-priority，發送消息時指定的優先級將被忽略。
• 取值范圍：推薦值 0-10，數值越高性能損耗越大（RabbitMQ 官方建議不超過 255）。
• 不可修改：已創建的隊列無法動態修改優先級范圍，需刪除重建。

3.2 發送消息時：指定消息優先級

在消息屬性中通過 priority 字段設置具體優先級值。

AMQP.BasicProperties props = new AMQP.BasicProperties.Builder().priority(7)  // 設置優先級為7（需≤隊列的x-max-priority）.deliveryMode(2)  // 持久化消息.build();channel.basicPublish("", "priority_queue", props, "消息內容".getBytes()
);

3.3 優先級規則

• 數值越大優先級越高：0 最低，最大值由 x-max-priority 決定。
• 自動截斷：若消息優先級超過 x-max-priority，會被截斷為該最大值。
• 如隊列聲明 x-max-priority=5，消息設置 priority=8 會被視為 5。
• 默認優先級：未顯式設置時，優先級為 0。

3.4 完整流程示例

步驟 1：聲明支持優先級的隊列

// 創建連接和Channel
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost");
try (Connection connection = factory.newConnection();Channel channel = connection.createChannel()) {// 定義優先級參數Map<String, Object> args = new HashMap<>();args.put("x-max-priority", 10);// 聲明隊列channel.queueDeclare("orders", true,   // 持久化隊列false, false, args);
}

步驟 2：發送帶優先級的消息

// 發送高優先級訂單消息
AMQP.BasicProperties highPriorityProps = new AMQP.BasicProperties.Builder().priority(8).build();
channel.basicPublish("", "orders", highPriorityProps, "VIP訂單".getBytes());// 發送普通優先級消息（默認0）
channel.basicPublish("", "orders", null, "普通訂單".getBytes()); 

步驟 3：消費者驗證優先級

DeliverCallback callback = (consumerTag, delivery) -> {String msg = new String(delivery.getBody(), "UTF-8");int priority = delivery.getProperties().getPriority();System.out.println("收到優先級 " + priority + " 的消息: " + msg);channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
};
channel.basicConsume("orders", false, callback, consumerTag -> {});

二、優先級隊列的常見問題和最佳實踐

1、常見問題與解決方案

問題場景	原因	解決方案
消息優先級未生效	隊列未聲明 x-max-priority	刪除舊隊列，重新聲明帶優先級的隊列
高優先級消息積壓低優先級消息	持續高負載導致饑餓現象	結合時間衰減（Aging）機制提升老消息優先級
消費者未按優先級順序處理消息	使用 basicConsume 自動分配消息	改用 basicGet 手動拉取消息并按優先級排序
優先級數值超過隊列限制	生產者設置值 > x-max-priority	在生產者端做優先級范圍校驗
消息順序錯亂	消費者使用了 prefetchCount > 1，導致低優先級消息被預取	channel.basicQos(1); 每次只接收一條消息

2、使用優先級隊列的最佳實踐

• 限制優先級層級：建議使用 0-5 級，避免過多層級增加調度復雜度。
• 監控隊列狀態：通過 RabbitMQ Management Plugin 監控：
  • queue_messages_ready（按優先級分組）
  • message_age（消息等待時間）
• 防御性設計：

  // 生產者端校驗優先級合法性
  int safePriority = Math.min(requestedPriority, queueMaxPriority);
  

• 混合隊列策略：

  // 高、中、低優先級分隊列處理
  channel.queueDeclare("high_pri", true, false, false, Map.of("x-max-priority", 10));
  channel.queueDeclare("low_pri", true, false, false, null); // 無優先級隊列
  

3、普通隊列變成優先級隊列會導致已有消息丟失

如果在隊列已經存在的情況下，修改 x-max-priority 參數會出現問題。

比如，原本隊列沒有設置優先級，后來想啟用，可能需要重新聲明隊列，但這會導致已有的消息丟失。

4、要注意優先級隊列的《預計取數》設置

優先級隊列和消費者的 預取計數（prefetch count） 之間的關系。

• 如果消費者設置了較高的預取數，可能會一次性獲取多個消息，這時候即使有更高優先級的消息到達，已經被預取的消息可能不會被中斷，導致高優先級消息不能及時處理。
• 因此，可能需要合理設置預取數，或者使用 單條預取（prefetch=1） 來保證高優先級消息盡快被處理。

5、多個消費者消費一個隊列，優先級消息如何分配？

如果有多個消費者，高優先級的消息會優先被哪個消費者獲取？可能取決于消費者的空閑情況，但優先級高的消息會先被投遞給空閑的消費者。

三、Java 示例實現優先級隊列

1、添加 Maven 依賴

在 pom.xml 中添加 RabbitMQ Java 客戶端依賴：

<dependency><groupId>com.rabbitmq</groupId><artifactId>amqp-client</artifactId><version>5.12.0</version>
</dependency>

2、完整 Java 代碼實現

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;
import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeoutException;public class PriorityQueueExample {private final static String QUEUE_NAME = "java_priority_queue";public static void main(String[] args) {// 創建生產者線程Thread producerThread = new Thread(() -> {try {sendMessages();} catch (IOException | TimeoutException e) {e.printStackTrace();}});// 創建消費者線程Thread consumerThread = new Thread(() -> {try {consumeMessages();} catch (IOException | TimeoutException e) {e.printStackTrace();}});producerThread.start();consumerThread.start();}/*** 生產者發送消息*/private static void sendMessages() throws IOException, TimeoutException {ConnectionFactory factory = new ConnectionFactory();factory.setHost("localhost");try (Connection connection = factory.newConnection();Channel channel = connection.createChannel()) {// 聲明優先級隊列參數Map<String, Object> args = new HashMap<>();args.put("x-max-priority", 10); // 設置最大優先級為10// 聲明隊列channel.queueDeclare(QUEUE_NAME, true, false, false, args);// 發送不同優先級的消息int[] priorities = {1, 3, 5};for (int priority : priorities) {String message = "Message with priority " + priority;// 設置消息屬性（包含優先級）AMQP.BasicProperties props = new AMQP.BasicProperties.Builder().priority(priority).build();channel.basicPublish("", QUEUE_NAME, props, message.getBytes());System.out.println("Sent: " + message);}}}/*** 消費者接收消息*/private static void consumeMessages() throws IOException, TimeoutException {ConnectionFactory factory = new ConnectionFactory();factory.setHost("localhost");Connection connection = factory.newConnection();Channel channel = connection.createChannel();// 定義消息回調處理DefaultConsumer consumer = new DefaultConsumer(channel) {@Overridepublic void handleDelivery(String consumerTag, Envelope envelope,AMQP.BasicProperties properties, byte[] body) {String message = new String(body);int priority = properties.getPriority();System.out.println("Consumed: " + message + " | Priority: " + priority);}};// 開始消費（自動確認）channel.basicConsume(QUEUE_NAME, true, consumer);// 保持消費者線程運行try {Thread.sleep(5000); // 等待5秒確保消息處理完成} catch (InterruptedException e) {e.printStackTrace();} finally {channel.close();connection.close();}}
}

3、關鍵代碼解釋

3.1 聲明優先級隊列

Map<String, Object> args = new HashMap<>();
args.put("x-max-priority", 10); // 必須參數
channel.queueDeclare(QUEUE_NAME, true, false, false, args);

• x-max-priority：隊列支持的最大優先級值（1-255）
• 參數必須通過 Map<String, Object> 傳遞

3.2 發送優先級消息

AMQP.BasicProperties props = new AMQP.BasicProperties.Builder().priority(priority) // 設置優先級.build();
channel.basicPublish("", QUEUE_NAME, props, message.getBytes());

3.3 消費消息

DefaultConsumer consumer = new DefaultConsumer(channel) {@Overridepublic void handleDelivery(String consumerTag, Envelope envelope,AMQP.BasicProperties properties, byte[] body) {int priority = properties.getPriority(); // 獲取優先級}
};

4、運行與驗證

• 啟動生產者：發送優先級為 1、3、5 的消息
• 啟動消費者：觀察輸出順序應為 5 → 3 → 1

輸出示例：

Sent: Message with priority 1
Sent: Message with priority 3
Sent: Message with priority 5
Consumed: Message with priority 5 | Priority: 5
Consumed: Message with priority 3 | Priority: 3
Consumed: Message with priority 1 | Priority: 1

5、參數設置

• 隊列持久化：channel.queueDeclare(..., true, ...) 中第二個參數表示隊列持久化
• 消息持久化：通過 AMQP.BasicProperties.Builder().deliveryMode(2) 設置

6、異常處理

• 必須處理 IOException 和 TimeoutException
• 使用 try-with-resources 自動關閉連接：

  try (Connection connection = factory.newConnection();Channel channel = connection.createChannel()) {// ...
  }
  

7、多線程消費

• 如果啟動多個消費者，高優先級消息會被空閑消費者優先獲取
• 建議設置 prefetchCount=1：單條預取

  channel.basicQos(1);
  

通過以上 Java 實現，可以完整地使用 RabbitMQ 的優先級隊列功能。如果需要進一步優化，可以結合線程池、消息確認機制等擴展功能。