RabbitMQ3.13.0起支持MQTT5.0協(xié)議及MQTT5.0特性功能列表

RabbitMQ 3.12 中發(fā)布的原生 MQTT 為物聯(lián)網用例提供了顯著的可擴展性和性能改進。

RabbitMQ 3.13 將支持 MQTT 5.0，因此將成為我們使 RabbitMQ 成為領先的 MQTT 代理之一的下一個重要步驟。

這篇博文解釋了如何在 RabbitMQ 中使用新的 MQTT 5.0 功能。

1. MQTT概覽

MQTT 是物聯(lián)網 （IoT） 的標準協(xié)議。

物聯(lián)網遠程設備在連接到代理時網絡質量可能較差。 因此，MQTT是輕量級的：MQTT協(xié)議頭很小，可以節(jié)省網絡帶寬。

由于物聯(lián)網設備可能經常斷開連接并重新連接（想象一下一輛汽車駛過隧道），MQTT 也很高效：與其他消息傳遞協(xié)議相比，客戶端通過更短的握手進行連接和身份驗證。

MQTT協(xié)議已經存在了很多年。 如下表所示，最新的 MQTT 協(xié)議版本為 5.0。

值得一提的是，面向用戶的協(xié)議版本和“內部”協(xié)議版本（也稱為協(xié)議級別）是有區(qū)別的。 后者在 CONNECT 數(shù)據包中從客戶端發(fā)送到服務器。 由于面向用戶的協(xié)議版本 3.1.1 映射到內部協(xié)議版本 4，為了避免進一步的混淆，MQTT 委員會決定跳過面向用戶的版本 4.0，以便面向用戶的版本 5.0 映射到內部協(xié)議版本 5。

2. MQTT 5.0 特性

1. 特性概要

附錄 C. MQTT v5.0 中的新功能摘要提供了 MQTT 5.0 新功能的完整列表。

由于您在 Web 上找到了很棒的 MQTT 5.0 資源，包括說明性圖表和使用模式，因此這篇博文僅關注 RabbitMQ 的細節(jié)。 本節(jié)介紹 PR #7263 中實現(xiàn)的最重要功能。 對于每個功能，我們提供了一個如何將其與 RabbitMQ 一起使用的示例，或者概述了如何在 RabbitMQ 中實現(xiàn)它的高級描述。

2. Docker中安裝RabbitMQ及啟用MQTT5.0協(xié)議

要自己運行示例，請啟動 RabbitMQ 服務器 3.13， 例如，使用以下 Docker 鏡像標記：

docker run -it --rm --name rabbitmq -p 1883:1883 -p 15672:15672 -p 15692:15692 rabbitmq:3.13.0-management

在另一個終端窗口中，啟用 MQTT 插件：

docker exec rabbitmq rabbitmq-plugins enable rabbitmq_mqtt

由于 MQTT 插件是動態(tài)啟用的，因此 MQTT 插件定義的功能標志被禁用。 啟用所有功能標志，包括功能標志：mqtt_v5

docker exec rabbitmq rabbitmqctl enable_feature_flag all

現(xiàn)在，列出功能標志應顯示所有功能標志都已啟用：

docker exec rabbitmq rabbitmqctl list_feature_flags --formatter=pretty_table

以下示例使用 MQTTX CLI V1.9.4。 我們使用 CLI 而不是圖形 UI，以便您可以通過復制粘貼命令輕松運行示例。

所有新功能也適用于 RabbitMQ Web MQTT 插件。

3. MQTT 5.0 功能列表

1. 消息過期

1. 描述

可以為發(fā)布到代理的每條消息設置以秒為單位的到期間隔。 如果在該過期時間間隔內未使用郵件，則該郵件將被丟棄或死信。

2. 舉例

為 主題 創(chuàng)建訂閱。 這將在 RabbitMQ 中創(chuàng)建一個隊列。 通過鍵入終端斷開客戶端連接。 由于我們使用 600 秒的會話到期間隔，因此此隊列將再存在 10 分鐘。t/1``Ctrl+C

mqttx sub --client-id sub-1 --topic t/1 --session-expiry-interval 600 --qos 1
…  Connecting...
?  Connected
…  Subscribing to t/1...
?  Subscribed to t/1
^C

將消息發(fā)布到同一主題，消息過期間隔為 30 秒：

mqttx pub --topic t/1 --message m1 --message-expiry-interval 30 --qos 1
…  Connecting...
?  Connected
…  Message publishing...
?  Message published

在接下來的 30 秒內，列出隊列：

docker exec rabbitmq rabbitmqctl --quiet --formatter=pretty_table list_queues name type messages
┌─────────────────────────────┬─────────┬──────────┐
│ name                        │ type    │ messages │
├─────────────────────────────┼─────────┼──────────┤
│ mqtt-subscription-sub-1qos1 │ classic │ 1        │
└─────────────────────────────┴─────────┴──────────┘

等待 30 秒，然后再次列出隊列：

