：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/主題訂閱時無法收到消息的問題根因與修復(fù)方案。讀者讀完本文后將掌握 MQTT v5 Subscription-Identifier 屬性在 EMQX 訂閱、投遞、橋接路由三個環(huán)節(jié)的完整流轉(zhuǎn)機制理解為什么$queue/隊列訂閱強制要求 MQTT v5 連接并能通過源碼與測試用例驗證該修復(fù)的可靠性。背景為什么 MQTT 橋接需要 Subscription-IdentifierEMQX 的 MQTT 橋接emqx_bridge_mqtt在作為 MQTT Sourceingress運行時會通過一條 MQTT 客戶端連接訂閱遠端 broker 的主題把收到的消息導(dǎo)入本地 broker 或觸發(fā)規(guī)則引擎處理。當(dāng)遠端 broker 同時開啟了 EMQX 的 Message QueueMQ特性時客戶端需要訂閱形如$queue/queue-name/topic-filter的主題才能消費隊列中的消息。MQTT v5 協(xié)議中客戶端可以在 SUBSCRIBE 包的屬性中攜帶一個或多個Subscription-Identifier訂閱標(biāo)識符服務(wù)端在向該訂閱投遞消息時會把對應(yīng)的訂閱標(biāo)識符原樣寫入 PUBLISH 包的屬性中。這讓訂閱方無需在客戶端維護主題過濾器到業(yè)務(wù)處理的映射即可根據(jù) PUBLISH 包屬性快速識別這條消息來自哪一條訂閱。EMQX MQTT 橋接的 ingress 正是依賴這一機制完成消息路由的橋接啟動時為每條 ingress 訂閱分配唯一的 Subscription-Identifier訂閱時在 SUBSCRIBE 屬性中帶上該標(biāo)識符收到消息后優(yōu)先根據(jù) PUBLISH 包中的 Subscription-Identifier 查索引定位對應(yīng)的通道配置channel config只有找不到時才退化為按主題匹配。如果遠端 broker 在投遞 MQ 隊列消息時沒有在 PUBLISH 包中攜帶 Subscription-Identifieringress 就無法通過標(biāo)識符定位到正確的通道消息隨之丟失——這正是本次修復(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)象遠端 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.修復(fù)前遠端 broker 的 MQ 投遞鏈路在生成 PUBLISH 包時未注入Subscription-Identifier屬性導(dǎo)致第 2 步永遠命中按標(biāo)識符查找的空結(jié)果消息被靜默丟棄。修復(fù)方案MQ 投遞鏈路補全 Subscription-Identifier修復(fù)的實質(zhì)是在 MQ 的消息投遞路徑上補齊 Subscription-Identifier 的注入同時完善橋接 ingress 側(cè)的訂閱協(xié)商與強制校驗邏輯。整個修復(fù)可以拆解為三個層面。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明確告知客戶端本服務(wù)端支持該特性。投遞階段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,該邏輯有對應(yīng)的單元測試佐證見 emqx_session.erl例如enrich_subid1_test/0驗證subid 42的訂閱會把Subscription-Identifier : 42寫入消息屬性。2. MQ 消費鏈路隊列消息經(jīng) broker 投遞時保留訂閱標(biāo)識符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 投遞管線前對訂閱標(biāo)識符的保留不完整導(dǎo)致最終生成的 PUBLISH 包缺失該屬性。修復(fù)后隊列消息在被投遞給$queue/訂閱者時會走與普通訂閱一致的enrich_message/3注入邏輯從而在 PUBLISH 包中攜帶訂閱時的 Subscription-Identifier。橋接 ingress 收到后即可按標(biāo)識符完成路由。3. 橋接 ingress 側(cè)訂閱協(xié)商與強制校驗橋接側(cè)在本次修復(fù)中同步完善了與 Subscription-Identifier 相關(guān)的訂閱行為涉及 emqx_bridge_mqtt_ingress.erl 與 emqx_bridge_mqtt_connector.erl 兩個模塊。訂閱標(biāo)識符分配每條 ingress 訂閱會被分配一個唯一標(biāo)識符最大取值為?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。因為隊列消息的路由完全依賴該屬性降級后必然導(dǎo)致消息丟失顯式報錯比靜默丟消息更安全。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直接拋出不可恢復(fù)錯誤見 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)識符索引僅在proto_ver : v5時創(chuàng)建見 emqx_bridge_mqtt_connector.erl與上面的強制校驗形成閉環(huán)。消息到達后的路由ingress 收到 PUBLISH 包后解析屬性中的Subscription-Identifier支持單個整數(shù)或多個標(biāo)識符的列表見 emqx_bridge_mqtt_ingress.erl到 ETS 索引中查找對應(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 訂閱遠端 broker 的$queue/主題時需要滿足以下條件遠端與本地連接均須使用 MQTT v5connector 配置中proto_ver v5Subscription-Identifier 是 MQTT v5 特性emqx_bridge_mqtt_connector.erl僅在 v5 下創(chuàng)建標(biāo)識符索引且$queue/訂閱在不支持該特性時會啟動失敗并報subscription_identifier_required_for_queue_subscription遠端 broker 須正確實現(xiàn) MQTT v5 訂閱標(biāo)識符語義訂閱端攜帶的 Subscription-Identifier 必須原樣出現(xiàn)在投遞的 PUBLISH 包屬性中。EMQX 自身通過 emqx_session.erl 的注入邏輯滿足該要求這也是本次修復(fù)的核心訂閱標(biāo)識符數(shù)量有上限每條 ingress 訂閱占用一個標(biāo)識符最大編號為 268435455實際可分配數(shù)量受?MAX_SUBSCRIPTION_ID約束。測試驗證本次修復(fù)的各個環(huán)節(jié)均有對應(yīng)測試用例覆蓋可作為回歸驗證的入口。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)。該用例直接驗證了修復(fù)目標(biāo)MQ 隊列消息投遞時 PUBLISH 包必須攜帶訂閱標(biāo)識符。橋接側(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/#場景下按訂閱標(biāo)識符分發(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 行起驗證遠端不支持訂閱標(biāo)識符時普通訂閱先發(fā)帶 subid 的 SUBSCRIBE、收到不支持響應(yīng)后自動降級重發(fā)不帶 subid 的 SUBSCRIBE第 597-598 行分別斷言subscribe_with_subid與subscribe_without_subid消息。結(jié)語本次修復(fù)changes/ee/fix-16999.en.md從根因上補齊了 MQ 投遞鏈路中 PUBLISH 包缺失 Subscription-Identifier 的問題使 MQTT 橋接 Source 在遠端 broker 開啟 MQ 特性時能夠正常從$queue/訂閱接收消息。修復(fù)同時完善了三層防護MQ 投遞側(cè)正確注入訂閱標(biāo)識符、橋接 ingress 側(cè)按標(biāo)識符優(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投遞修復(fù)解析Subscription Identifier 等訂閱選項的完整傳遞鏈路EMQX 消息流Streams投遞修復(fù)解析Subscription Identifier 等訂閱選項的完整傳遞鏈路 本指南圍繞變更記錄 changes/e后端物聯(lián)網(wǎng)消息隊列通信EMQX 修復(fù)解析Message Queue 啟用時 $queue/ 訂閱無法向 MQTT 橋接投遞消息的問題EMQX 修復(fù)解析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/` 訂閱消息丟失修復(fù)：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/主題訂閱時無法收到消息的問題根因與修復(fù)方案。讀者讀完本文后將掌握 MQTT v5 Subscription-Identifier 屬性在 EMQX 訂閱、投遞、橋接路由三個環(huán)節(jié)的完整流轉(zhuǎn)機制理解為什么$queue/隊列訂閱強制要求 MQTT v5 連接并能通過源碼與測試用例驗證該修復(fù)的可靠性。背景為什么 MQTT 橋接需要 Subscription-IdentifierEMQX 的 MQTT 橋接emqx_bridge_mqtt在作為 MQTT Sourceingress運行時會通過一條 MQTT 客戶端連接訂閱遠端 broker 的主題把收到的消息導(dǎo)入本地 broker 或觸發(fā)規(guī)則引擎處理。當(dāng)遠端 broker 同時開啟了 EMQX 的 Message QueueMQ特性時客戶端需要訂閱形如$queue/queue-name/topic-filter的主題才能消費隊列中的消息。MQTT v5 協(xié)議中客戶端可以在 SUBSCRIBE 包的屬性中攜帶一個或多個Subscription-Identifier訂閱標(biāo)識符服務(wù)端在向該訂閱投遞消息時會把對應(yīng)的訂閱標(biāo)識符原樣寫入 PUBLISH 包的屬性中。這讓訂閱方無需在客戶端維護主題過濾器到業(yè)務(wù)處理的映射即可根據(jù) PUBLISH 包屬性快速識別這條消息來自哪一條訂閱。EMQX MQTT 橋接的 ingress 正是依賴這一機制完成消息路由的橋接啟動時為每條 ingress 訂閱分配唯一的 Subscription-Identifier訂閱時在 SUBSCRIBE 屬性中帶上該標(biāo)識符收到消息后優(yōu)先根據(jù) PUBLISH 包中的 Subscription-Identifier 查索引定位對應(yīng)的通道配置channel config只有找不到時才退化為按主題匹配。如果遠端 broker 在投遞 MQ 隊列消息時沒有在 PUBLISH 包中攜帶 Subscription-Identifieringress 就無法通過標(biāo)識符定位到正確的通道消息隨之丟失——這正是本次修復(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)象遠端 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.修復(fù)前遠端 broker 的 MQ 投遞鏈路在生成 PUBLISH 包時未注入Subscription-Identifier屬性導(dǎo)致第 2 步永遠命中按標(biāo)識符查找的空結(jié)果消息被靜默丟棄。修復(fù)方案MQ 投遞鏈路補全 Subscription-Identifier修復(fù)的實質(zhì)是在 MQ 的消息投遞路徑上補齊 Subscription-Identifier 的注入同時完善橋接 ingress 側(cè)的訂閱協(xié)商與強制校驗邏輯。整個修復(fù)可以拆解為三個層面。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明確告知客戶端本服務(wù)端支持該特性。投遞階段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,該邏輯有對應(yīng)的單元測試佐證見 emqx_session.erl例如enrich_subid1_test/0驗證subid 42的訂閱會把Subscription-Identifier : 42寫入消息屬性。2. MQ 消費鏈路隊列消息經(jīng) broker 投遞時保留訂閱標(biāo)識符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 投遞管線前對訂閱標(biāo)識符的保留不完整導(dǎo)致最終生成的 PUBLISH 包缺失該屬性。修復(fù)后隊列消息在被投遞給$queue/訂閱者時會走與普通訂閱一致的enrich_message/3注入邏輯從而在 PUBLISH 包中攜帶訂閱時的 Subscription-Identifier。橋接 ingress 收到后即可按標(biāo)識符完成路由。3. 橋接 ingress 側(cè)訂閱協(xié)商與強制校驗橋接側(cè)在本次修復(fù)中同步完善了與 Subscription-Identifier 相關(guān)的訂閱行為涉及 emqx_bridge_mqtt_ingress.erl 與 emqx_bridge_mqtt_connector.erl 兩個模塊。訂閱標(biāo)識符分配每條 ingress 訂閱會被分配一個唯一標(biāo)識符最大取值為?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。因為隊列消息的路由完全依賴該屬性降級后必然導(dǎo)致消息丟失顯式報錯比靜默丟消息更安全。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直接拋出不可恢復(fù)錯誤見 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)識符索引僅在proto_ver : v5時創(chuàng)建見 emqx_bridge_mqtt_connector.erl與上面的強制校驗形成閉環(huán)。消息到達后的路由ingress 收到 PUBLISH 包后解析屬性中的Subscription-Identifier支持單個整數(shù)或多個標(biāo)識符的列表見 emqx_bridge_mqtt_ingress.erl到 ETS 索引中查找對應(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 訂閱遠端 broker 的$queue/主題時需要滿足以下條件遠端與本地連接均須使用 MQTT v5connector 配置中proto_ver v5Subscription-Identifier 是 MQTT v5 特性emqx_bridge_mqtt_connector.erl僅在 v5 下創(chuàng)建標(biāo)識符索引且$queue/訂閱在不支持該特性時會啟動失敗并報subscription_identifier_required_for_queue_subscription遠端 broker 須正確實現(xiàn) MQTT v5 訂閱標(biāo)識符語義訂閱端攜帶的 Subscription-Identifier 必須原樣出現(xiàn)在投遞的 PUBLISH 包屬性中。EMQX 自身通過 emqx_session.erl 的注入邏輯滿足該要求這也是本次修復(fù)的核心訂閱標(biāo)識符數(shù)量有上限每條 ingress 訂閱占用一個標(biāo)識符最大編號為 268435455實際可分配數(shù)量受?MAX_SUBSCRIPTION_ID約束。測試驗證本次修復(fù)的各個環(huán)節(jié)均有對應(yīng)測試用例覆蓋可作為回歸驗證的入口。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)。該用例直接驗證了修復(fù)目標(biāo)MQ 隊列消息投遞時 PUBLISH 包必須攜帶訂閱標(biāo)識符。橋接側(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/#場景下按訂閱標(biāo)識符分發(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 行起驗證遠端不支持訂閱標(biāo)識符時普通訂閱先發(fā)帶 subid 的 SUBSCRIBE、收到不支持響應(yīng)后自動降級重發(fā)不帶 subid 的 SUBSCRIBE第 597-598 行分別斷言subscribe_with_subid與subscribe_without_subid消息。結(jié)語本次修復(fù)changes/ee/fix-16999.en.md從根因上補齊了 MQ 投遞鏈路中 PUBLISH 包缺失 Subscription-Identifier 的問題使 MQTT 橋接 Source 在遠端 broker 開啟 MQ 特性時能夠正常從$queue/訂閱接收消息。修復(fù)同時完善了三層防護MQ 投遞側(cè)正確注入訂閱標(biāo)識符、橋接 ingress 側(cè)按標(biāo)識符優(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投遞修復(fù)解析Subscription Identifier 等訂閱選項的完整傳遞鏈路EMQX 消息流Streams投遞修復(fù)解析Subscription Identifier 等訂閱選項的完整傳遞鏈路 本指南圍繞變更記錄 changes/e后端物聯(lián)網(wǎng)消息隊列通信EMQX 修復(fù)解析Message Queue 啟用時 $queue/ 訂閱無法向 MQTT 橋接投遞消息的問題EMQX 修復(fù)解析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），僅供參考