发布链讲完之后下一块最容易踩坑的就是订阅链。很多工程师第一次写订阅脑子里想得很简单我把主题字符串发给 Broker不就完了吗真到现场问题马上就来了为什么有些主题一订就报错为什么sport/#/a不行为什么$SYS/#和普通通配订阅不是一回事为什么 MQTT 5.0 下共享订阅、订阅标识符还要看 Broker 脸色为什么取消订阅也会把状态机卡死这一篇就把这条链一次讲清楚。先给结论SUBSCRIBE / UNSUBSCRIBE 真正难的地方不是构包而是主题过滤器校验 服务端能力边界 ACK 等待链。一、先看订阅链在状态机里的位置flowchart LR A[iConnected] -- B[iSubscribe] B -- C[iSubAck] C -- A A -- D[iUnsubscribe] D -- E[iUnsubAck] E -- A这张图先说明两个事实订阅和退订都不是“发完就算了”而是都要等 ACK。订阅链和发布链一样都是标准状态链而不是随手塞到iConnected里发一下。二、SUBSCRIBE 报文最核心的组成是什么1. 固定报头字段说明报文类型SUBSCRIBERemaining Length后续字节总长度2. 可变报头字段说明Packet Identifier订阅请求标识Properties仅 MQTT 5.0 存在3. 载荷字段说明Topic Filter订阅过滤器Subscribe OptionsQoS 等选项所以你会发现SUBSCRIBE 跟 PUBLISH 很像都有 Packet ID都要等后续确认都会进入等待状态但它的重点不在载荷内容多复杂而在于Topic Filter 合法不合法。三、主题过滤器和发布主题不是一回事这一点必须先钉死。场景是否允许/#发布主题 Topic Name不允许订阅过滤器 Topic Filter允许但要按规则用所以你不能把发布用的 Topic Name订阅用的 Topic Filter当成同一种字符串看。这也是为什么源码里对发布和订阅用了两套不同校验路径。四、Topic Filter 的 3 条硬规则规则 1#只能出现在最后一级合法factory/##非法factory/#/alarma/b/#/c规则 2必须独占一个层级合法factory//temp/status非法factory1/temproom/statusa/b/c规则 3以$开头的系统主题不能直接拿#或在第二个字符位上乱配也就是说这种要拦$#$五、M_IsValidTopicFilter怎么把规则落成代码这段方法非常值得拿出来看因为它不是抽象讲规则而是真正把规则敲成了离线校验逻辑。1. 先看#的判断IF byChar 16#23 THEN IF i uiLen THEN M_IsValidTopicFilter : FALSE; RETURN; END_IF IF (i 1) AND (TO_BYTE(sTopicFilter[i - 2]) 16#2F) THEN M_IsValidTopicFilter : FALSE; RETURN; END_IF END_IF翻成人话就是#只能在最后如果前面还有字符那它前一个字符必须是/2. 再看的判断IF byChar 16#2B THEN byPrev : 0; byNext : 0; IF i 1 THEN byPrev : TO_BYTE(sTopicFilter[i - 2]); END_IF IF i uiLen THEN byNext : TO_BYTE(sTopicFilter[i]); END_IF IF ((i 1) AND (byPrev 16#2F)) OR ((i uiLen) AND (byNext 16#2F)) THEN M_IsValidTopicFilter : FALSE; RETURN; END_IF END_IF翻成人话就是左边如果有字符必须是/右边如果还有字符也必须是/。3. 再看$的特殊限制IF (uiLen 2) AND (sTopicFilter[0] 16#24) THEN IF (sTopicFilter[1] 16#23) OR (sTopicFilter[1] 16#2B) THEN M_IsValidTopicFilter : FALSE; RETURN; END_IF END_IF这段就是你之前已经碰到过的那类边界条件修正点。别看代码短它非常关键。六、M_BuildSubscribePacket的核心工作流把这个方法压缩一下大概就是下面这张图flowchart TD A[检查 Topic 是否为空] -- B[长度与 UTF-8 校验] B -- C[M_IsValidTopicFilter] C -- D[识别是否包含通配符] D -- E[识别是否共享订阅] E -- F[检查 MQTT 5.0 服务端能力] F -- G[分配 Packet ID] G -- H[写 Packet ID Properties Topic Filter QoS]你会发现它在真正构包前做了很多“前置裁决”。这其实就是成熟实现和 demo 实现的差别。七、为什么 MQTT 5.0 下订阅不能只看本地字符串如果是 MQTT 5.0这个库在 SUBSCRIBE 前还会额外检查 Broker 能力。1. 订阅标识符能力IF (eVersion E_MqttVersion.byMqttVersion50) AND (udiSubscriptionId 0) AND (NOT bServerSubIdAvail) THEN M_SetError( uiErrorCode : TO_UINT(E_ReasonCode.uiErrInvalidParameter), sMessage : Server does not support subscription identifiers); M_BuildSubscribePacket : FALSE; RETURN; END_IF2. 通配订阅能力IF (eVersion E_MqttVersion.byMqttVersion50) AND (NOT bServerWildcardSubAvail) AND bHasWildcard THEN M_SetError( uiErrorCode : TO_UINT(E_ReasonCode.uiErrInvalidParameter), sMessage : Server does not support wildcard subscriptions); M_BuildSubscribePacket : FALSE; RETURN; END_IF3. 共享订阅能力IF (eVersion E_MqttVersion.byMqttVersion50) AND (NOT bServerSharedSubAvail) AND bIsSharedSubscription THEN M_SetError( uiErrorCode : TO_UINT(E_ReasonCode.uiErrInvalidParameter), sMessage : Server does not support shared subscriptions); M_BuildSubscribePacket : FALSE; RETURN; END_IF这三段把 MQTT 5.0 的工程味一下就拉满了不是你会拼格式就够了八、共享订阅为什么不是随便写个$share/...就完了源码里识别共享订阅的方式很直接IF LEN(sSubTopic) 7 THEN IF (sSubTopic[0] 16#24) AND (sSubTopic[1] 16#73) AND (sSubTopic[2] 16#68) AND (sSubTopic[3] 16#61) AND (sSubTopic[4] 16#72) AND (sSubTopic[5] 16#65) AND (sSubTopic[6] 16#2F) THEN bIsSharedSubscription : TRUE; END_IF END_IF也就是明确检查前缀是不是$share/但识别出来不等于就能用。还得继续看 Broker 在 CONNACK 里有没有声明支持Shared Subscription。九、UNSUBSCRIBE 为什么也不能写得太随便很多人会觉得取消订阅不就是把主题再发一次吗实际上UNSUBSCRIBE 也有自己的 ACK 链sequenceDiagram participant PLC as PLC Client participant Broker as Broker PLC-Broker: UNSUBSCRIBE Broker--PLC: UNSUBACK对应状态机是iConnected - iUnsubscribe - iUnsubAck - iConnected所以退订卡死往往不是“主题字符串错了”这么简单。也可能是Packet ID 没管好ACK 没等对状态机等待状态没释放十、M_BuildUnsubscribePacket的重点其实更纯粹这个方法的逻辑相对订阅更简单主线就是校验退订主题分配 Packet ID如果是 MQTT 5.0写一个空属性长度写主题过滤器注意这句uiPendingUnsubPacketId : M_GetNextPacketId(); IF uiPendingUnsubPacketId 0 THEN M_SetError( uiErrorCode : TO_UINT(E_ReasonCode.uiErrReceiveMaxExceeded), sMessage : No Packet ID available for unsubscribe); M_BuildUnsubscribePacket : FALSE; RETURN; END_IF这说明退订不是随便发个字符串它同样是一个有身份、有等待链的协议动作。十一、为什么订阅链常见的问题不是“完全错”而是“边界错”订阅链很有意思。它比发布链更容易出现“多数时候能用但总有几个边角不对”的情况。最典型的边界就是问题常见原因某些过滤器一订就错#/规则没写严系统主题订阅不对$前缀规则没处理共享订阅格式能过但 Broker 拒绝只做了字符串判断没做能力判断MQTT 5.0 订阅标识符报错Broker 不支持但客户端硬发退订后状态机卡住UNSUBACK等待链处理不完整所以订阅链的成熟度不看“能不能订到一个主题”而看边界条件能不能全站住。十二、把订阅链的四层关系串起来1. 标准层SUBSCRIBE - SUBACKUNSUBSCRIBE - UNSUBACK2. 报文层都带 Packet IDSUBSCRIBE 载荷里是 Topic Filter OptionsUNSUBSCRIBE 载荷里是 Topic Filter3. 状态机层iSubscribe - iSubAckiUnsubscribe - iUnsubAck4. ST 实现层动作方法订阅过滤器校验M_IsValidTopicFilter构造订阅报文M_BuildSubscribePacket构造退订报文M_BuildUnsubscribePacket十三、这一篇你最该记住的 5 句话订阅过滤器和发布主题不是一种字符串。#只能在最后必须独占一个层级。共享订阅不是只看$share/前缀还要看 Broker 能力。SUBSCRIBE / UNSUBSCRIBE 都是完整状态链不是发完就结束。订阅链真正考验的是边界条件而不是最简单那一个主题能不能跑通。十四、下篇预告下一篇我们不再盯单个报文直接上总控PLC 里写 MQTT最难的其实不是报文而是状态机重点会讲FB_MqttClient主状态机怎么分层为什么iConnected才是真正的调度中心为什么接收链和发送链不能乱绞在一起到这里我们才算真正进入“这套客户端为什么能跑稳”的核心。完整 ST 代码复制使用说明这部分给出的是与本篇主题直接对应的完整 ST 代码不是零碎片段。如果你只是想先跑通优先整段复制不要只摘几行变量或几条赋值语句。如果是METHOD请确认它仍然属于FB_MqttClient如果是PROGRAM请确认相关 DUT、GVL、FB 已一并导入。代码阅读重点先按报文结构 - 状态机入口 - 关键变量 - 返回结果的顺序看。再把正文里的十六进制拆解和这里的字节写入、字节解析语句一行行对上。最后回到在线调试重点盯uiTxLength、uiRxLength、eState、xWaitingForAck这类状态量。完整代码 1M_BuildSubscribePacket对应源码路径10 MQTT/MqttClient_V1_0/Device/Application/MQTT/POUs/MqttClient NBS/FB_MqttClient/??????/M_BuildSubscribePacket.st复制使用说明这是 SUBSCRIBE 主方法复制后就能直接对应到订阅报文的组包流程。阅读重点重点看 Topic Filter 校验、Packet ID 生成、MQTT 5.0 订阅属性以及最后一字节订阅选项的拼接。/// /// 名称 : M_BuildSubscribePacket /// 功能 : 构建 SUBSCRIBE 发送报文 /// 说明 : 根据 MQTT 版本组装订阅报文并在发送前执行基础协议校验。 /// 编程人员 : ControlRookie /// 时间 : 2026-05-05 /// 版本 : V1.2 /// {attribute hide_all_locals} METHOD M_BuildSubscribePacket : BOOL VAR uiPos : UINT : 0; // 当前写入偏移 uiVarHeaderLen : UINT; // 可变报头长度 uiPayloadLen : UINT; // 载荷长度 uiRemainingLen : UINT; // 剩余长度 uiPropsLen : UINT; // MQTT 5.0 属性总长度 uiSubIdVbiBytes : UINT; // 订阅标识符 VBI 编码字节数 i : DINT; // 通用循环索引 bHasWildcard : BOOL; // 是否包含通配符 bIsSharedSubscription : BOOL; // 是否共享订阅 END_VAR // IMPLEMENTATION IF SIZEOF(aTxBuf) 256 THEN M_BuildSubscribePacket : FALSE; RETURN; END_IF IF sSubTopic THEN M_SetError( uiErrorCode : TO_UINT(E_ReasonCode.uiErrInvalidParameter), sMessage : Subscribe topic is required); M_BuildSubscribePacket : FALSE; RETURN; END_IF IF TO_UINT(LEN(sSubTopic)) GVL_Mqtt.cnMaxTopicLen THEN M_SetError( uiErrorCode : TO_UINT(E_ReasonCode.uiErrInvalidParameter), sMessage : Subscribe topic exceeds maximum length); M_BuildSubscribePacket : FALSE; RETURN; END_IF IF NOT M_IsValidUtf8String(sValue : sSubTopic) THEN M_SetError( uiErrorCode : TO_UINT(E_ReasonCode.uiErrInvalidParameter), sMessage : Subscribe topic filter is not valid UTF-8); M_BuildSubscribePacket : FALSE; RETURN; END_IF IF NOT M_IsValidTopicFilter(sTopicFilter : sSubTopic) THEN M_SetError( uiErrorCode : TO_UINT(E_ReasonCode.uiErrInvalidParameter), sMessage : Subscribe topic filter is invalid); M_BuildSubscribePacket : FALSE; RETURN; END_IF bHasWildcard : FALSE; FOR i : 1 TO TO_DINT(LEN(sSubTopic)) DO IF (sSubTopic[i - 1] 16#2B) OR (sSubTopic[i - 1] 16#23) THEN bHasWildcard : TRUE; EXIT; END_IF END_FOR bIsSharedSubscription : FALSE; IF LEN(sSubTopic) 7 THEN IF (sSubTopic[0] 16#24) AND (sSubTopic[1] 16#73) AND (sSubTopic[2] 16#68) AND (sSubTopic[3] 16#61) AND (sSubTopic[4] 16#72) AND (sSubTopic[5] 16#65) AND (sSubTopic[6] 16#2F) THEN bIsSharedSubscription : TRUE; END_IF END_IF IF (eVersion E_MqttVersion.byMqttVersion50) AND (udiSubscriptionId 0) AND (NOT bServerSubIdAvail) THEN M_SetError( uiErrorCode : TO_UINT(E_ReasonCode.uiErrInvalidParameter), sMessage : Server does not support subscription identifiers); M_BuildSubscribePacket : FALSE; RETURN; END_IF IF (eVersion E_MqttVersion.byMqttVersion50) AND (NOT bServerWildcardSubAvail) AND bHasWildcard THEN M_SetError( uiErrorCode : TO_UINT(E_ReasonCode.uiErrInvalidParameter), sMessage : Server does not support wildcard subscriptions); M_BuildSubscribePacket : FALSE; RETURN; END_IF IF (eVersion E_MqttVersion.byMqttVersion50) AND (NOT bServerSharedSubAvail) AND bIsSharedSubscription THEN M_SetError( uiErrorCode : TO_UINT(E_ReasonCode.uiErrInvalidParameter), sMessage : Server does not support shared subscriptions); M_BuildSubscribePacket : FALSE; RETURN; END_IF FOR i : LOWER_BOUND(aTxBuf, 1) TO UPPER_BOUND(aTxBuf, 1) DO aTxBuf[i] : 0; END_FOR uiVarHeaderLen : 2; uiPropsLen : 0; uiSubIdVbiBytes : 0; IF eVersion E_MqttVersion.byMqttVersion50 THEN IF udiSubscriptionId 0 THEN IF udiSubscriptionId 128 THEN uiSubIdVbiBytes : 1; ELSIF udiSubscriptionId 16384 THEN uiSubIdVbiBytes : 2; ELSIF udiSubscriptionId 2097152 THEN uiSubIdVbiBytes : 3; ELSE uiSubIdVbiBytes : 4; END_IF uiPropsLen : uiPropsLen 1 uiSubIdVbiBytes; END_IF IF uiPropsLen 128 THEN uiVarHeaderLen : uiVarHeaderLen 1 uiPropsLen; ELSE uiVarHeaderLen : uiVarHeaderLen 2 uiPropsLen; END_IF END_IF uiPayloadLen : TO_UINT(LEN(sSubTopic)) 3; uiRemainingLen : uiVarHeaderLen uiPayloadLen; IF uiRemainingLen 5 SIZEOF(aTxBuf) THEN M_BuildSubscribePacket : FALSE; RETURN; END_IF uiPos : 0; aTxBuf[uiPos] : E_MqttPacketType.bySubscribe; uiPos : uiPos 1; uiPos : uiPos M_EncodeRemainingLength(udiLength : uiRemainingLen, pBuffer : ADR(aTxBuf[uiPos])); uiPendingSubPacketId : M_GetNextPacketId(); IF uiPendingSubPacketId 0 THEN M_SetError( uiErrorCode : TO_UINT(E_ReasonCode.uiErrReceiveMaxExceeded), sMessage : No Packet ID available for subscribe); M_BuildSubscribePacket : FALSE; RETURN; END_IF uiExpectedPacketId : uiPendingSubPacketId; aTxBuf[uiPos] : UINT_TO_BYTE(SHR(uiPendingSubPacketId, 8)); uiPos : uiPos 1; aTxBuf[uiPos] : UINT_TO_BYTE(uiPendingSubPacketId AND 16#FF); uiPos : uiPos 1; IF eVersion E_MqttVersion.byMqttVersion50 THEN uiPos : uiPos M_EncodeRemainingLength(udiLength : uiPropsLen, pBuffer : ADR(aTxBuf[uiPos])); IF udiSubscriptionId 0 THEN aTxBuf[uiPos] : GVL_Mqtt.cnPropSubscriptionId; uiPos : uiPos 1; uiPos : uiPos M_EncodeRemainingLength(udiLength : udiSubscriptionId, pBuffer : ADR(aTxBuf[uiPos])); END_IF END_IF uiPos : uiPos M_AppendString(sStr : sSubTopic, pBuffer : ADR(aTxBuf[uiPos])); aTxBuf[uiPos] : TO_BYTE(eSubQoS AND 16#03); uiPos : uiPos 1; uiTxLength : uiPos; M_BuildSubscribePacket : TRUE;完整代码 2M_BuildUnsubscribePacket对应源码路径10 MQTT/MqttClient_V1_0/Device/Application/MQTT/POUs/MqttClient NBS/FB_MqttClient/??????/M_BuildUnsubscribePacket.st复制使用说明这是取消订阅的完整组包方法和订阅一起看最容易把配对逻辑理清。阅读重点重点看固定报头、Packet ID、主题字符串写入以及为什么 UNSUBSCRIBE 也必须走独立的等待 ACK 流程。/// /// 名称 : M_BuildUnsubscribePacket /// 功能 : 构建 UNSUBSCRIBE 发送报文 /// 说明 : 根据 MQTT 版本组装取消订阅报文并更新发送长度。 /// 编程人员 : ControlRookie /// 时间 : 2026-05-05 /// 版本 : V1.0 /// {attribute hide_all_locals} METHOD M_BuildUnsubscribePacket : BOOL VAR uiPos : UINT : 0; uiVarHeaderLen : UINT; uiPayloadLen : UINT; uiRemainingLen : UINT; uiPropsLen : UINT; //MQTT 5.0属性总长度 i : DINT; END_VAR // IMPLEMENTATION // BUG-10: 缓冲区溢出保护 IF SIZEOF(aTxBuf) 256 THEN M_BuildUnsubscribePacket : FALSE; RETURN; END_IF IF sUnsubTopic THEN M_SetError( uiErrorCode : TO_UINT(E_ReasonCode.uiErrInvalidParameter), sMessage : Unsubscribe topic is required); M_BuildUnsubscribePacket : FALSE; RETURN; END_IF IF TO_UINT(LEN(sUnsubTopic)) GVL_Mqtt.cnMaxTopicLen THEN M_SetError( uiErrorCode : TO_UINT(E_ReasonCode.uiErrInvalidParameter), sMessage : Unsubscribe topic exceeds maximum length); M_BuildUnsubscribePacket : FALSE; RETURN; END_IF IF NOT M_IsValidUtf8String(sValue : sUnsubTopic) THEN M_SetError( uiErrorCode : TO_UINT(E_ReasonCode.uiErrInvalidParameter), sMessage : Unsubscribe topic filter is not valid UTF-8); M_BuildUnsubscribePacket : FALSE; RETURN; END_IF IF NOT M_IsValidTopicFilter(sTopicFilter : sUnsubTopic) THEN M_SetError( uiErrorCode : TO_UINT(E_ReasonCode.uiErrInvalidParameter), sMessage : Unsubscribe topic filter is invalid); M_BuildUnsubscribePacket : FALSE; RETURN; END_IF // 清空发送缓冲区避免数据干扰 FOR i : LOWER_BOUND(aTxBuf, 1) TO UPPER_BOUND(aTxBuf, 1) DO aTxBuf[i] : 0; END_FOR /// /// 长度计算 /// // 可变报头长度 Packet ID(2) uiVarHeaderLen : 2; // MQTT 5.0: 取消订阅属性当前无属性长度为0 uiPropsLen : 0; IF eVersion E_MqttVersion.byMqttVersion50 THEN // 属性长度(0)的VBI编码 1字节 uiVarHeaderLen : uiVarHeaderLen 1 uiPropsLen; END_IF // 载荷长度 主题(2len) uiPayloadLen : TO_UINT(LEN(sUnsubTopic) 2); uiRemainingLen : uiVarHeaderLen uiPayloadLen; // BUG-10: 缓冲区长度检查 IF uiRemainingLen 5 SIZEOF(aTxBuf) THEN M_BuildUnsubscribePacket : FALSE; RETURN; END_IF /// /// 创建报文 /// uiPos : 0; // ****************** 固定报文头 报文类型 剩余长度 ****************** aTxBuf[0] : E_MqttPacketType.byUnsubscribe; uiPos : uiPos 1; uiPos : uiPos M_EncodeRemainingLength(uiRemainingLen, ADR(aTxBuf[uiPos])); // ****************** 可变报文头: Packet ID ****************** uiPendingUnsubPacketId : M_GetNextPacketId(); IF uiPendingUnsubPacketId 0 THEN M_SetError( uiErrorCode : TO_UINT(E_ReasonCode.uiErrReceiveMaxExceeded), sMessage : No Packet ID available for unsubscribe); M_BuildUnsubscribePacket : FALSE; RETURN; END_IF uiExpectedPacketId : uiPendingUnsubPacketId; aTxBuf[uiPos] : UINT_TO_BYTE(SHR(uiPendingUnsubPacketId, 8)); uiPos : uiPos 1; aTxBuf[uiPos] : UINT_TO_BYTE(uiPendingUnsubPacketId AND 16#FF); uiPos : uiPos 1; // ****************** MQTT 5.0: 取消订阅属性 ****************** IF eVersion E_MqttVersion.byMqttVersion50 THEN // 属性长度VBI编码当前为0无属性 uiPos : uiPos M_EncodeRemainingLength(uiPropsLen, ADR(aTxBuf[uiPos])); END_IF // ****************** 载荷: 主题 ****************** uiPos : uiPos M_AppendString(sUnsubTopic, ADR(aTxBuf[uiPos])); uiTxLength : uiPos; M_BuildUnsubscribePacket : TRUE;完整代码 3M_IsValidTopicFilter对应源码路径10 MQTT/MqttClient_V1_0/Device/Application/MQTT/POUs/MqttClient NBS/FB_MqttClient/????/M_IsValidTopicFilter.st复制使用说明这是主题过滤器合法性校验方法建议和 SUBSCRIBE 一起复制不要只复制订阅组包而漏掉校验层。阅读重点重点看#、、$系统主题这些规则怎么落成 ST 判断这正是“标准语言”翻译成“工程代码”的最好例子。/// /// 名称 : M_IsValidTopicFilter /// 功能 : 校验 MQTT Topic Filter 合法性 /// 说明 : 用于 SUBSCRIBE 主题过滤器的离线格式校验 /// 编程人员 : ControlRookie /// 时间 : 2026-05-05 /// 版本 : V1.0 /// {attribute hide_all_locals} METHOD M_IsValidTopicFilter : BOOL VAR_INPUT sTopicFilter : STRING; // 待校验主题过滤器 END_VAR VAR uiLen : UINT; // 过滤器长度 i : UINT; // 遍历索引 byChar : BYTE; // 当前字符 byPrev : BYTE; // 前一字符 byNext : BYTE; // 后一字符 END_VAR // IMPLEMENTATION uiLen : TO_UINT(LEN(sTopicFilter)); IF uiLen 0 THEN M_IsValidTopicFilter : FALSE; RETURN; END_IF FOR i : 1 TO uiLen DO byChar : TO_BYTE(sTopicFilter[i - 1]); IF byChar 0 THEN M_IsValidTopicFilter : FALSE; RETURN; END_IF IF byChar 16#23 THEN IF i uiLen THEN M_IsValidTopicFilter : FALSE; RETURN; END_IF IF (i 1) AND (TO_BYTE(sTopicFilter[i - 2]) 16#2F) THEN M_IsValidTopicFilter : FALSE; RETURN; END_IF END_IF IF byChar 16#2B THEN byPrev : 0; byNext : 0; IF i 1 THEN byPrev : TO_BYTE(sTopicFilter[i - 2]); END_IF IF i uiLen THEN byNext : TO_BYTE(sTopicFilter[i]); END_IF IF ((i 1) AND (byPrev 16#2F)) OR ((i uiLen) AND (byNext 16#2F)) THEN M_IsValidTopicFilter : FALSE; RETURN; END_IF END_IF END_FOR IF (uiLen 2) AND (sTopicFilter[0] 16#24) THEN IF (sTopicFilter[1] 16#23) OR (sTopicFilter[1] 16#2B) THEN M_IsValidTopicFilter : FALSE; RETURN; END_IF END_IF M_IsValidTopicFilter : TRUE;系列导航系列定位第 5 篇上一篇第4篇 QoS1 / QoS2 ACK 与重发下一篇第6篇 PLC 里写 MQTT最难的不是报文而是状态机项目与资料开源项目名称MqttClient_V1_0GitHub 仓库关键字ControlRookie MqttClient_V1_0配套资料说明我已经同步整理 MQTT 标准、控制报文、十六进制拆解、状态机图和 ST 代码映射知识库后续会持续补全和公开入口适合谁收藏正在做 CODESYS / PLC / MQTT 项目的人想把 MQTT 从报文真正看到 ST 代码的人正在排查 QoS1 / QoS2 超时、掉线、重连问题的人 SUBSCRIBE / UNSUBSCRIBE 怎么实现？主题过滤器到底该怎么校验
发布链讲完之后下一块最容易踩坑的就是订阅链。很多工程师第一次写订阅脑子里想得很简单我把主题字符串发给 Broker不就完了吗真到现场问题马上就来了为什么有些主题一订就报错为什么sport/#/a不行为什么$SYS/#和普通通配订阅不是一回事为什么 MQTT 5.0 下共享订阅、订阅标识符还要看 Broker 脸色为什么取消订阅也会把状态机卡死这一篇就把这条链一次讲清楚。先给结论SUBSCRIBE / UNSUBSCRIBE 真正难的地方不是构包而是主题过滤器校验 服务端能力边界 ACK 等待链。一、先看订阅链在状态机里的位置flowchart LR A[iConnected] -- B[iSubscribe] B -- C[iSubAck] C -- A A -- D[iUnsubscribe] D -- E[iUnsubAck] E -- A这张图先说明两个事实订阅和退订都不是“发完就算了”而是都要等 ACK。订阅链和发布链一样都是标准状态链而不是随手塞到iConnected里发一下。二、SUBSCRIBE 报文最核心的组成是什么1. 固定报头字段说明报文类型SUBSCRIBERemaining Length后续字节总长度2. 可变报头字段说明Packet Identifier订阅请求标识Properties仅 MQTT 5.0 存在3. 载荷字段说明Topic Filter订阅过滤器Subscribe OptionsQoS 等选项所以你会发现SUBSCRIBE 跟 PUBLISH 很像都有 Packet ID都要等后续确认都会进入等待状态但它的重点不在载荷内容多复杂而在于Topic Filter 合法不合法。三、主题过滤器和发布主题不是一回事这一点必须先钉死。场景是否允许/#发布主题 Topic Name不允许订阅过滤器 Topic Filter允许但要按规则用所以你不能把发布用的 Topic Name订阅用的 Topic Filter当成同一种字符串看。这也是为什么源码里对发布和订阅用了两套不同校验路径。四、Topic Filter 的 3 条硬规则规则 1#只能出现在最后一级合法factory/##非法factory/#/alarma/b/#/c规则 2必须独占一个层级合法factory//temp/status非法factory1/temproom/statusa/b/c规则 3以$开头的系统主题不能直接拿#或在第二个字符位上乱配也就是说这种要拦$#$五、M_IsValidTopicFilter怎么把规则落成代码这段方法非常值得拿出来看因为它不是抽象讲规则而是真正把规则敲成了离线校验逻辑。1. 先看#的判断IF byChar 16#23 THEN IF i uiLen THEN M_IsValidTopicFilter : FALSE; RETURN; END_IF IF (i 1) AND (TO_BYTE(sTopicFilter[i - 2]) 16#2F) THEN M_IsValidTopicFilter : FALSE; RETURN; END_IF END_IF翻成人话就是#只能在最后如果前面还有字符那它前一个字符必须是/2. 再看的判断IF byChar 16#2B THEN byPrev : 0; byNext : 0; IF i 1 THEN byPrev : TO_BYTE(sTopicFilter[i - 2]); END_IF IF i uiLen THEN byNext : TO_BYTE(sTopicFilter[i]); END_IF IF ((i 1) AND (byPrev 16#2F)) OR ((i uiLen) AND (byNext 16#2F)) THEN M_IsValidTopicFilter : FALSE; RETURN; END_IF END_IF翻成人话就是左边如果有字符必须是/右边如果还有字符也必须是/。3. 再看$的特殊限制IF (uiLen 2) AND (sTopicFilter[0] 16#24) THEN IF (sTopicFilter[1] 16#23) OR (sTopicFilter[1] 16#2B) THEN M_IsValidTopicFilter : FALSE; RETURN; END_IF END_IF这段就是你之前已经碰到过的那类边界条件修正点。别看代码短它非常关键。六、M_BuildSubscribePacket的核心工作流把这个方法压缩一下大概就是下面这张图flowchart TD A[检查 Topic 是否为空] -- B[长度与 UTF-8 校验] B -- C[M_IsValidTopicFilter] C -- D[识别是否包含通配符] D -- E[识别是否共享订阅] E -- F[检查 MQTT 5.0 服务端能力] F -- G[分配 Packet ID] G -- H[写 Packet ID Properties Topic Filter QoS]你会发现它在真正构包前做了很多“前置裁决”。这其实就是成熟实现和 demo 实现的差别。七、为什么 MQTT 5.0 下订阅不能只看本地字符串如果是 MQTT 5.0这个库在 SUBSCRIBE 前还会额外检查 Broker 能力。1. 订阅标识符能力IF (eVersion E_MqttVersion.byMqttVersion50) AND (udiSubscriptionId 0) AND (NOT bServerSubIdAvail) THEN M_SetError( uiErrorCode : TO_UINT(E_ReasonCode.uiErrInvalidParameter), sMessage : Server does not support subscription identifiers); M_BuildSubscribePacket : FALSE; RETURN; END_IF2. 通配订阅能力IF (eVersion E_MqttVersion.byMqttVersion50) AND (NOT bServerWildcardSubAvail) AND bHasWildcard THEN M_SetError( uiErrorCode : TO_UINT(E_ReasonCode.uiErrInvalidParameter), sMessage : Server does not support wildcard subscriptions); M_BuildSubscribePacket : FALSE; RETURN; END_IF3. 共享订阅能力IF (eVersion E_MqttVersion.byMqttVersion50) AND (NOT bServerSharedSubAvail) AND bIsSharedSubscription THEN M_SetError( uiErrorCode : TO_UINT(E_ReasonCode.uiErrInvalidParameter), sMessage : Server does not support shared subscriptions); M_BuildSubscribePacket : FALSE; RETURN; END_IF这三段把 MQTT 5.0 的工程味一下就拉满了不是你会拼格式就够了八、共享订阅为什么不是随便写个$share/...就完了源码里识别共享订阅的方式很直接IF LEN(sSubTopic) 7 THEN IF (sSubTopic[0] 16#24) AND (sSubTopic[1] 16#73) AND (sSubTopic[2] 16#68) AND (sSubTopic[3] 16#61) AND (sSubTopic[4] 16#72) AND (sSubTopic[5] 16#65) AND (sSubTopic[6] 16#2F) THEN bIsSharedSubscription : TRUE; END_IF END_IF也就是明确检查前缀是不是$share/但识别出来不等于就能用。还得继续看 Broker 在 CONNACK 里有没有声明支持Shared Subscription。九、UNSUBSCRIBE 为什么也不能写得太随便很多人会觉得取消订阅不就是把主题再发一次吗实际上UNSUBSCRIBE 也有自己的 ACK 链sequenceDiagram participant PLC as PLC Client participant Broker as Broker PLC-Broker: UNSUBSCRIBE Broker--PLC: UNSUBACK对应状态机是iConnected - iUnsubscribe - iUnsubAck - iConnected所以退订卡死往往不是“主题字符串错了”这么简单。也可能是Packet ID 没管好ACK 没等对状态机等待状态没释放十、M_BuildUnsubscribePacket的重点其实更纯粹这个方法的逻辑相对订阅更简单主线就是校验退订主题分配 Packet ID如果是 MQTT 5.0写一个空属性长度写主题过滤器注意这句uiPendingUnsubPacketId : M_GetNextPacketId(); IF uiPendingUnsubPacketId 0 THEN M_SetError( uiErrorCode : TO_UINT(E_ReasonCode.uiErrReceiveMaxExceeded), sMessage : No Packet ID available for unsubscribe); M_BuildUnsubscribePacket : FALSE; RETURN; END_IF这说明退订不是随便发个字符串它同样是一个有身份、有等待链的协议动作。十一、为什么订阅链常见的问题不是“完全错”而是“边界错”订阅链很有意思。它比发布链更容易出现“多数时候能用但总有几个边角不对”的情况。最典型的边界就是问题常见原因某些过滤器一订就错#/规则没写严系统主题订阅不对$前缀规则没处理共享订阅格式能过但 Broker 拒绝只做了字符串判断没做能力判断MQTT 5.0 订阅标识符报错Broker 不支持但客户端硬发退订后状态机卡住UNSUBACK等待链处理不完整所以订阅链的成熟度不看“能不能订到一个主题”而看边界条件能不能全站住。十二、把订阅链的四层关系串起来1. 标准层SUBSCRIBE - SUBACKUNSUBSCRIBE - UNSUBACK2. 报文层都带 Packet IDSUBSCRIBE 载荷里是 Topic Filter OptionsUNSUBSCRIBE 载荷里是 Topic Filter3. 状态机层iSubscribe - iSubAckiUnsubscribe - iUnsubAck4. ST 实现层动作方法订阅过滤器校验M_IsValidTopicFilter构造订阅报文M_BuildSubscribePacket构造退订报文M_BuildUnsubscribePacket十三、这一篇你最该记住的 5 句话订阅过滤器和发布主题不是一种字符串。#只能在最后必须独占一个层级。共享订阅不是只看$share/前缀还要看 Broker 能力。SUBSCRIBE / UNSUBSCRIBE 都是完整状态链不是发完就结束。订阅链真正考验的是边界条件而不是最简单那一个主题能不能跑通。十四、下篇预告下一篇我们不再盯单个报文直接上总控PLC 里写 MQTT最难的其实不是报文而是状态机重点会讲FB_MqttClient主状态机怎么分层为什么iConnected才是真正的调度中心为什么接收链和发送链不能乱绞在一起到这里我们才算真正进入“这套客户端为什么能跑稳”的核心。完整 ST 代码复制使用说明这部分给出的是与本篇主题直接对应的完整 ST 代码不是零碎片段。如果你只是想先跑通优先整段复制不要只摘几行变量或几条赋值语句。如果是METHOD请确认它仍然属于FB_MqttClient如果是PROGRAM请确认相关 DUT、GVL、FB 已一并导入。代码阅读重点先按报文结构 - 状态机入口 - 关键变量 - 返回结果的顺序看。再把正文里的十六进制拆解和这里的字节写入、字节解析语句一行行对上。最后回到在线调试重点盯uiTxLength、uiRxLength、eState、xWaitingForAck这类状态量。完整代码 1M_BuildSubscribePacket对应源码路径10 MQTT/MqttClient_V1_0/Device/Application/MQTT/POUs/MqttClient NBS/FB_MqttClient/??????/M_BuildSubscribePacket.st复制使用说明这是 SUBSCRIBE 主方法复制后就能直接对应到订阅报文的组包流程。阅读重点重点看 Topic Filter 校验、Packet ID 生成、MQTT 5.0 订阅属性以及最后一字节订阅选项的拼接。/// /// 名称 : M_BuildSubscribePacket /// 功能 : 构建 SUBSCRIBE 发送报文 /// 说明 : 根据 MQTT 版本组装订阅报文并在发送前执行基础协议校验。 /// 编程人员 : ControlRookie /// 时间 : 2026-05-05 /// 版本 : V1.2 /// {attribute hide_all_locals} METHOD M_BuildSubscribePacket : BOOL VAR uiPos : UINT : 0; // 当前写入偏移 uiVarHeaderLen : UINT; // 可变报头长度 uiPayloadLen : UINT; // 载荷长度 uiRemainingLen : UINT; // 剩余长度 uiPropsLen : UINT; // MQTT 5.0 属性总长度 uiSubIdVbiBytes : UINT; // 订阅标识符 VBI 编码字节数 i : DINT; // 通用循环索引 bHasWildcard : BOOL; // 是否包含通配符 bIsSharedSubscription : BOOL; // 是否共享订阅 END_VAR // IMPLEMENTATION IF SIZEOF(aTxBuf) 256 THEN M_BuildSubscribePacket : FALSE; RETURN; END_IF IF sSubTopic THEN M_SetError( uiErrorCode : TO_UINT(E_ReasonCode.uiErrInvalidParameter), sMessage : Subscribe topic is required); M_BuildSubscribePacket : FALSE; RETURN; END_IF IF TO_UINT(LEN(sSubTopic)) GVL_Mqtt.cnMaxTopicLen THEN M_SetError( uiErrorCode : TO_UINT(E_ReasonCode.uiErrInvalidParameter), sMessage : Subscribe topic exceeds maximum length); M_BuildSubscribePacket : FALSE; RETURN; END_IF IF NOT M_IsValidUtf8String(sValue : sSubTopic) THEN M_SetError( uiErrorCode : TO_UINT(E_ReasonCode.uiErrInvalidParameter), sMessage : Subscribe topic filter is not valid UTF-8); M_BuildSubscribePacket : FALSE; RETURN; END_IF IF NOT M_IsValidTopicFilter(sTopicFilter : sSubTopic) THEN M_SetError( uiErrorCode : TO_UINT(E_ReasonCode.uiErrInvalidParameter), sMessage : Subscribe topic filter is invalid); M_BuildSubscribePacket : FALSE; RETURN; END_IF bHasWildcard : FALSE; FOR i : 1 TO TO_DINT(LEN(sSubTopic)) DO IF (sSubTopic[i - 1] 16#2B) OR (sSubTopic[i - 1] 16#23) THEN bHasWildcard : TRUE; EXIT; END_IF END_FOR bIsSharedSubscription : FALSE; IF LEN(sSubTopic) 7 THEN IF (sSubTopic[0] 16#24) AND (sSubTopic[1] 16#73) AND (sSubTopic[2] 16#68) AND (sSubTopic[3] 16#61) AND (sSubTopic[4] 16#72) AND (sSubTopic[5] 16#65) AND (sSubTopic[6] 16#2F) THEN bIsSharedSubscription : TRUE; END_IF END_IF IF (eVersion E_MqttVersion.byMqttVersion50) AND (udiSubscriptionId 0) AND (NOT bServerSubIdAvail) THEN M_SetError( uiErrorCode : TO_UINT(E_ReasonCode.uiErrInvalidParameter), sMessage : Server does not support subscription identifiers); M_BuildSubscribePacket : FALSE; RETURN; END_IF IF (eVersion E_MqttVersion.byMqttVersion50) AND (NOT bServerWildcardSubAvail) AND bHasWildcard THEN M_SetError( uiErrorCode : TO_UINT(E_ReasonCode.uiErrInvalidParameter), sMessage : Server does not support wildcard subscriptions); M_BuildSubscribePacket : FALSE; RETURN; END_IF IF (eVersion E_MqttVersion.byMqttVersion50) AND (NOT bServerSharedSubAvail) AND bIsSharedSubscription THEN M_SetError( uiErrorCode : TO_UINT(E_ReasonCode.uiErrInvalidParameter), sMessage : Server does not support shared subscriptions); M_BuildSubscribePacket : FALSE; RETURN; END_IF FOR i : LOWER_BOUND(aTxBuf, 1) TO UPPER_BOUND(aTxBuf, 1) DO aTxBuf[i] : 0; END_FOR uiVarHeaderLen : 2; uiPropsLen : 0; uiSubIdVbiBytes : 0; IF eVersion E_MqttVersion.byMqttVersion50 THEN IF udiSubscriptionId 0 THEN IF udiSubscriptionId 128 THEN uiSubIdVbiBytes : 1; ELSIF udiSubscriptionId 16384 THEN uiSubIdVbiBytes : 2; ELSIF udiSubscriptionId 2097152 THEN uiSubIdVbiBytes : 3; ELSE uiSubIdVbiBytes : 4; END_IF uiPropsLen : uiPropsLen 1 uiSubIdVbiBytes; END_IF IF uiPropsLen 128 THEN uiVarHeaderLen : uiVarHeaderLen 1 uiPropsLen; ELSE uiVarHeaderLen : uiVarHeaderLen 2 uiPropsLen; END_IF END_IF uiPayloadLen : TO_UINT(LEN(sSubTopic)) 3; uiRemainingLen : uiVarHeaderLen uiPayloadLen; IF uiRemainingLen 5 SIZEOF(aTxBuf) THEN M_BuildSubscribePacket : FALSE; RETURN; END_IF uiPos : 0; aTxBuf[uiPos] : E_MqttPacketType.bySubscribe; uiPos : uiPos 1; uiPos : uiPos M_EncodeRemainingLength(udiLength : uiRemainingLen, pBuffer : ADR(aTxBuf[uiPos])); uiPendingSubPacketId : M_GetNextPacketId(); IF uiPendingSubPacketId 0 THEN M_SetError( uiErrorCode : TO_UINT(E_ReasonCode.uiErrReceiveMaxExceeded), sMessage : No Packet ID available for subscribe); M_BuildSubscribePacket : FALSE; RETURN; END_IF uiExpectedPacketId : uiPendingSubPacketId; aTxBuf[uiPos] : UINT_TO_BYTE(SHR(uiPendingSubPacketId, 8)); uiPos : uiPos 1; aTxBuf[uiPos] : UINT_TO_BYTE(uiPendingSubPacketId AND 16#FF); uiPos : uiPos 1; IF eVersion E_MqttVersion.byMqttVersion50 THEN uiPos : uiPos M_EncodeRemainingLength(udiLength : uiPropsLen, pBuffer : ADR(aTxBuf[uiPos])); IF udiSubscriptionId 0 THEN aTxBuf[uiPos] : GVL_Mqtt.cnPropSubscriptionId; uiPos : uiPos 1; uiPos : uiPos M_EncodeRemainingLength(udiLength : udiSubscriptionId, pBuffer : ADR(aTxBuf[uiPos])); END_IF END_IF uiPos : uiPos M_AppendString(sStr : sSubTopic, pBuffer : ADR(aTxBuf[uiPos])); aTxBuf[uiPos] : TO_BYTE(eSubQoS AND 16#03); uiPos : uiPos 1; uiTxLength : uiPos; M_BuildSubscribePacket : TRUE;完整代码 2M_BuildUnsubscribePacket对应源码路径10 MQTT/MqttClient_V1_0/Device/Application/MQTT/POUs/MqttClient NBS/FB_MqttClient/??????/M_BuildUnsubscribePacket.st复制使用说明这是取消订阅的完整组包方法和订阅一起看最容易把配对逻辑理清。阅读重点重点看固定报头、Packet ID、主题字符串写入以及为什么 UNSUBSCRIBE 也必须走独立的等待 ACK 流程。/// /// 名称 : M_BuildUnsubscribePacket /// 功能 : 构建 UNSUBSCRIBE 发送报文 /// 说明 : 根据 MQTT 版本组装取消订阅报文并更新发送长度。 /// 编程人员 : ControlRookie /// 时间 : 2026-05-05 /// 版本 : V1.0 /// {attribute hide_all_locals} METHOD M_BuildUnsubscribePacket : BOOL VAR uiPos : UINT : 0; uiVarHeaderLen : UINT; uiPayloadLen : UINT; uiRemainingLen : UINT; uiPropsLen : UINT; //MQTT 5.0属性总长度 i : DINT; END_VAR // IMPLEMENTATION // BUG-10: 缓冲区溢出保护 IF SIZEOF(aTxBuf) 256 THEN M_BuildUnsubscribePacket : FALSE; RETURN; END_IF IF sUnsubTopic THEN M_SetError( uiErrorCode : TO_UINT(E_ReasonCode.uiErrInvalidParameter), sMessage : Unsubscribe topic is required); M_BuildUnsubscribePacket : FALSE; RETURN; END_IF IF TO_UINT(LEN(sUnsubTopic)) GVL_Mqtt.cnMaxTopicLen THEN M_SetError( uiErrorCode : TO_UINT(E_ReasonCode.uiErrInvalidParameter), sMessage : Unsubscribe topic exceeds maximum length); M_BuildUnsubscribePacket : FALSE; RETURN; END_IF IF NOT M_IsValidUtf8String(sValue : sUnsubTopic) THEN M_SetError( uiErrorCode : TO_UINT(E_ReasonCode.uiErrInvalidParameter), sMessage : Unsubscribe topic filter is not valid UTF-8); M_BuildUnsubscribePacket : FALSE; RETURN; END_IF IF NOT M_IsValidTopicFilter(sTopicFilter : sUnsubTopic) THEN M_SetError( uiErrorCode : TO_UINT(E_ReasonCode.uiErrInvalidParameter), sMessage : Unsubscribe topic filter is invalid); M_BuildUnsubscribePacket : FALSE; RETURN; END_IF // 清空发送缓冲区避免数据干扰 FOR i : LOWER_BOUND(aTxBuf, 1) TO UPPER_BOUND(aTxBuf, 1) DO aTxBuf[i] : 0; END_FOR /// /// 长度计算 /// // 可变报头长度 Packet ID(2) uiVarHeaderLen : 2; // MQTT 5.0: 取消订阅属性当前无属性长度为0 uiPropsLen : 0; IF eVersion E_MqttVersion.byMqttVersion50 THEN // 属性长度(0)的VBI编码 1字节 uiVarHeaderLen : uiVarHeaderLen 1 uiPropsLen; END_IF // 载荷长度 主题(2len) uiPayloadLen : TO_UINT(LEN(sUnsubTopic) 2); uiRemainingLen : uiVarHeaderLen uiPayloadLen; // BUG-10: 缓冲区长度检查 IF uiRemainingLen 5 SIZEOF(aTxBuf) THEN M_BuildUnsubscribePacket : FALSE; RETURN; END_IF /// /// 创建报文 /// uiPos : 0; // ****************** 固定报文头 报文类型 剩余长度 ****************** aTxBuf[0] : E_MqttPacketType.byUnsubscribe; uiPos : uiPos 1; uiPos : uiPos M_EncodeRemainingLength(uiRemainingLen, ADR(aTxBuf[uiPos])); // ****************** 可变报文头: Packet ID ****************** uiPendingUnsubPacketId : M_GetNextPacketId(); IF uiPendingUnsubPacketId 0 THEN M_SetError( uiErrorCode : TO_UINT(E_ReasonCode.uiErrReceiveMaxExceeded), sMessage : No Packet ID available for unsubscribe); M_BuildUnsubscribePacket : FALSE; RETURN; END_IF uiExpectedPacketId : uiPendingUnsubPacketId; aTxBuf[uiPos] : UINT_TO_BYTE(SHR(uiPendingUnsubPacketId, 8)); uiPos : uiPos 1; aTxBuf[uiPos] : UINT_TO_BYTE(uiPendingUnsubPacketId AND 16#FF); uiPos : uiPos 1; // ****************** MQTT 5.0: 取消订阅属性 ****************** IF eVersion E_MqttVersion.byMqttVersion50 THEN // 属性长度VBI编码当前为0无属性 uiPos : uiPos M_EncodeRemainingLength(uiPropsLen, ADR(aTxBuf[uiPos])); END_IF // ****************** 载荷: 主题 ****************** uiPos : uiPos M_AppendString(sUnsubTopic, ADR(aTxBuf[uiPos])); uiTxLength : uiPos; M_BuildUnsubscribePacket : TRUE;完整代码 3M_IsValidTopicFilter对应源码路径10 MQTT/MqttClient_V1_0/Device/Application/MQTT/POUs/MqttClient NBS/FB_MqttClient/????/M_IsValidTopicFilter.st复制使用说明这是主题过滤器合法性校验方法建议和 SUBSCRIBE 一起复制不要只复制订阅组包而漏掉校验层。阅读重点重点看#、、$系统主题这些规则怎么落成 ST 判断这正是“标准语言”翻译成“工程代码”的最好例子。/// /// 名称 : M_IsValidTopicFilter /// 功能 : 校验 MQTT Topic Filter 合法性 /// 说明 : 用于 SUBSCRIBE 主题过滤器的离线格式校验 /// 编程人员 : ControlRookie /// 时间 : 2026-05-05 /// 版本 : V1.0 /// {attribute hide_all_locals} METHOD M_IsValidTopicFilter : BOOL VAR_INPUT sTopicFilter : STRING; // 待校验主题过滤器 END_VAR VAR uiLen : UINT; // 过滤器长度 i : UINT; // 遍历索引 byChar : BYTE; // 当前字符 byPrev : BYTE; // 前一字符 byNext : BYTE; // 后一字符 END_VAR // IMPLEMENTATION uiLen : TO_UINT(LEN(sTopicFilter)); IF uiLen 0 THEN M_IsValidTopicFilter : FALSE; RETURN; END_IF FOR i : 1 TO uiLen DO byChar : TO_BYTE(sTopicFilter[i - 1]); IF byChar 0 THEN M_IsValidTopicFilter : FALSE; RETURN; END_IF IF byChar 16#23 THEN IF i uiLen THEN M_IsValidTopicFilter : FALSE; RETURN; END_IF IF (i 1) AND (TO_BYTE(sTopicFilter[i - 2]) 16#2F) THEN M_IsValidTopicFilter : FALSE; RETURN; END_IF END_IF IF byChar 16#2B THEN byPrev : 0; byNext : 0; IF i 1 THEN byPrev : TO_BYTE(sTopicFilter[i - 2]); END_IF IF i uiLen THEN byNext : TO_BYTE(sTopicFilter[i]); END_IF IF ((i 1) AND (byPrev 16#2F)) OR ((i uiLen) AND (byNext 16#2F)) THEN M_IsValidTopicFilter : FALSE; RETURN; END_IF END_IF END_FOR IF (uiLen 2) AND (sTopicFilter[0] 16#24) THEN IF (sTopicFilter[1] 16#23) OR (sTopicFilter[1] 16#2B) THEN M_IsValidTopicFilter : FALSE; RETURN; END_IF END_IF M_IsValidTopicFilter : TRUE;系列导航系列定位第 5 篇上一篇第4篇 QoS1 / QoS2 ACK 与重发下一篇第6篇 PLC 里写 MQTT最难的不是报文而是状态机项目与资料开源项目名称MqttClient_V1_0GitHub 仓库关键字ControlRookie MqttClient_V1_0配套资料说明我已经同步整理 MQTT 标准、控制报文、十六进制拆解、状态机图和 ST 代码映射知识库后续会持续补全和公开入口适合谁收藏正在做 CODESYS / PLC / MQTT 项目的人想把 MQTT 从报文真正看到 ST 代码的人正在排查 QoS1 / QoS2 超时、掉线、重连问题的人