docker exec rabbitmq rabbitmqctl --quiet --formatter=pretty_table list_queues
┌─────────────────────────────┬─────────┬──────────┐
│ name                        │ type    │ messages │
├─────────────────────────────┼─────────┼──────────┤
│ mqtt-subscription-sub-1qos1 │ classic │ 0        │
└─────────────────────────────┴─────────┴──────────┘

該消息已過期，因為客戶端尚未連接到代理以使用該消息。 如果設置了死字策略，則郵件將死信發(fā)送到交易所。 在我們的例子中，死字被禁用。 查詢 Prometheus 端點可證明經典隊列中有 1 條消息已過期。sub-1

curl --silent localhost:15692/metrics | grep rabbitmq_global_messages_dead_lettered_expired_total
# TYPE rabbitmq_global_messages_dead_lettered_expired_total counter
# HELP rabbitmq_global_messages_dead_lettered_expired_total Total number of messages dead-lettered due to message TTL exceeded
rabbitmq_global_messages_dead_lettered_expired_total{queue_type="rabbit_classic_queue",dead_letter_strategy="at_most_once"} 0
rabbitmq_global_messages_dead_lettered_expired_total{queue_type="rabbit_classic_queue",dead_letter_strategy="disabled"} 1
rabbitmq_global_messages_dead_lettered_expired_total{queue_type="rabbit_quorum_queue",dead_letter_strategy="at_least_once"} 0
rabbitmq_global_messages_dead_lettered_expired_total{queue_type="rabbit_quorum_queue",dead_letter_strategy="at_most_once"} 0
rabbitmq_global_messages_dead_lettered_expired_total{queue_type="rabbit_quorum_queue",dead_letter_strategy="disabled"} 0

另一個有趣的功能是以下要求：

服務器發(fā)送到客戶端的 PUBLISH 數(shù)據包必須包含設置為接收值減去應用程序消息在服務器中等待的時間的消息到期間隔。

向代理發(fā)送第二條消息，消息到期間隔為 60 秒：

mqttx pub --topic t/1 --message m2 --message-expiry-interval 60 --qos 1

等待 20 秒，然后重新連接訂閱客戶端：

mqttx sub --client-id sub-1 --topic t/1 --no-clean --session-expiry-interval 0  --qos 1 --output-mode clean
{
  "topic": "t/1",
  "payload": "m2",
  "packet": {
    ...
    "properties": {
      "messageExpiryInterval": 40
    }
  }
}

根據 MQTT 5.0 協(xié)議規(guī)范的規(guī)定，客戶端接收第二條消息，消息到期間隔設置為 40 秒： 代理接收的 60 秒減去消息在代理中等待的 20 秒。

3. 實現(xiàn)

MQTT 5.0 消息到期是在 RabbitMQ 中使用每條消息 TTL 實現(xiàn)的，類似于 AMQP 0.9.1 發(fā)布者中的字段。expiration

2. 訂閱標識符

1.描述

客戶端可以在 SUBSCRIBE 數(shù)據包中設置訂閱標識符。 如果客戶端因該訂閱而收到消息，則代理會將該訂閱標識符包含在 PUBLISH 數(shù)據包中。

訂閱標識符的用例列在 SUBSCRIBE 操作部分。

2. 舉例

從同一客戶端向服務器發(fā)送 3 個單獨的 SUBSCRIBE 數(shù)據包，每個數(shù)據包具有不同的主題過濾器和不同的訂閱標識符：

mqttx sub --client-id sub-2 --topic t/1 --subscription-identifier 1 --session-expiry-interval 600
^C
mqttx sub --client-id sub-2 --topic t/2 --subscription-identifier 2 --session-expiry-interval 600 --no-clean
^C
mqttx sub --client-id sub-2 --topic "t/#" --subscription-identifier 3 --session-expiry-interval 0 --no-clean --output-mode clean

在第二個終端窗口中，我們看到從同一隊列到同一主題交換的 3 個綁定，每個綁定都具有不同的路由鍵：

docker exec rabbitmq rabbitmqctl --quiet --formatter=pretty_table list_bindings \
    source_name source_kind destination_name destination_kind routing_key
┌─────────────┬─────────────┬─────────────────────────────┬──────────────────┬─────────────────────────────┐
│ source_name │ source_kind │ destination_name            │ destination_kind │ routing_key                 │
├─────────────┼─────────────┼─────────────────────────────┼──────────────────┼─────────────────────────────┤
│             │ exchange    │ mqtt-subscription-sub-2qos0 │ queue            │ mqtt-subscription-sub-2qos0 │
├─────────────┼─────────────┼─────────────────────────────┼──────────────────┼─────────────────────────────┤
│ amq.topic   │ exchange    │ mqtt-subscription-sub-2qos0 │ queue            │ t.#                         │
├─────────────┼─────────────┼─────────────────────────────┼──────────────────┼─────────────────────────────┤
│ amq.topic   │ exchange    │ mqtt-subscription-sub-2qos0 │ queue            │ t.1                         │
├─────────────┼─────────────┼─────────────────────────────┼──────────────────┼─────────────────────────────┤
│ amq.topic   │ exchange    │ mqtt-subscription-sub-2qos0 │ queue            │ t.2                         │
└─────────────┴─────────────┴─────────────────────────────┴──────────────────┴─────────────────────────────┘

