后端物聯(lián)網(wǎng)消息隊列通信【免費下載鏈接】emqxThe most scalable and reliable MQTT broker for AI, IoT, IIoT and connected vehicles項目地址https://gitcode.com/gh_mirrors/em/emqx點擊查看免費下載本文基于倉庫變更記錄 changes/ee/fix-16999.en.md 展開深入剖析 EMQX 在啟用 Message QueueMQ特性的遠端 broker 上MQTT 橋接MQTT Bridge作為 MQTT Source 從$queue/主題訂閱時無法收到消息的問題根因與修復方案。讀者讀完本文后將掌握 MQTT v5 Subscription-Identifier 屬性在 EMQX 訂閱、投遞、橋接路由三個環(huán)節(jié)的完整流轉(zhuǎn)機制理解為什么$queue/隊列訂閱強制要求 MQTT v5 連接并能通過源碼與測試用例驗證該修復的可靠性。背景為什么 MQTT 橋接需要 Subscription-IdentifierEMQX 的 MQTT 橋接emqx_bridge_mqtt在作為 MQTT Sourceingress運行時會通過一條 MQTT 客戶端連接訂閱遠端 broker 的主題把收到的消息導入本地 broker 或觸發(fā)規(guī)則引擎處理。當遠端 broker 同時開啟了 EMQX 的 Message QueueMQ特性時客戶端需要訂閱形如$queue/queue-name/topic-filter的主題才能消費隊列中的消息。MQTT v5 協(xié)議中客戶端可以在 SUBSCRIBE 包的屬性中攜帶一個或多個Subscription-Identifier訂閱標識符服務端在向該訂閱投遞消息時會把對應的訂閱標識符原樣寫入 PUBLISH 包的屬性中。這讓訂閱方無需在客戶端維護主題過濾器到業(yè)務處理的映射即可根據(jù) PUBLISH 包屬性快速識別這條消息來自哪一條訂閱。EMQX MQTT 橋接的 ingress 正是依賴這一機制完成消息路由的橋接啟動時為每條 ingress 訂閱分配唯一的 Subscription-Identifier訂閱時在 SUBSCRIBE 屬性中帶上該標識符收到消息后優(yōu)先根據(jù) PUBLISH 包中的 Subscription-Identifier 查索引定位對應的通道配置channel config只有找不到時才退化為按主題匹配。如果遠端 broker 在投遞 MQ 隊列消息時沒有在 PUBLISH 包中攜帶 Subscription-Identifieringress 就無法通過標識符定位到正確的通道消息隨之丟失——這正是本次修復要解決的核心問題。問題描述與根因定位根據(jù) changes/ee/fix-16999.en.md 的記載Fixed an issue where MQTT source failed to receive messages from$queue/subscriptions when the remote broker has the Message Queue (mq) feature enabled.問題現(xiàn)象遠端 broker 開啟 MQ 特性時MQTT 橋接 Source 從$queue/訂閱收不到消息。根因MQ 消息投遞時PUBLISH 包中沒有包含 MQTT v5 Subscription-Identifier 屬性而 MQTT 橋接 ingress 正是依賴該屬性來路由來自隊列訂閱的消息。從源碼層面可以還原這條因果鏈橋接 ingress 在訂閱遠端主題時會攜帶Subscription-Identifier屬性見 emqx_bridge_mqtt_ingress.erlsubscribe_properties(#{subscription_id : SubscriptionId}) - #{Subscription-Identifier SubscriptionId}; subscribe_properties(_Ingress) - #{}.消息到達 ingress 后先按 Subscription-Identifier 查找通道配置找不到才按主題回退匹配見 emqx_bridge_mqtt_ingress.erlfind_channel_configs(Topic, Props, SubscriptionIdToHandlerIndex, TopicToHandlerIndex) - case find_channel_configs_by_subscription_identifier(Props, SubscriptionIdToHandlerIndex) of [] - find_channel_configs_by_topic(Topic, TopicToHandlerIndex); ChannelConfigs - ChannelConfigs end.修復前遠端 broker 的 MQ 投遞鏈路在生成 PUBLISH 包時未注入Subscription-Identifier屬性導致第 2 步永遠命中按標識符查找的空結(jié)果消息被靜默丟棄。修復方案MQ 投遞鏈路補全 Subscription-Identifier修復的實質(zhì)是在 MQ 的消息投遞路徑上補齊 Subscription-Identifier 的注入同時完善橋接 ingress 側(cè)的訂閱協(xié)商與強制校驗邏輯。整個修復可以拆解為三個層面。1. Broker 側(cè)訂閱解析與投遞注入MQTT v5 通用能力EMQX 本身作為 MQTT broker 時對 Subscription-Identifier 的支持分為訂閱與投遞兩步這也是遠端 broker 正確投遞隊列消息的前提。訂閱階段emqx_channel在處理 SUBSCRIBE 包時把屬性中的Subscription-Identifier提取為subid存入訂閱選項見 emqx_channel.erlenrich_subopts_subid(TopicFilters, #{sub_props : #{Subscription-Identifier : SubId}}) - [{Topic, SubOpts#{subid SubId}} || {Topic, SubOpts} - TopicFilters]; enrich_subopts_subid(TopicFilters, _State) - TopicFilters.同時EMQX 在 CONNACK 屬性中聲明Subscription-Identifier-Available 1見 emqx_channel.erl明確告知客戶端本服務端支持該特性。投遞階段emqx_session的enrich_message/3在把消息投遞給訂閱者時若該訂閱帶有subid則在消息頭 properties 中注入Subscription-Identifier見 emqx_session.erl%% Add Subscription-Identifier if its specified for the %% subscription: Headers case SubId of undefined - Headers0; _ - case Headers0 of #{properties : Properties} - Headers0#{properties : Properties#{Subscription-Identifier SubId}}; #{} - Headers0#{properties #{Subscription-Identifier SubId}} end end,該邏輯有對應的單元測試佐證見 emqx_session.erl例如enrich_subid1_test/0驗證subid 42的訂閱會把Subscription-Identifier : 42寫入消息屬性。2. MQ 消費鏈路隊列消息經(jīng) broker 投遞時保留訂閱標識符MQ 消息從入隊到出隊再進入 broker 投遞管線的路徑大致如下emqx_mq模塊掛載message.publishhook見 emqx_mq.erl攔截發(fā)往 MQ 主題的消息通過publish_to_queue/2調(diào)用emqx_mq_message_db:insert/2寫入消息數(shù)據(jù)庫見 emqx_mq.erlMQ 消費者側(cè)把出隊消息通過emqx_mq_sub:messages/3送回訂閱端見 emqx_mq_sub.erl跨節(jié)點時走 RPC見 emqx_mq_sub_proto_v1.erl消息最終經(jīng)擴展訂閱處理器emqx_mq_extsub_handler進入 broker 的投遞流程見 emqx_mq_extsub_handler.erl。MQ 投遞鏈路此前在消息進入 broker 投遞管線前對訂閱標識符的保留不完整導致最終生成的 PUBLISH 包缺失該屬性。修復后隊列消息在被投遞給$queue/訂閱者時會走與普通訂閱一致的enrich_message/3注入邏輯從而在 PUBLISH 包中攜帶訂閱時的 Subscription-Identifier。橋接 ingress 收到后即可按標識符完成路由。3. 橋接 ingress 側(cè)訂閱協(xié)商與強制校驗橋接側(cè)在本次修復中同步完善了與 Subscription-Identifier 相關(guān)的訂閱行為涉及 emqx_bridge_mqtt_ingress.erl 與 emqx_bridge_mqtt_connector.erl 兩個模塊。訂閱標識符分配每條 ingress 訂閱會被分配一個唯一標識符最大取值為?MAX_SUBSCRIPTION_ID定義為268435455即 MQTT v5 中 Subscription-Identifier 的最大合法值 2^28?1見 emqx_bridge_mqtt_ingress.erl。分配邏輯會跳過已被占用的編號編號耗盡時報no_available_subscription_id見 emqx_bridge_mqtt_ingress.erl。降級重試協(xié)商訂閱時若遠端 broker 返回SUBSCRIPTION_IDENTIFIERS_NOT_SUPPORTED原因碼行為分兩種見 emqx_bridge_mqtt_ingress.erl普通主題訂閱自動降級去掉 Subscription-Identifier 屬性后重新訂閱保證普通場景的兼容性$queue/隊列訂閱不降級直接返回錯誤subscription_identifier_required_for_queue_subscription。因為隊列消息的路由完全依賴該屬性降級后必然導致消息丟失顯式報錯比靜默丟消息更安全。maybe_retry_without_subscription_identifier( #{subscription_id : _SubscriptionId, remote : #{topic : $queue/, _/binary}}, {ok, _Props, ReasonCodes} Result, _Pid ) - case lists:member(?RC_SUBSCRIPTION_IDENTIFIERS_NOT_SUPPORTED, ReasonCodes) of true - {error, subscription_identifier_required_for_queue_subscription}; false - Result end;啟動期強制校驗connector 在創(chuàng)建 ingress 配置時調(diào)用ensure_queue_subscription_supported/2若發(fā)現(xiàn)配置了$queue/主題但協(xié)議不是 MQTT v5不支持 Subscription-Identifier直接拋出不可恢復錯誤見 emqx_bridge_mqtt_connector.erlensure_queue_subscription_supported(#{topic : Topic}, SubscriptionIdToHandlerIndex) - case is_queue_topic(Topic) andalso not supports_queue_subscription(SubscriptionIdToHandlerIndex) of false - ok; true - error({unrecoverable_error, subscription_identifier_required_for_queue_subscription}) end.其中 ETS 訂閱標識符索引僅在proto_ver : v5時創(chuàng)建見 emqx_bridge_mqtt_connector.erl與上面的強制校驗形成閉環(huán)。消息到達后的路由ingress 收到 PUBLISH 包后解析屬性中的Subscription-Identifier支持單個整數(shù)或多個標識符的列表見 emqx_bridge_mqtt_ingress.erl到 ETS 索引中查找對應的通道配置find_channel_configs_by_subscription_identifier(Props, SubscriptionIdToHandlerIndex) - lists:flatmap( fun(SubscriptionId) - case ets:lookup(SubscriptionIdToHandlerIndex, SubscriptionId) of [{SubscriptionId, ChannelConfig}] - [ChannelConfig]; [] - [] end end, subscription_ids(Props) ).配置要求與限制綜合上述代碼邏輯使用 MQTT 橋接 Source 訂閱遠端 broker 的$queue/主題時需要滿足以下條件遠端與本地連接均須使用 MQTT v5connector 配置中proto_ver v5Subscription-Identifier 是 MQTT v5 特性emqx_bridge_mqtt_connector.erl僅在 v5 下創(chuàng)建標識符索引且$queue/訂閱在不支持該特性時會啟動失敗并報subscription_identifier_required_for_queue_subscription遠端 broker 須正確實現(xiàn) MQTT v5 訂閱標識符語義訂閱端攜帶的 Subscription-Identifier 必須原樣出現(xiàn)在投遞的 PUBLISH 包屬性中。EMQX 自身通過 emqx_session.erl 的注入邏輯滿足該要求這也是本次修復的核心訂閱標識符數(shù)量有上限每條 ingress 訂閱占用一個標識符最大編號為 268435455實際可分配數(shù)量受?MAX_SUBSCRIPTION_ID約束。測試驗證本次修復的各個環(huán)節(jié)均有對應測試用例覆蓋可作為回歸驗證的入口。MQ 側(cè)保留驗證emqx_mq_SUITE.erl 中的t_subscription_identifier/1用例完整模擬了問題場景創(chuàng)建名為subid的非 lastvalue 隊列主題過濾器為t/#用SubId 42、屬性#{Subscription-Identifier 42}訂閱$queue/subid/t/#向t/0發(fā)布消息斷言收到的 PUBLISH 包中properties含Subscription-Identifier : 42否則ct:fail(t/0 message from MQ not received or missing Subscription-Identifier)。該用例直接驗證了修復目標MQ 隊列消息投遞時 PUBLISH 包必須攜帶訂閱標識符。橋接側(cè)路由與協(xié)商驗證emqx_bridge_mqtt_source_SUITE.erl 中針對該場景有以下用例t_mqtt_conn_bridge_ingress_subid_dispatch第 486 行起文檔注釋為 MQTT ingress dispatches queue deliveries by Subscription Identifier.覆蓋遠端$queue/orders/t/#場景下按訂閱標識符分發(fā)隊列投遞t_mqtt_conn_bridge_rejects_queue_source_without_subscription_identifier第 523 行起文檔注釋為 MQTT ingress rejects queue subscriptions when Subscription Identifiers are unavailable.斷言報錯原因字符串為queue subscriptions require connector proto_ver v5驗證了啟動期強制校驗t_mqtt_conn_bridge_ingress_retries_without_subid_on_a1第 543 行起驗證遠端不支持訂閱標識符時普通訂閱先發(fā)帶 subid 的 SUBSCRIBE、收到不支持響應后自動降級重發(fā)不帶 subid 的 SUBSCRIBE第 597-598 行分別斷言subscribe_with_subid與subscribe_without_subid消息。結(jié)語本次修復changes/ee/fix-16999.en.md從根因上補齊了 MQ 投遞鏈路中 PUBLISH 包缺失 Subscription-Identifier 的問題使 MQTT 橋接 Source 在遠端 broker 開啟 MQ 特性時能夠正常從$queue/訂閱接收消息。修復同時完善了三層防護MQ 投遞側(cè)正確注入訂閱標識符、橋接 ingress 側(cè)按標識符優(yōu)先路由、以及隊列訂閱對不支持 MQTT v5 的遠端顯式報錯而非靜默丟消息。對于運維與集成人員核心要點是只要橋接遠端涉及$queue/隊列訂閱連接協(xié)議必須配置為 MQTT v5且遠端 broker 需完整支持 Subscription-Identifier 語義。贊分享后端物聯(lián)網(wǎng)消息隊列通信【免費下載鏈接】emqxThe most scalable and reliable MQTT broker for AI, IoT, IIoT and connected vehicles項目地址https://gitcode.com/gh_mirrors/em/emqx點擊查看免費下載相關(guān)推薦EMQX 消息流Streams投遞修復解析Subscription Identifier 等訂閱選項的完整傳遞鏈路EMQX 消息流Streams投遞修復解析Subscription Identifier 等訂閱選項的完整傳遞鏈路 本指南圍繞變更記錄 changes/e后端物聯(lián)網(wǎng)消息隊列通信EMQX 修復解析Message Queue 啟用時 $queue/ 訂閱無法向 MQTT 橋接投遞消息的問題EMQX 修復解析Message Queue 啟用時 $queue/ 訂閱無法向 MQTT 橋接投遞消息的問題 本文基于倉庫變更記錄 changes/ee/f后端物聯(lián)網(wǎng)消息隊列通信EMQX MQTT Ingress 橋接隊列訂閱$queue與 MQTT 5 Subscription Identifiers 深度解析EMQX MQTT Ingress 橋接隊列訂閱 $queue 與 MQTT 5 Subscription Identifiers 深度解析 MQTT in后端物聯(lián)網(wǎng)消息隊列通信創(chuàng)作聲明：本文部分內(nèi)容由AI輔助生成（AIGC），僅供參考 EMQX MQTT 橋接 `$queue/` 訂閱消息丟失修復：Subscription-Identifier 在 MQ 投遞鏈路中的完整解析
后端物聯(lián)網(wǎng)消息隊列通信【免費下載鏈接】emqxThe most scalable and reliable MQTT broker for AI, IoT, IIoT and connected vehicles項目地址https://gitcode.com/gh_mirrors/em/emqx點擊查看免費下載本文基于倉庫變更記錄 changes/ee/fix-16999.en.md 展開深入剖析 EMQX 在啟用 Message QueueMQ特性的遠端 broker 上MQTT 橋接MQTT Bridge作為 MQTT Source 從$queue/主題訂閱時無法收到消息的問題根因與修復方案。讀者讀完本文后將掌握 MQTT v5 Subscription-Identifier 屬性在 EMQX 訂閱、投遞、橋接路由三個環(huán)節(jié)的完整流轉(zhuǎn)機制理解為什么$queue/隊列訂閱強制要求 MQTT v5 連接并能通過源碼與測試用例驗證該修復的可靠性。背景為什么 MQTT 橋接需要 Subscription-IdentifierEMQX 的 MQTT 橋接emqx_bridge_mqtt在作為 MQTT Sourceingress運行時會通過一條 MQTT 客戶端連接訂閱遠端 broker 的主題把收到的消息導入本地 broker 或觸發(fā)規(guī)則引擎處理。當遠端 broker 同時開啟了 EMQX 的 Message QueueMQ特性時客戶端需要訂閱形如$queue/queue-name/topic-filter的主題才能消費隊列中的消息。MQTT v5 協(xié)議中客戶端可以在 SUBSCRIBE 包的屬性中攜帶一個或多個Subscription-Identifier訂閱標識符服務端在向該訂閱投遞消息時會把對應的訂閱標識符原樣寫入 PUBLISH 包的屬性中。這讓訂閱方無需在客戶端維護主題過濾器到業(yè)務處理的映射即可根據(jù) PUBLISH 包屬性快速識別這條消息來自哪一條訂閱。EMQX MQTT 橋接的 ingress 正是依賴這一機制完成消息路由的橋接啟動時為每條 ingress 訂閱分配唯一的 Subscription-Identifier訂閱時在 SUBSCRIBE 屬性中帶上該標識符收到消息后優(yōu)先根據(jù) PUBLISH 包中的 Subscription-Identifier 查索引定位對應的通道配置channel config只有找不到時才退化為按主題匹配。如果遠端 broker 在投遞 MQ 隊列消息時沒有在 PUBLISH 包中攜帶 Subscription-Identifieringress 就無法通過標識符定位到正確的通道消息隨之丟失——這正是本次修復要解決的核心問題。問題描述與根因定位根據(jù) changes/ee/fix-16999.en.md 的記載Fixed an issue where MQTT source failed to receive messages from$queue/subscriptions when the remote broker has the Message Queue (mq) feature enabled.問題現(xiàn)象遠端 broker 開啟 MQ 特性時MQTT 橋接 Source 從$queue/訂閱收不到消息。根因MQ 消息投遞時PUBLISH 包中沒有包含 MQTT v5 Subscription-Identifier 屬性而 MQTT 橋接 ingress 正是依賴該屬性來路由來自隊列訂閱的消息。從源碼層面可以還原這條因果鏈橋接 ingress 在訂閱遠端主題時會攜帶Subscription-Identifier屬性見 emqx_bridge_mqtt_ingress.erlsubscribe_properties(#{subscription_id : SubscriptionId}) - #{Subscription-Identifier SubscriptionId}; subscribe_properties(_Ingress) - #{}.消息到達 ingress 后先按 Subscription-Identifier 查找通道配置找不到才按主題回退匹配見 emqx_bridge_mqtt_ingress.erlfind_channel_configs(Topic, Props, SubscriptionIdToHandlerIndex, TopicToHandlerIndex) - case find_channel_configs_by_subscription_identifier(Props, SubscriptionIdToHandlerIndex) of [] - find_channel_configs_by_topic(Topic, TopicToHandlerIndex); ChannelConfigs - ChannelConfigs end.修復前遠端 broker 的 MQ 投遞鏈路在生成 PUBLISH 包時未注入Subscription-Identifier屬性導致第 2 步永遠命中按標識符查找的空結(jié)果消息被靜默丟棄。修復方案MQ 投遞鏈路補全 Subscription-Identifier修復的實質(zhì)是在 MQ 的消息投遞路徑上補齊 Subscription-Identifier 的注入同時完善橋接 ingress 側(cè)的訂閱協(xié)商與強制校驗邏輯。整個修復可以拆解為三個層面。1. Broker 側(cè)訂閱解析與投遞注入MQTT v5 通用能力EMQX 本身作為 MQTT broker 時對 Subscription-Identifier 的支持分為訂閱與投遞兩步這也是遠端 broker 正確投遞隊列消息的前提。訂閱階段emqx_channel在處理 SUBSCRIBE 包時把屬性中的Subscription-Identifier提取為subid存入訂閱選項見 emqx_channel.erlenrich_subopts_subid(TopicFilters, #{sub_props : #{Subscription-Identifier : SubId}}) - [{Topic, SubOpts#{subid SubId}} || {Topic, SubOpts} - TopicFilters]; enrich_subopts_subid(TopicFilters, _State) - TopicFilters.同時EMQX 在 CONNACK 屬性中聲明Subscription-Identifier-Available 1見 emqx_channel.erl明確告知客戶端本服務端支持該特性。投遞階段emqx_session的enrich_message/3在把消息投遞給訂閱者時若該訂閱帶有subid則在消息頭 properties 中注入Subscription-Identifier見 emqx_session.erl%% Add Subscription-Identifier if its specified for the %% subscription: Headers case SubId of undefined - Headers0; _ - case Headers0 of #{properties : Properties} - Headers0#{properties : Properties#{Subscription-Identifier SubId}}; #{} - Headers0#{properties #{Subscription-Identifier SubId}} end end,該邏輯有對應的單元測試佐證見 emqx_session.erl例如enrich_subid1_test/0驗證subid 42的訂閱會把Subscription-Identifier : 42寫入消息屬性。2. MQ 消費鏈路隊列消息經(jīng) broker 投遞時保留訂閱標識符MQ 消息從入隊到出隊再進入 broker 投遞管線的路徑大致如下emqx_mq模塊掛載message.publishhook見 emqx_mq.erl攔截發(fā)往 MQ 主題的消息通過publish_to_queue/2調(diào)用emqx_mq_message_db:insert/2寫入消息數(shù)據(jù)庫見 emqx_mq.erlMQ 消費者側(cè)把出隊消息通過emqx_mq_sub:messages/3送回訂閱端見 emqx_mq_sub.erl跨節(jié)點時走 RPC見 emqx_mq_sub_proto_v1.erl消息最終經(jīng)擴展訂閱處理器emqx_mq_extsub_handler進入 broker 的投遞流程見 emqx_mq_extsub_handler.erl。MQ 投遞鏈路此前在消息進入 broker 投遞管線前對訂閱標識符的保留不完整導致最終生成的 PUBLISH 包缺失該屬性。修復后隊列消息在被投遞給$queue/訂閱者時會走與普通訂閱一致的enrich_message/3注入邏輯從而在 PUBLISH 包中攜帶訂閱時的 Subscription-Identifier。橋接 ingress 收到后即可按標識符完成路由。3. 橋接 ingress 側(cè)訂閱協(xié)商與強制校驗橋接側(cè)在本次修復中同步完善了與 Subscription-Identifier 相關(guān)的訂閱行為涉及 emqx_bridge_mqtt_ingress.erl 與 emqx_bridge_mqtt_connector.erl 兩個模塊。訂閱標識符分配每條 ingress 訂閱會被分配一個唯一標識符最大取值為?MAX_SUBSCRIPTION_ID定義為268435455即 MQTT v5 中 Subscription-Identifier 的最大合法值 2^28?1見 emqx_bridge_mqtt_ingress.erl。分配邏輯會跳過已被占用的編號編號耗盡時報no_available_subscription_id見 emqx_bridge_mqtt_ingress.erl。降級重試協(xié)商訂閱時若遠端 broker 返回SUBSCRIPTION_IDENTIFIERS_NOT_SUPPORTED原因碼行為分兩種見 emqx_bridge_mqtt_ingress.erl普通主題訂閱自動降級去掉 Subscription-Identifier 屬性后重新訂閱保證普通場景的兼容性$queue/隊列訂閱不降級直接返回錯誤subscription_identifier_required_for_queue_subscription。因為隊列消息的路由完全依賴該屬性降級后必然導致消息丟失顯式報錯比靜默丟消息更安全。maybe_retry_without_subscription_identifier( #{subscription_id : _SubscriptionId, remote : #{topic : $queue/, _/binary}}, {ok, _Props, ReasonCodes} Result, _Pid ) - case lists:member(?RC_SUBSCRIPTION_IDENTIFIERS_NOT_SUPPORTED, ReasonCodes) of true - {error, subscription_identifier_required_for_queue_subscription}; false - Result end;啟動期強制校驗connector 在創(chuàng)建 ingress 配置時調(diào)用ensure_queue_subscription_supported/2若發(fā)現(xiàn)配置了$queue/主題但協(xié)議不是 MQTT v5不支持 Subscription-Identifier直接拋出不可恢復錯誤見 emqx_bridge_mqtt_connector.erlensure_queue_subscription_supported(#{topic : Topic}, SubscriptionIdToHandlerIndex) - case is_queue_topic(Topic) andalso not supports_queue_subscription(SubscriptionIdToHandlerIndex) of false - ok; true - error({unrecoverable_error, subscription_identifier_required_for_queue_subscription}) end.其中 ETS 訂閱標識符索引僅在proto_ver : v5時創(chuàng)建見 emqx_bridge_mqtt_connector.erl與上面的強制校驗形成閉環(huán)。消息到達后的路由ingress 收到 PUBLISH 包后解析屬性中的Subscription-Identifier支持單個整數(shù)或多個標識符的列表見 emqx_bridge_mqtt_ingress.erl到 ETS 索引中查找對應的通道配置find_channel_configs_by_subscription_identifier(Props, SubscriptionIdToHandlerIndex) - lists:flatmap( fun(SubscriptionId) - case ets:lookup(SubscriptionIdToHandlerIndex, SubscriptionId) of [{SubscriptionId, ChannelConfig}] - [ChannelConfig]; [] - [] end end, subscription_ids(Props) ).配置要求與限制綜合上述代碼邏輯使用 MQTT 橋接 Source 訂閱遠端 broker 的$queue/主題時需要滿足以下條件遠端與本地連接均須使用 MQTT v5connector 配置中proto_ver v5Subscription-Identifier 是 MQTT v5 特性emqx_bridge_mqtt_connector.erl僅在 v5 下創(chuàng)建標識符索引且$queue/訂閱在不支持該特性時會啟動失敗并報subscription_identifier_required_for_queue_subscription遠端 broker 須正確實現(xiàn) MQTT v5 訂閱標識符語義訂閱端攜帶的 Subscription-Identifier 必須原樣出現(xiàn)在投遞的 PUBLISH 包屬性中。EMQX 自身通過 emqx_session.erl 的注入邏輯滿足該要求這也是本次修復的核心訂閱標識符數(shù)量有上限每條 ingress 訂閱占用一個標識符最大編號為 268435455實際可分配數(shù)量受?MAX_SUBSCRIPTION_ID約束。測試驗證本次修復的各個環(huán)節(jié)均有對應測試用例覆蓋可作為回歸驗證的入口。MQ 側(cè)保留驗證emqx_mq_SUITE.erl 中的t_subscription_identifier/1用例完整模擬了問題場景創(chuàng)建名為subid的非 lastvalue 隊列主題過濾器為t/#用SubId 42、屬性#{Subscription-Identifier 42}訂閱$queue/subid/t/#向t/0發(fā)布消息斷言收到的 PUBLISH 包中properties含Subscription-Identifier : 42否則ct:fail(t/0 message from MQ not received or missing Subscription-Identifier)。該用例直接驗證了修復目標MQ 隊列消息投遞時 PUBLISH 包必須攜帶訂閱標識符。橋接側(cè)路由與協(xié)商驗證emqx_bridge_mqtt_source_SUITE.erl 中針對該場景有以下用例t_mqtt_conn_bridge_ingress_subid_dispatch第 486 行起文檔注釋為 MQTT ingress dispatches queue deliveries by Subscription Identifier.覆蓋遠端$queue/orders/t/#場景下按訂閱標識符分發(fā)隊列投遞t_mqtt_conn_bridge_rejects_queue_source_without_subscription_identifier第 523 行起文檔注釋為 MQTT ingress rejects queue subscriptions when Subscription Identifiers are unavailable.斷言報錯原因字符串為queue subscriptions require connector proto_ver v5驗證了啟動期強制校驗t_mqtt_conn_bridge_ingress_retries_without_subid_on_a1第 543 行起驗證遠端不支持訂閱標識符時普通訂閱先發(fā)帶 subid 的 SUBSCRIBE、收到不支持響應后自動降級重發(fā)不帶 subid 的 SUBSCRIBE第 597-598 行分別斷言subscribe_with_subid與subscribe_without_subid消息。結(jié)語本次修復changes/ee/fix-16999.en.md從根因上補齊了 MQ 投遞鏈路中 PUBLISH 包缺失 Subscription-Identifier 的問題使 MQTT 橋接 Source 在遠端 broker 開啟 MQ 特性時能夠正常從$queue/訂閱接收消息。修復同時完善了三層防護MQ 投遞側(cè)正確注入訂閱標識符、橋接 ingress 側(cè)按標識符優(yōu)先路由、以及隊列訂閱對不支持 MQTT v5 的遠端顯式報錯而非靜默丟消息。對于運維與集成人員核心要點是只要橋接遠端涉及$queue/隊列訂閱連接協(xié)議必須配置為 MQTT v5且遠端 broker 需完整支持 Subscription-Identifier 語義。贊分享后端物聯(lián)網(wǎng)消息隊列通信【免費下載鏈接】emqxThe most scalable and reliable MQTT broker for AI, IoT, IIoT and connected vehicles項目地址https://gitcode.com/gh_mirrors/em/emqx點擊查看免費下載相關(guān)推薦EMQX 消息流Streams投遞修復解析Subscription Identifier 等訂閱選項的完整傳遞鏈路EMQX 消息流Streams投遞修復解析Subscription Identifier 等訂閱選項的完整傳遞鏈路 本指南圍繞變更記錄 changes/e后端物聯(lián)網(wǎng)消息隊列通信EMQX 修復解析Message Queue 啟用時 $queue/ 訂閱無法向 MQTT 橋接投遞消息的問題EMQX 修復解析Message Queue 啟用時 $queue/ 訂閱無法向 MQTT 橋接投遞消息的問題 本文基于倉庫變更記錄 changes/ee/f后端物聯(lián)網(wǎng)消息隊列通信EMQX MQTT Ingress 橋接隊列訂閱$queue與 MQTT 5 Subscription Identifiers 深度解析EMQX MQTT Ingress 橋接隊列訂閱 $queue 與 MQTT 5 Subscription Identifiers 深度解析 MQTT in后端物聯(lián)網(wǎng)消息隊列通信創(chuàng)作聲明：本文部分內(nèi)容由AI輔助生成（AIGC），僅供參考