第5篇_SUBSCRIBE / UNSUBSCRIBE 怎么实现？主题过滤器到底该怎么校验

这一篇把订阅和退订链完整拆开，重点讲 Topic Filter 和发布 Topic 的区别、# + $ 的校验规则、MQTT 5.0 能力边界，以及为什么退订也不是发个字符串这么简单。

适合谁收藏

正在做 CODESYS / PLC / MQTT 项目的人
想把 MQTT 从报文真正看到 ST 代码的人
正在排查 QoS1 / QoS2 超时、掉线、重连问题的人

发布链讲完之后，下一块最容易踩坑的就是订阅链。

很多工程师第一次写订阅，脑子里想得很简单：

我把主题字符串发给 Broker，不就完了吗？

真到现场，问题马上就来了：

为什么有些主题一订就报错
为什么 sport/#/a 不行
为什么 $SYS/# 和普通通配订阅不是一回事
为什么 MQTT 5.0 下共享订阅、订阅标识符还要看 Broker 脸色
为什么取消订阅也会把状态机卡死

这一篇就把这条链一次讲清楚。

先给结论：

SUBSCRIBE / UNSUBSCRIBE 真正难的地方，不是构包，而是 主题过滤器校验 + 服务端能力边界 + ACK 等待链。

一、先看订阅链在状态机里的位置

Mermaid

flowchart LR
    A[iConnected] --> B[iSubscribe]
    B --> C[iSubAck]
    C --> A

    A --> D[iUnsubscribe]
    D --> E[iUnsubAck]
    E --> A

这张图先说明两个事实：

订阅和退订都不是“发完就算了”，而是都要等 ACK。
订阅链和发布链一样，都是标准状态链，而不是随手塞到 iConnected 里发一下。

1. 固定报头

字段	说明
报文类型	`SUBSCRIBE`
Remaining Length	后续字节总长度

2. 可变报头

字段	说明
Packet Identifier	订阅请求标识
Properties	仅 MQTT 5.0 存在

3. 载荷

字段	说明
Topic Filter	订阅过滤器
Subscribe Options	QoS 等选项

所以你会发现，SUBSCRIBE 跟 PUBLISH 很像：

都有 Packet ID
都要等后续确认
都会进入等待状态

但它的重点不在载荷内容多复杂，而在于 Topic Filter 合法不合法。

三、主题过滤器和发布主题，不是一回事

这一点必须先钉死。

场景	是否允许 `+` / `#`
发布主题 Topic Name	不允许
订阅过滤器 Topic Filter	允许，但要按规则用

所以你不能把：

发布用的 Topic Name
订阅用的 Topic Filter

当成同一种字符串看。

这也是为什么源码里对发布和订阅用了两套不同校验路径。

四、Topic Filter 的 3 条硬规则

规则 1：只能出现在最后一级

合法：

factory/#
#

非法：

factory/#/alarm
a/b/#/c

规则 2：+ 必须独占一个层级

合法：

factory/+/temp
+/status

非法：

factory+1/temp
room+/status
a/+b/c

规则 3：以 $ 开头的系统主题，不能直接拿或 + 在第二个字符位上乱配

也就是说，这种要拦：

$#
$+

五、M_IsValidTopicFilter 怎么把规则落成代码

这段方法非常值得拿出来看，因为它不是抽象讲规则，而是真正把规则敲成了离线校验逻辑。

1. 先看的判断

iecst

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. 再看 + 的判断

iecst

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. 再看 $ 的特殊限制

iecst

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 的核心工作流

把这个方法压缩一下，大概就是下面这张图：

Mermaid

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. 订阅标识符能力

iecst

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

2. 通配订阅能力

iecst

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

3. 共享订阅能力

iecst

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 的工程味一下就拉满了：

不是你会拼格式就够了，

源码里识别共享订阅的方式很直接：

iecst

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

也就是明确检查前缀是不是：

text

$share/

但识别出来不等于就能用。还得继续看 Broker 在 CONNACK 里有没有声明支持 Shared Subscription。

九、UNSUBSCRIBE 为什么也不能写得太随便

很多人会觉得：

取消订阅不就是把主题再发一次吗？

实际上，UNSUBSCRIBE 也有自己的 ACK 链：

Mermaid