第一個條目是與默認交換的隱式綁定。

每個具有 MQTT 主題篩選器的 MQTT 訂閱對應一個帶有綁定鍵的 AMQP 0.9.1 綁定。 準確地說，表列的名稱錯誤：應該改為調用它。 MQTT 中的主題級分隔符是 “” 字符，而 AMQP 0.9.1 主題交換中的主題級分隔符是 “” 字符。routing_key``binding_key``/``.

再次在第二個終端窗口中，向主題發(fā)送消息：t/1

mqttx pub --topic t/1 --message m1

（訂閱客戶端的）第一個終端窗口接收以下 PUBLISH 數(shù)據包：

{
  "topic": "t/1",
  "payload": "m1",
  "packet": {
    ...
    "properties": {
      "subscriptionIdentifier": [
        1,
        3
      ]
    }
  }
}

它包含訂閱標識符 1 和 3，因為 topic filters 和 match topic .t/1``t/#``t/1

同樣，如果向主題發(fā)送第二條消息，訂閱客戶端將收到包含訂閱標識符 2 和 3 的 PUBLISH 數(shù)據包。t/2

3. 實現(xiàn)

訂閱標識符是 MQTT 會話狀態(tài)的一部分。 因此，在客戶端斷開連接時，訂閱標識符必須保留在服務器的數(shù)據庫中，直到 MQTT 會話結束。 RabbitMQ 將訂閱標識符存儲在綁定參數(shù)中：

docker exec rabbitmq rabbitmqctl --quiet --formatter=pretty_table list_bindings routing_key arguments
┌─────────────────────────────┬───────────────────────────────────────────────────────────────────────────────────┐
│ routing_key                 │ arguments                                                                         │
├─────────────────────────────┼───────────────────────────────────────────────────────────────────────────────────┤
│ mqtt-subscription-sub-2qos0 │                                                                                   │
├─────────────────────────────┼───────────────────────────────────────────────────────────────────────────────────┤
│ t.#                         │ {mqtt_subscription_opts,0,false,false,0,3}{<<"x-binding-key">>,longstr,<<"t.#">>} │
├─────────────────────────────┼───────────────────────────────────────────────────────────────────────────────────┤
│ t.1                         │ {mqtt_subscription_opts,0,false,false,0,1}{<<"x-binding-key">>,longstr,<<"t.1">>} │
├─────────────────────────────┼───────────────────────────────────────────────────────────────────────────────────┤
│ t.2                         │ {mqtt_subscription_opts,0,false,false,0,2}{<<"x-binding-key">>,longstr,<<"t.2">>} │
└─────────────────────────────┴───────────────────────────────────────────────────────────────────────────────────┘

綁定參數(shù)的確切結構并不重要，并且可能會在將來的 RabbitMQ 版本中更改。 但是，可以在綁定參數(shù)中看到整數(shù) 1、2 和 3，這些參數(shù)對應于訂閱標識符。

當主題交換路由消息時，發(fā)布 Erlang 進程會將所有匹配的綁定鍵包含在消息中。 訂閱 MQTT 客戶端的 Erlang 進程將匹配的綁定密鑰與其知道的 MQTT 主題過濾器進行比較，并將訂閱標識符包含在發(fā)送到 MQTT 客戶端的 PUBLISH 數(shù)據包中。

發(fā)布 Erlang 進程可以是 MQTT 連接進程，也可以是 AMQP 0.9.1 通道進程。 一如既往，RabbitMQ 在跨協(xié)議互操作性方面表現(xiàn)出色：當 AMQP 0.9.1（或 STOMP 或 AMQP 1.0）客戶端向主題交換發(fā)送消息時， 正確的訂閱標識符將包含在發(fā)送到 MQTT 客戶端的 PUBLISH 數(shù)據包中。

3. 訂閱選項

1. 描述

MQTT 5.0 提供了 3 個新的訂閱選項：

1. 無本地
2. 保留為已發(fā)布
3. 保留處理

所有訂閱選項均由 RabbitMQ 實現(xiàn)。 在這里，我們只關注“保留處理”選項：

此選項指定在建立訂閱時是否發(fā)送保留的消息。
這些值為：
0 = 在訂閱
時發(fā)送保留消息 1 = 僅在訂閱當前不存在
時在訂閱時發(fā)送保留消息 2 = 在訂閱時不發(fā)送保留的消息

2. 舉例例

發(fā)送保留的消息：

mqttx pub --topic mytopic --message m --retain

保留處理值 0 將接收保留的消息，而值 2 不會：

mqttx sub --topic mytopic --retain-handling 0
…  Connecting...
?  Connected
…  Subscribing to mytopic...
?  Subscribed to mytopic
payload: m
retain: true
^C

