更新於 2023/04/29發佈於 2023/04/29閱讀時間約 7 分鐘

【Message Queue - RabbitMQ】訊息的如何插隊？(Priority)

圖片來源

雖然我們的一般的狀況下我們都希望訊息能夠按照順序被處理，但有時候我們仍希望某些重要的訊息能夠優先被處理，也就是插隊的概念，正好RabbitMQ也有提供這樣需求的解決方案，以下是需要知道的幾個重點。

宣告Queue的時候必須要設定 x-max-priority，通常10就夠用了。
數字越大優先序越高。
一般不要設定太大，因為這算是特殊狀況，正常狀況下應該照順序來處理。
沒有指定優先序的訊息會預設為0。
生產端發送的訊息優先序若超過配置的最大值，則會以最大值做為優先序的標示。

配置範例

這邊會以管理界面作為範例，其他配置方式則大同小異，可以參考官方文件。

首先進到管理界面的Queues，並加入新的Queue，並配置優先序最大號碼為20。
參數的欄位填上： x-max-priority:20 。
型態記得要選number。

圖片來源

配置完我們的Queue會有Pri的標記。

接著我們將Queue綁定到交換機上。

圖片來源

開始進行實驗

實驗方向：依照訊息重要程度分為1~20個號碼，雖然我們先送出一個19號的訊息，再送出20號的訊息，但正確堆積在佇列的訊息應該要確保20號的訊息優先於19號。
Producer端會送出1、5、3、20的訊息。

(async () => {

  const conn = require("amqplib").connect(

    "amqp://test:test@127.0.0.1:5672/test"

  );

  const channel = await (await conn).createChannel();

  const priorityNums = [1, 5, 3, 20];

  const routingKey = "priority";

  const exchange = "qa_exchange_direct";

  for (let i = 0; i < priorityNums.length; i++) {

    const priority = priorityNums[i];

    const msg = `Priority_${priority}`;

    channel.publish(exchange, routingKey, Buffer.from(msg), {

      priority,

    });

    console.log(`Send Message: ${msg}`)

  }

})();Send Message: Priority_1

Send Message: Priority_5

Send Message: Priority_3

Send Message: Priority_20

Comsumer端則會依照優先序接收20、5、3、1的訊息

(async() => {

    const conn = require('amqplib').connect('amqp://inuqa:inuqa@127.0.0.1:5672/inuqa');

    const channel = await (await conn).createChannel();

    await channel.prefetch(1); // 最多能消費unacked的數量

    channel.consume("優先序隊列", (msg) => {

        setTimeout(() => {

            console.log(msg.content.toString());

            channel.ack(msg);

        }, 1000)

    }, {

        noAck: false

    })

})()Priority_20

Priority_5

Priority_3

Priority_1

注意事項

RabbitMQ預設是不支援優先序的，因為每個隊列的每個優先級都會有一些CPU跟磁碟的開銷成本。
官方建議不要使用Policy的方式來配置，因為Policy是動態的，可以在Queue宣告後進行參數的修改，但優先序隊列在宣告後就無法更改其數量。
假設一開始Queue沒有指定優先序，後續要加上優先序會需要將原本的Queue移除再重新宣告，因此最好一開始規劃功能時就將這一塊給考慮進去！！！！！！！！！！

如何取得訊息中某個優先序號碼剩下的數量？

雖然有了優先序可以幫我們依照訊息的重要程度來分配工作，但就應用面上來看，我們可能會需要去掌握某個優先序號碼在Queue裡面還剩下幾個未處理，此時我們可以藉由Management Plugin的API來進行查詢，查詢方式如下：

<http://$>{ip}/api/queues/${vhost}/${隊列名稱}

// <http://127.0.0.1:8083/api/queues/test/優先序隊列>

GET之後會得到以下的Response，我們只需要取backing_queue_status.priority_lengths.${優先序號碼}，就能知道剩餘數量有多少：

{

  ...,

  "backing_queue_status": {

    ...,

    "priority_lengths": {

      "0": 0,

      "1": 2,

      "2": 0,

      "3": 2,

      "4": 0,

      "5": 2,

      "6": 0,

      "7": 0,

      "8": 0,

      "9": 0,

      "10": 0,

      "11": 0,

      "12": 0,

      "13": 0,

      "14": 0,

      "15": 0,

      "16": 0,

      "17": 0,

      "18": 0,

      "19": 0,

      "20": 2

    },

     ...

  },

  ...

}