sequenceDiagram
    participant PLC as PLC Client
    participant Broker as Broker

    PLC->>Broker: UNSUBSCRIBE
    Broker-->>PLC: UNSUBACK

对应状态机是：

text

iConnected -> iUnsubscribe -> iUnsubAck -> iConnected

所以退订卡死，往往不是“主题字符串错了”这么简单。也可能是：

Packet ID 没管好
ACK 没等对
状态机等待状态没释放

十、M_BuildUnsubscribePacket 的重点其实更纯粹

这个方法的逻辑相对订阅更简单，主线就是：

校验退订主题
分配 Packet ID
如果是 MQTT 5.0，写一个空属性长度
写主题过滤器

注意这句：

iecst

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 -> SUBACK
UNSUBSCRIBE -> UNSUBACK

2. 报文层

都带 Packet ID
SUBSCRIBE 载荷里是 Topic Filter + Options
UNSUBSCRIBE 载荷里是 Topic Filter

3. 状态机层

iSubscribe -> iSubAck
iUnsubscribe -> iUnsubAck

4. 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 这类状态量。

完整代码 1：M_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 订阅属性以及最后一字节订阅选项的拼接。

iecst

/// =======================================================================
/// 名称      : 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;

完整代码 2：M_BuildUnsubscribePacket

对应源码路径：10 MQTT/MqttClient_V1_0/Device/Application/MQTT/POUs/MqttClient NBS/FB_MqttClient/处理发送报文/M_BuildUnsubscribePacket.st
复制使用说明：这是取消订阅的完整组包方法，和订阅一起看最容易把配对逻辑理清。
阅读重点：重点看固定报头、Packet ID、主题字符串写入，以及为什么 UNSUBSCRIBE 也必须走独立的等待 ACK 流程。

iecst

/// =======================================================================
/// 名称      : 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

// 载荷长度 = 主题(2+len)
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;

完整代码 3：M_IsValidTopicFilter

对应源码路径：10 MQTT/MqttClient_V1_0/Device/Application/MQTT/POUs/MqttClient NBS/FB_MqttClient/辅助功能/M_IsValidTopicFilter.st
复制使用说明：这是主题过滤器合法性校验方法，建议和 SUBSCRIBE 一起复制，不要只复制订阅组包而漏掉校验层。
阅读重点：重点看 #、+、$ 系统主题这些规则怎么落成 ST 判断，这正是“标准语言”翻译成“工程代码”的最好例子。

iecst

/// =======================================================================
/// 名称      : 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，最难的不是报文，而是状态机

适合谁收藏

一、先看订阅链在状态机里的位置

二、SUBSCRIBE 报文最核心的组成是什么

1. 固定报头

2. 可变报头

3. 载荷

三、主题过滤器和发布主题，不是一回事

四、Topic Filter 的 3 条硬规则

规则 1： 只能出现在最后一级

规则 2：+ 必须独占一个层级

规则 3：以 $ 开头的系统主题，不能直接拿 或 + 在第二个字符位上乱配

五、M_IsValidTopicFilter 怎么把规则落成代码

1. 先看 的判断

2. 再看 + 的判断

3. 再看 $ 的特殊限制

六、M_BuildSubscribePacket 的核心工作流

七、为什么 MQTT 5.0 下订阅不能只看本地字符串

1. 订阅标识符能力

2. 通配订阅能力

3. 共享订阅能力

八、共享订阅为什么不是随便写个 $share/... 就完了

九、UNSUBSCRIBE 为什么也不能写得太随便

十、M_BuildUnsubscribePacket 的重点其实更纯粹

十一、为什么订阅链常见的问题不是“完全错”，而是“边界错”

十二、把订阅链的四层关系串起来

1. 标准层

2. 报文层

3. 状态机层

4. ST 实现层

十三、这一篇你最该记住的 5 句话

十四、下篇预告

完整 ST 代码

复制使用说明

代码阅读重点

完整代码 1：M_BuildSubscribePacket

完整代码 2：M_BuildUnsubscribePacket

完整代码 3：M_IsValidTopicFilter

系列导航

评论区预留

规则 1：只能出现在最后一级

规则 3：以 $ 开头的系统主题，不能直接拿或 + 在第二个字符位上乱配

1. 先看的判断