mqttx sub --topic mytopic --retain-handling 2
…  Connecting...
?  Connected
…  Subscribing to mytopic...
?  Subscribed to mytopic

4 所有ACK上的原因代碼

1. 描述

數(shù)據包 CONNACK、PUBACK、SUBACK、UNSUBACK 和 DISCONNECT 包含原因碼。

2. 實現(xiàn)

一個實現(xiàn)示例是，如果消息未路由到任何隊列，則 RabbitMQ 將在 PUBACK 數(shù)據包中使用原因代碼進行回復。 MQTT 5.0 原因代碼在概念上對應于 AMQP 0.9.1 中的強制消息屬性和處理程序。No matching subscribers``No matching subscribers``BasicReturn

5. 用戶屬性

1. 描述

大多數(shù) MQTT 數(shù)據包可以包含用戶屬性。 用戶屬性的含義不是由 MQTT 規(guī)范定義的。

2. 示例 PUBLISH 數(shù)據包

PUBLISH 數(shù)據包中的用戶屬性由客戶端應用程序定義，并由服務器原封不動地轉發(fā)。

在第一個終端窗口中訂閱：

mqttx sub --topic t/5

在第二個終端窗口中發(fā)布包含用戶屬性的消息：

mqttx pub --topic t/5 --message m --user-properties "key1: value1"

第一個終端窗口將接收用戶屬性，原封不動：

payload: m
userProperties: [ { key: 'key1', value: 'value1' } ]

MQTT 5.0 PUBLISH 數(shù)據包中的用戶屬性類似于 AMQP 0.9.1 中的消息屬性。headers

3. 示例 CONNECT 數(shù)據包

使用用戶屬性進行連接：

mqttx conn --client-id myclient --user-properties "connecting-from: London"

在瀏覽器中打開管理 UI http://localhost:15672/#/connections（用戶名和密碼都是 ），然后單擊 MQTT 連接：guest

RabbitMQ 將在管理 UI 中顯示 CONNECT 數(shù)據包中的用戶屬性。

6. 有效負載格式和內容類型

1. 描述

發(fā)布者可以指定 MIME 內容類型。 它還可以設置有效負載格式指示器，指示有效負載是由 UTF-8 編碼的字符數(shù)據還是未指定的二進制數(shù)據組成。

2. 舉例

在第一個終端窗口中，訂閱一個主題：

mqttx sub --topic t/6 --output-mode clean

在第二個終端窗口中，發(fā)送一條帶有內容類型和有效負載格式指示符的消息：

mqttx pub --topic t/6 --message "my UTF-8 encoded data ??" --content-type text/plain --payload-format-indicator

第一個終端窗口將原封不動地接收內容類型和有效負載格式指示器：

{
  "topic": "t/6",
  "payload": "my UTF-8 encoded data ??",
  "packet": {
    ...
    "properties": {
      "payloadFormatIndicator": true,
      "contentType": "text/plain"
    }
  }
}

7. 請求/響應

1. 描述

MQTT 5.0 正式化了請求/響應模式。

在發(fā)布消息之前，MQTT 客戶端（請求者）訂閱響應主題。 請求者將響應主題和一些關聯(lián)數(shù)據包含在請求消息中。

另一個 MQTT 客戶端（響應者）接收到請求消息，執(zhí)行一些操作，并將具有相同關聯(lián)數(shù)據的響應消息發(fā)布到響應主題。

MQTT 5.0 請求/響應功能對應于 AMQP 0.9.1 中的遠程過程調用。 但是，在 AMQP 0.9.1 中，請求者將在 AMQP 0.9.1 消息屬性中包含回調隊列的名稱。 MQTT 協(xié)議沒有定義隊列的概念。因此，在 MQTT 中，被回復的“地址”是一個主題名稱。reply_to

盡管協(xié)議規(guī)范之間存在不兼容性，但 RabbitMQ 在協(xié)議互操作性方面大放異彩： 因此，RabbitMQ 支持跨協(xié)議的請求/響應交互。

例如，MQTT 客戶端可以在請求消息中包含響應主題和關聯(lián)數(shù)據。 如果 AMQP 0.9.1 客戶端創(chuàng)建了一個綁定到主題交換的隊列，該隊列的綁定密鑰與請求消息的主題匹配，它將收到一條 AMQP 0.9.1 消息，其屬性設置為 MQTT 客戶端發(fā)送的關聯(lián)數(shù)據和名為 . 然后，AMQP 0.9.1 客戶端可以使用相同的 MQTT 5.0 客戶端響應，并將響應消息發(fā)布到具有標頭中存在的主題的主題交換。amq.topic``correlation_id``x-opt-reply-to-topic``correlation_id``amq.topic``x-opt-reply-to-topic

2. 舉例

此示例僅關注 MQTT 客戶端。

在第一個終端窗口中，響應的 MQTT 客戶端訂閱主題t/7;

mqttx sub --client-id responder --topic t/7 --session-expiry-interval 600 --output-mode clean --qos 1

