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