在第二個終端窗口中，請求的 MQTT 客戶端訂閱了一個名為 ：my/response/topic

mqttx sub --client-id requester --topic my/response/topic --session-expiry-interval 600 --qos 1
…  Connecting...
?  Connected
…  Subscribing to my/response/topic...
?  Subscribed to my/response/topic
^C

在第二個終端窗口中，請求者隨后發(fā)布一條請求消息：

mqttx pub --client-id requester --topic t/7 --message "my request" \
    --correlation-data abc-123 --response-topic my/response/topic \
    --session-expiry-interval 600 --no-clean

在第 1 個終端窗口中，響應方收到請求消息：

{
  "topic": "t/7",
  "payload": "my request",
  "packet": {
    ...
    "properties": {
      "responseTopic": "my/response/topic",
      "correlationData": {
        "type": "Buffer",
        "data": [
          97,
          98,
          99,
          45,
          49,
          50,
          51
        ]
      }
    }
  }
}
^C

在第一個終端窗口中，響應方通過復制關聯(lián)數(shù)據并發(fā)布到響應主題來響應請求者：

mqttx pub --client-id responder --topic my/response/topic --message "my response" --correlation-data abc-123

在第 2 個終端窗口中，請求者收到響應。

mqttx sub --client-id requester --topic my/response/topic --no-clean --qos 1 --output-mode clean
{
  "topic": "my/response/topic",
  "payload": "my response",
  "packet": {
    ...
    "properties": {
      "correlationData": {
        "type": "Buffer",
        "data": [
          97,
          98,
          99,
          45,
          49,
          50,
          51
        ]
      }
    }
  }
}

關聯(lián)數(shù)據可用于將響應與請求相關聯(lián)。 請求者通常為其發(fā)布的每個請求選取唯一的關聯(lián)數(shù)據。

8. 分配的客戶端標識符

1. 描述

如果客戶端使用零長度的客戶端標識符進行連接，則服務器必須使用包含分配的客戶端標識符的 CONNACK 進行響應。

與 MQTT 3.1.1 相比，這解除了服務器分配的客戶端 ID 只能用于連接的限制。Clean Session = 1

2. 實現(xiàn)

RabbitMQ 將生成一些隨機的客戶端 ID（例如 ），并在 CONNACK 數(shù)據包中返回它。dcGB2kSwS0JlXnaBa1A6QA

9. 主題別名

1. 描述

主題別名是一個整數(shù)值，用于標識主題，而不是使用主題名稱。 這減小了 PUBLISH 數(shù)據包的大小，并且在主題名稱很長且在網絡連接中重復使用相同的主題名稱時非常有用。

2. 實現(xiàn)

RabbitMQ 中的默認主題別名最大值為 16。 您可以在 中配置此值，例如：rabbitmq.conf

mqtt.topic_alias_maximum = 32

此配置值映射到從 RabbitMQ 發(fā)送到客戶端的 CONNACK 數(shù)據包中的 Topic Alias Maximum。 它限制了任一方向的主題別名數(shù)，即從客戶端到 RabbitMQ 和 RabbitMQ 到客戶端。 如果客戶端發(fā)送到許多不同的主題或從許多不同的主題接收，則設置更高的值將需要更多的內存使用量。

RabbitMQ 運算符可以通過設置以下設置來禁止使用主題別名：

mqtt.topic_alias_maximum = 0

10. 流量控制

1. 描述

MQTT 5.0 屬性 Receive Maximum 定義了未確認的 QoS 1 PUBLISH 數(shù)據包的上限。

2. 實現(xiàn)

從 RabbitMQ 發(fā)送到客戶端的未確認 QoS 1 PUBLISH 數(shù)據包的最大數(shù)量由 CONNECT 數(shù)據包中從客戶端發(fā)送到 RabbitMQ 的 Receive Maximum 和配置值：mqtt.prefetch

mqtt.prefetch = 10

默認值為 10。mqtt.prefetch

該值在 MQTT 3.1 和 3.1.1 的 RabbitMQ 3.13 之前已存在。 它映射到 RabbitMQ 中的使用者預取。 換句話說，它定義隊列發(fā)送到其 MQTT 連接進程的動態(tài)消息數(shù)量。mqtt.prefetch

11. 最大數(shù)據包大小

1. 描述

客戶端和服務器可以獨立指定它們支持的最大數(shù)據包大小。

2. 舉例

此示例演示如何限制從客戶端發(fā)送到 RabbitMQ 的最大 MQTT 數(shù)據包大小。

假設身份驗證成功后，RabbitMQ 操作員不希望 RabbitMQ 接受任何大于 1 KiB 的 MQTT 數(shù)據包。 將以下配置寫入 rabbitmq.conf（在當前工作目錄中）：

mqtt.max_packet_size_authenticated = 1024

停止 RabbitMQ 服務器后，啟動 RabbitMQ 服務器并應用新配置：

docker run -it --rm --name rabbitmq -p 1883:1883 -p 15672:15672 -p 15692:15692 \
    --mount type=bind,source="$(pwd)"/rabbitmq.conf,target=/etc/rabbitmq/conf.d/11-blog-post.conf \
    rabbitmq:3.13.0-beta.2-management
docker exec rabbitmq rabbitmq-plugins enable rabbitmq_mqtt
docker exec rabbitmq rabbitmqctl enable_feature_flag all

在第一個終端窗口中，訂閱一個主題：

mqttx sub --topic t/11

在第二個終端窗口中，向該主題發(fā)送有效負載為 3 字節(jié)的消息：

payload=$(head --bytes 3 < /dev/zero | tr '\0' x)
mqttx pub --topic t/11 -m "$payload"

第一行從特殊文件中讀取 3 個字節(jié)（3 個 null 字符），將每個 null 字符轉換為 ASCII 字符并將結果保存在變量中。/dev/zero``x``xxx``payload

第一個終端窗口將收到該消息：

payload: xxx

接下來，在第 2 個終端窗口中，發(fā)送有效負載為 2,000 字節(jié)的消息：

payload=$(head --bytes 2000 < /dev/zero | tr '\0' x)
mqttx pub --topic t/11 -m "$payload"

這一次，第一個終端窗口不會收到消息，因為從客戶端發(fā)送到 RabbitMQ 的 PUBLISH 數(shù)據包大于配置的最大數(shù)據包大小 1024 字節(jié)。

相反，RabbitMQ 會記錄一條描述性錯誤消息：

[error] <0.836.0> MQTT packet size (2007 bytes, type 3) exceeds mqtt.max_packet_size_authenticated (1024 bytes)

日志消息聲明 2,007 字節(jié)，因為 PUBLISH 數(shù)據包的固定和可變標頭需要 7 個字節(jié)（其中 4 個字節(jié)用于主題名稱）。t/11

12. 服務器啟動的斷開連接

1. 描述

在 MQTT 5.0 中，DISCONNECT 數(shù)據包不僅可以從客戶端發(fā)送到服務器，還可以從服務器發(fā)送到客戶端。

2. 實現(xiàn)

在終止連接之前，RabbitMQ 會在以下情況下向客戶端發(fā)送 DISCONNECT 數(shù)據包：

13. 會話到期

1. 描述

在 MQTT 5.0 中，客戶端可以在 CONNECT 數(shù)據包中向服務器建議會話到期間隔。 服務器可以接受建議的會話到期間隔，也可以在 CONNACK 數(shù)據包中強制要求不同的會話到期間隔。

會話可以跨一系列網絡連接繼續(xù)進行。它的持續(xù)時間與最新的網絡連接加上會話到期間隔一樣長。

當會話到期間隔到期時，客戶端和服務器都將刪除任何會話狀態(tài)。

2. 實現(xiàn)

只要會話持續(xù)，客戶端和服務器就會保持會話狀態(tài)。

服務器中的會話狀態(tài)包括已發(fā)送到客戶端但尚未確認的消息、待發(fā)送到客戶端的消息以及客戶端的訂閱。 RabbitMQ 以隊列和綁定的形式對這種 MQTT 會話狀態(tài)進行建模。

因此，會話到期間隔映射到 RabbitMQ 中的隊列 TTL。 當 MQTT 會話過期時，隊列及其消息和綁定將被刪除。

3. 舉例

默認情況下，服務器允許的最大會話到期間隔為 1 天。 如果 MQTT 客戶端在 1 天內沒有重新連接，則其會話狀態(tài)將在 RabbitMQ 中刪除。

此值是可配置的。 出于此示例的目的，讓我們在以下情況下設置一個非常低的會話到期間隔 1 分鐘：rabbitmq.conf

mqtt.max_session_expiry_interval_seconds = 60

設置名稱包含前綴，因為 MQTT 5.0 客戶端可以通過在 CONNECT 數(shù)據包中發(fā)送會話到期間隔來選擇較低的值。 如最大數(shù)據包大小示例中所做的那樣，重新啟動 RabbitMQ 節(jié)點，以便應用新設置。max

以 20 秒的會話到期間隔連接到 RabbitMQ 并創(chuàng)建訂閱：

mqttx sub --client-id sub-13 --topic t/13 --session-expiry-interval 20 --qos 1
…  Connecting...
?  Connected
…  Subscribing to t/13...
?  Subscribed to t/13
^C

鍵入終端以斷開客戶端連接。Ctrl+C

在接下來的 20 秒內，列出隊列和綁定：

docker exec rabbitmq rabbitmqctl list_queues name
Timeout: 60.0 seconds ...
Listing queues for vhost / ...
name
mqtt-subscription-sub-13qos1

docker exec rabbitmq rabbitmqctl list_bindings source_name destination_name routing_key --formatter=pretty_table
Listing bindings for vhost /...
┌─────────────┬──────────────────────────────┬──────────────────────────────┐
│ source_name │ destination_name             │ routing_key                  │
├─────────────┼──────────────────────────────┼──────────────────────────────┤
│             │ mqtt-subscription-sub-13qos1 │ mqtt-subscription-sub-13qos1 │
├─────────────┼──────────────────────────────┼──────────────────────────────┤
│ amq.topic   │ mqtt-subscription-sub-13qos1 │ t.13                         │
└─────────────┴──────────────────────────────┴──────────────────────────────┘

20 秒后，再次列出隊列和綁定：

docker exec rabbitmq rabbitmqctl list_queues name
Timeout: 60.0 seconds ...
Listing queues for vhost / ...

docker exec rabbitmq rabbitmqctl list_bindings source_name destination_name routing_key --formatter=pretty_table
Listing bindings for vhost /...

RabbitMQ 刪除了隊列及其綁定，因為我們的客戶端未在 20 秒的會話到期間隔內連接到 RabbitMQ。Clean Session = 0

接下來，執(zhí)行相同的測試，但會話到期間隔較長，例如 1 小時：

mqttx sub --client-id sub-13 --topic t/13 --session-expiry-interval 3600 --qos 1
…  Connecting...
?  Connected
…  Subscribing to t/13...
?  Subscribed to t/13
^C

您應該注意到，隊列及其綁定將在 1 分鐘后被刪除，因為有效的會話到期間隔 是客戶端請求的最小值（1 小時）和 RabbitMQ 中配置的值（1 分鐘）。mqtt.max_session_expiry_interval_seconds

14. 會延遲

1. 描述

客戶端可以在 CONNECT 數(shù)據包中定義 Will Delay Interval。

服務器會延遲發(fā)布客戶的遺囑消息，直到遺囑延遲間隔過去或會話結束，以先發(fā)生者為準。 如果在將延遲間隔過去之前與此會話建立了新的網絡連接，則服務器不得發(fā)送將消息。 這樣做的一個用途是，如果存在臨時網絡斷開連接，并且客戶端在發(fā)布遺囑消息之前成功重新連接并繼續(xù)其會話，則避免發(fā)布遺囑消息。

Will Delay Interval 的另一個用例是通知會話到期：

客戶端可以通過將 Will Delay Interval 設置為 Session Expiry Interval 長并發(fā)送帶有原因0x04代碼的 DISCONNECT（Disconnect with Will Message），來安排 Will Message 通知會話到期已發(fā)生。

2. 實現(xiàn)

盡管 will 消息有效負載通常很小，但 MQTT 規(guī)范允許 will 消息有效負載大小高達 64 KiB。

為了避免在 Khepri（RabbitMQ 未來的元數(shù)據存儲）中存儲大型二進制數(shù)據，RabbitMQ 創(chuàng)建了一個包含此單個遺囑消息的經典隊列。 我們稱此隊列為 Will 隊列。 此消息具有每條消息的 TTL 集，該集以毫秒為單位定義，對應于以秒為單位的 Will Delay Interval。 此外，Will 隊列還設置了一個隊列 TTL，該隊列以毫秒為單位定義，對應于以秒為單位的會話到期間隔。 每條消息的有效 TTL 至少比隊列 TTL 低幾毫秒，以便消息將在隊列（會話）過期前不久發(fā)布。

Will 隊列還定義（MQTT 插件使用的默認主題交換）為死信交換，將 will 主題定義為死信路由鍵。amq.topic

如果 MQTT 客戶端未在其 Will 延遲間隔內重新連接，則 Will 隊列中的消息將死信發(fā)送到主題交換。

讓我們用一個例子來說明這一點。

3. 舉例

在第一個終端窗口中，創(chuàng)建一個將使用遺囑消息的訂閱：

mqttx sub --client-id sub-14 --topic t/14

在第二個終端窗口中，創(chuàng)建一個 Will Delay Interval 為 20 秒的連接：

mqttx conn --client-id conn-14 --will-topic t/14 --will-message my-will-message --will-delay-interval 20 --session-expiry-interval 40

在第 3 個終端窗口中，我們看到到目前為止，訂閱 MQTT 客戶端創(chuàng)建了一個隊列：

docker exec rabbitmq rabbitmqctl --quiet --formatter=pretty_table list_queues name type messages arguments
┌──────────────────────────────┬────────────┬──────────┬───────────┐
│ name                         │ type       │ messages │ arguments │
├──────────────────────────────┼────────────┼──────────┼───────────┤
│ mqtt-subscription-sub-14qos0 │ MQTT QoS 0 │ 0        │           │
└──────────────────────────────┴────────────┴──────────┴───────────┘

在第二個終端窗口中，鍵入 to disconnect the MQTT connection with client ID。Ctrl+C``conn-14

這一次，列出隊列顯示已創(chuàng)建 Will 隊列：

docker exec rabbitmq rabbitmqctl --quiet --formatter=pretty_table list_queues name type messages arguments
┌──────────────────────────────┬────────────┬──────────┬────────────────────────────────────────────────────────────┐
│ name                         │ type       │ messages │ arguments                                                  │
├──────────────────────────────┼────────────┼──────────┼────────────────────────────────────────────────────────────┤
│ mqtt-subscription-sub-14qos0 │ MQTT QoS 0 │ 0        │                                                            │
├──────────────────────────────┼────────────┼──────────┼────────────────────────────────────────────────────────────┤
│ mqtt-will-conn-14            │ classic    │ 1        │ {<<"x-expires">>,long,40000}                               │
│                              │            │          │ {<<"x-dead-letter-exchange">>,longstr,<<"amq.topic">>}     │
│                              │            │          │ {<<"x-dead-letter-routing-key">>,longstr,<<"t.14">>}       │
└──────────────────────────────┴────────────┴──────────┴────────────────────────────────────────────────────────────┘

Will 隊列的命名模式為 。 它包含一條消息：遺囑消息。mqtt-will-<MQTT Client ID>

如上一節(jié)所述，隊列 TTL （） 為 40,000 毫秒，因此與上面命令中的 40 秒會話到期間隔匹配。 如果您等待 20 秒，您的第一個終端窗口應該會收到遺囑消息，因為我們的客戶沒有在遺囑延遲間隔內重新連接：x-expires

? payload: my-will-message

15. 可選的服務器功能可用性

1. 描述

定義一組服務器不允許的功能，并為服務器提供一種機制，以便將其指定給客戶端。 可以通過這種方式指定的功能包括：

• 最大 QoS
• 保留可用
• 提供通配符訂閱
• 可用的訂閱標識符
• 提供共享訂閱

客戶端使用服務器聲明不可用的功能是錯誤的。

2. 實現(xiàn)

RabbitMQ 3.13 在 CONNACK 屬性中包括 Maximum QoS = 1 和 Shared Subscription Available = 0。

RabbitMQ 不支持 QoS 2。

如下一節(jié)所述，將來的 RabbitMQ 版本將支持共享訂閱。

4. 局限性

本節(jié)列出了 RabbitMQ MQTT 實現(xiàn)的限制。

1. MQTT 5.0 特定限制

1. 共享訂閱

共享訂閱將在將來的 RabbitMQ 版本中添加。 盡管此功能很好地映射到 RabbitMQ 中的隊列，但共享訂閱是會話狀態(tài)的一部分，并且需要進行某些 RabbitMQ 數(shù)據庫遷移才能有效地查詢給定 MQTT 客戶端 ID 的共享訂閱。

1. 延遲和保留的遺囑消息

延遲和保留的遺囑信息將不會被保留。 這是因為延遲的遺囑消息將死信到主題交換，但保留進程當前不會從隊列中使用。 將來可以通過保留郵件的新存儲來解決此限制。

2. 非 MQTT 5.0 特定限制

為了完整起見，本節(jié)列出了在 RabbitMQ 3.13 中支持 MQTT 5.0 之前和 RabbitMQ 3.12 中提供原生 MQTT 之前存在的限制。

1. 保留的消息

保留消息的功能在 RabbitMQ 中受到限制。

保留的消息僅在本地節(jié)點上存儲和查詢。

一個有效的示例如下： MQTT 客戶端向節(jié)點 A 發(fā)布一條保留的消息，主題為 。此后，另一個客戶端在節(jié)點 A 上使用主題過濾器進行訂閱。新訂閱者將收到保留的消息。topic/1``topic/1

但是，如果主題篩選器包含通配符（多級通配符 “”或單級通配符 “”），則不會發(fā)送保留的消息 （問題 #8824）。#``+

此外，如果客戶端在節(jié)點 A 上發(fā)布了保留的消息，而另一個客戶端隨后在節(jié)點 B 上訂閱，則該訂閱客戶端將不會收到存儲在節(jié)點 A 上的任何保留消息（問題 #8096）。

將來的 RabbitMQ 版本將復制集群中保留的消息，并發(fā)送與包含通配符的主題篩選器匹配的保留消息。

5. 總結

綜上所述，RabbitMQ

• 是領先的 AMQP 0.9.1 代理
• 是一個流是代理
• 擅長跨協(xié)議互操作性
• 由于支持 3.13 中發(fā)布的 MQTT 5.0 和 3.12 中發(fā)布的原生 MQTT，它正在成為領先的 MQTT 代理之一

我們將 RabbitMQ 轉變?yōu)槌墒斓奈锫?lián)網代理的旅程尚未完成，并計劃在未來幾個月和幾年內進行更多的開發(fā)工作。 敬請關注！

6. 相關鏈接

MQTT 5.0 support is coming in RabbitMQ 3.13 | RabbitMQ文章來源地址http://www.zghlxwxcb.cn/news/detail-850265.html

到了這里，關于RabbitMQ3.13.0起支持MQTT5.0協(xié)議及MQTT5.0特性功能列表的文章就介紹完了。如果您還想了解更多內容，請在右上角搜索TOY模板網以前的文章或繼續(xù)瀏覽下面的相關文章，希望大家以后多多支持TOY模板網！