夜雨聆风源码加更05_订阅表、Retain、PUBLISH 路由和业务事件
这一组源码加更只有一个目标：把 MqttBroker 的真实 ST 源码按工程阅读顺序讲完整。不是再补几段“看起来像源码”的片段，而是让读者能沿着源码对象理解这个 Broker 怎么组织、怎么运行、怎么排障。
适合谁收藏
已经读过 MqttBroker 主线教程，想继续看真实源码实现的工程师。 想学习 CodeSys ST 工程如何拆分 Broker、连接池、编解码、路由和 QoS 调度的人。 想把 MQTT Broker 移植到 PLC、边缘控制器或教学工程里的开发者。
先给结论
这一篇看订阅表和路由器如何协同：订阅进入表，Retain 更新缓存，PUBLISH 按 Topic Filter 匹配 fanout，最后把消息送进连接发送队列。
SUBSCRIBE 不是保存一个字符串，PUBLISH 也不是收到就群发。Broker 的核心价值在于把订阅关系、通配符匹配、Retain 回放和业务事件组织成可控路由。
这篇覆盖 13 个源码文件，合计约 819 行 ST 代码。为了保持公开教程可读性，正文先讲源码阅读路径，再给完整源码。读代码时建议不要从第一个代码块一路机械读到底，而是按本篇的“读代码顺序”来抓主线。
从工程问题到代码职责
本篇源码覆盖表
FB_MqttBroker.M_HandlePublish.st | ||
FB_MqttBroker.M_HandleSubscribe.st | ||
FB_MqttBroker.M_HandleUnsubscribe.st | ||
FB_MqttBroker.M_RoutePublishNow.st | ||
FB_MqttBrokerRouter.M_AddSubscription.st | ||
FB_MqttBrokerRouter.M_ClearSlot.st | ||
FB_MqttBrokerRouter.M_FindNextRetain.st | ||
FB_MqttBrokerRouter.M_FindNextRoute.st | ||
FB_MqttBrokerRouter.M_RemoveSubscription.st | ||
FB_MqttBrokerRouter.M_UpdateRetain.st | ||
FB_MqttBrokerRouter.st | ||
F_MqttMinQoS.st | ||
F_MqttTopicMatch.st |
推荐阅读顺序
先看 Router 主体和订阅增删。 再看 Retain 查询与更新。 最后看 Broker 顶层如何处理 Publish/Subscribe/Unsubscribe 业务事件。
验证和排障边界
测试精确主题、 +、#三类订阅。测试 Retain 首次订阅回放，以及取消订阅后不再接收。
本篇完整开源代码
下面代码来自对应 .st 源文件的连续完整内容。为方便公开阅读，只保留源码对象名，不放本机工程路径。
完整代码 01: FB_MqttBroker.M_HandlePublish.st
/// =======================================================================/// 名称 : M_HandlePublish/// 功能 : 处理一条入站或 Will PUBLISH/// 说明 : 统一完成入站事务、PUBACK、Retain 更新和订阅路由投递。/// 编程人员 : ControlRookie/// 时间 : 2026-05-08/// 版本 : V1.0/// ======================================================================={attribute 'hide_all_locals'}METHOD M_HandlePublish : BOOLVAR_INPUTuiSourceSlot : UINT; // 发布来源槽位编号，Will 或系统内部发布也使用原异常槽位[1..cnMaxClientSlots]stPublish : ST_MqttBrokerPublishFrame; // 需要处理的发布帧END_VAR// === IMPLEMENTATION ===IF NOT stPublish.xValid THENM_HandlePublish := FALSE;RETURN;END_IFIF (uiSourceSlot >= 1) AND (uiSourceSlot <= GVL_MqttBroker.cnMaxClientSlots) THENIF NOT M_CheckPublishAcl(uiSlot := uiSourceSlot, sTopic := stPublish.sTopic) THENstMetrics.udiAclRejected := stMetrics.udiAclRejected + 1;M_LogDiag(uiSlot := uiSourceSlot, eError := E_MqttBrokerError.uiInvalidTopic, sMessage := 'Publish topic rejected by ACL');M_HandlePublish := FALSE;RETURN;END_IFEND_IFstMetrics.udiPublishReceived := stMetrics.udiPublishReceived + 1;IF stPublish.eQoS = E_MqttQoS.byQoS1 THENIF fbRxScheduler.M_RecordPublish(stPublish := stPublish, udiNowMs := udiNowMs) THENIF (uiSourceSlot >= 1) AND (uiSourceSlot <= GVL_MqttBroker.cnMaxClientSlots) THENaConnections[uiSourceSlot].M_EnqueueProtocolAck(ePacketType := E_MqttPacketType.byPubAck,uiPacketId := stPublish.uiPacketId,byReturnCode := 0);fbRxScheduler.M_CompletePublish(uiSlot := uiSourceSlot, uiPacketId := stPublish.uiPacketId);stMetrics.udiPubAckSent := stMetrics.udiPubAckSent + 1;END_IFEND_IFEND_IFIF stPublish.eQoS = E_MqttQoS.byQoS2 THENIF fbRxScheduler.M_RecordPublish(stPublish := stPublish, udiNowMs := udiNowMs) THENIF (uiSourceSlot >= 1) AND (uiSourceSlot <= GVL_MqttBroker.cnMaxClientSlots) THENaConnections[uiSourceSlot].M_EnqueueProtocolAck(ePacketType := E_MqttPacketType.byPubRec,uiPacketId := stPublish.uiPacketId,byReturnCode := 0);stMetrics.udiPubRecSent := stMetrics.udiPubRecSent + 1;END_IFEND_IFM_HandlePublish := TRUE;RETURN;END_IFM_RoutePublishNow(stPublish := stPublish);M_HandlePublish := TRUE;
完整代码 02: FB_MqttBroker.M_HandleSubscribe.st
/// =======================================================================/// 名称 : M_HandleSubscribe/// 功能 : 处理连接槽位输出的多 Topic SUBSCRIBE/// 说明 : 每个 Topic Filter 独立写订阅表、生成 SUBACK 返回码，并触发 Retain 补发。/// 编程人员 : ControlRookie/// 时间 : 2026-05-08/// 版本 : V1.0/// ======================================================================={attribute 'hide_all_locals'}METHOD M_HandleSubscribe : BOOLVAR_INPUTuiSlot : UINT; // 发起 SUBSCRIBE 的客户端槽位编号[1..cnMaxClientSlots]END_VARVARuiItemIndex : UINT; // SUBSCRIBE 多 Topic 条目扫描索引[1..cnMaxTopicItemsPerPacket]uiRetainReplayCount : UINT; // 本次 SUBSCRIBE 已补发的 Retain 消息数量[条]aReturnCodes : ARRAY[1..GVL_MqttBroker.cnMaxTopicItemsPerPacket] OF BYTE; // SUBACK 多 Topic 返回码数组uiReturnCount : UINT; // SUBACK 返回码数量[条]END_VAR// === IMPLEMENTATION ===IF (uiSlot < 1) OR (uiSlot > GVL_MqttBroker.cnMaxClientSlots) THENM_HandleSubscribe := FALSE;RETURN;END_IFuiReturnCount := 0;uiRetainReplayCount := 0;FOR uiItemIndex := 1 TO aConnections[uiSlot].uiSubItemCount DOIF uiItemIndex > GVL_MqttBroker.cnMaxTopicItemsPerPacket THENEXIT;END_IFuiReturnCount := uiReturnCount + 1;IF aConnections[uiSlot].aSubItems[uiItemIndex].xUsedAND M_CheckSubscribeAcl(uiSlot := uiSlot, sTopicFilter := aConnections[uiSlot].aSubItems[uiItemIndex].sTopicFilter)AND fbRouter.M_AddSubscription(uiSlot := uiSlot,sClientId := aConnectionStates[uiSlot].sClientId,sTopicFilter := aConnections[uiSlot].aSubItems[uiItemIndex].sTopicFilter,eMaxQoS := aConnections[uiSlot].aSubItems[uiItemIndex].eQoS) THENaReturnCodes[uiReturnCount] := TO_BYTE(aConnections[uiSlot].aSubItems[uiItemIndex].eQoS);IF fbRouter.M_FindNextRetain(stDelivery := stDelivery,uiTargetSlot := uiSlot,sTopicFilter := aConnections[uiSlot].aSubItems[uiItemIndex].sTopicFilter,eMaxQoS := aConnections[uiSlot].aSubItems[uiItemIndex].eQoS,xRestart := TRUE,xFound => xRetainFound) THENWHILE xRetainFound AND (uiRetainReplayCount < GVL_MqttBroker.cnMaxRetainReplayPerScan) DOIF aConnections[uiSlot].M_EnqueueDelivery(stDelivery := stDelivery) THENstMetrics.udiPublishDelivered := stMetrics.udiPublishDelivered + 1;sLastRetainTopic := stDelivery.sTopic;uiLastRetainPayloadLen := stDelivery.uiPayloadLen;uiRetainReplayCount := uiRetainReplayCount + 1;IF stDelivery.eQoS <> E_MqttQoS.byQoS0 THEN// Retain 补发本质上也是 Broker 到订阅者方向的出站 PUBLISH。// 当补发 QoS>0 时，必须登记出站事务，否则订阅者后续 PUBACK/PUBREC 无法正确收口。fbTxScheduler.M_RegisterPublish(stPublish := stDelivery, udiNowMs := udiNowMs);END_IFEND_IFfbRouter.M_FindNextRetain(stDelivery := stDelivery,uiTargetSlot := uiSlot,sTopicFilter := aConnections[uiSlot].aSubItems[uiItemIndex].sTopicFilter,eMaxQoS := aConnections[uiSlot].aSubItems[uiItemIndex].eQoS,xRestart := FALSE,xFound => xRetainFound);END_WHILEIF xRetainFound THEN// 默认预算等于 Retain 表容量，因此正常配置下不会漏发。// 如果工程师为了扫描周期把 cnMaxRetainReplayPerScan 调小，本诊断能明确提示本次订阅仍有 Retain 未在同一扫描周期补发。M_LogDiag(uiSlot := uiSlot, eError := E_MqttBrokerError.uiQueueFull, sMessage := 'Retain replay budget exhausted');END_IFEND_IFELSEaReturnCodes[uiReturnCount] := 16#80;stMetrics.udiAclRejected := stMetrics.udiAclRejected + 1;END_IFEND_FORaConnections[uiSlot].M_EnqueueProtocolAckList(aReturnCodes := aReturnCodes,ePacketType := E_MqttPacketType.bySubAck,uiPacketId := aConnections[uiSlot].uiSubPacketId,uiReturnCount := uiReturnCount);M_HandleSubscribe := TRUE;
完整代码 03: FB_MqttBroker.M_HandleUnsubscribe.st
/// =======================================================================/// 名称 : M_HandleUnsubscribe/// 功能 : 处理连接槽位输出的多 Topic UNSUBSCRIBE/// 说明 : 每个 Topic Filter 独立从订阅表删除，最后返回一个 UNSUBACK。/// 编程人员 : ControlRookie/// 时间 : 2026-05-08/// 版本 : V1.0/// ======================================================================={attribute 'hide_all_locals'}METHOD M_HandleUnsubscribe : BOOLVAR_INPUTuiSlot : UINT; // 发起 UNSUBSCRIBE 的客户端槽位编号[1..cnMaxClientSlots]END_VARVARuiItemIndex : UINT; // UNSUBSCRIBE 多 Topic 条目扫描索引[1..cnMaxTopicItemsPerPacket]END_VAR// === IMPLEMENTATION ===IF (uiSlot < 1) OR (uiSlot > GVL_MqttBroker.cnMaxClientSlots) THENM_HandleUnsubscribe := FALSE;RETURN;END_IFFOR uiItemIndex := 1 TO aConnections[uiSlot].uiUnsubItemCount DOIF uiItemIndex > GVL_MqttBroker.cnMaxTopicItemsPerPacket THENEXIT;END_IFIF aConnections[uiSlot].aUnsubItems[uiItemIndex].xUsed THENfbRouter.M_RemoveSubscription(uiSlot := uiSlot,sTopicFilter := aConnections[uiSlot].aUnsubItems[uiItemIndex].sTopicFilter);END_IFEND_FORaConnections[uiSlot].M_EnqueueProtocolAck(ePacketType := E_MqttPacketType.byUnsubAck,uiPacketId := aConnections[uiSlot].uiUnsubPacketId,byReturnCode := 0);M_HandleUnsubscribe := TRUE;
完整代码 04: FB_MqttBroker.M_RoutePublishNow.st
/// =======================================================================/// 名称 : M_RoutePublishNow/// 功能 : 对已确认可路由的发布帧执行 Retain 与订阅路由/// 说明 : QoS0/QoS1 直接调用；QoS2 仅在首次 PUBREL 时调用，保证 exactly once。/// 编程人员 : ControlRookie/// 时间 : 2026-05-08/// 版本 : V1.0/// ======================================================================={attribute 'hide_all_locals'}METHOD M_RoutePublishNow : BOOLVAR_INPUTstPublish : ST_MqttBrokerPublishFrame; // 已确认允许执行路由的发布帧END_VARVARuiRouteCount : UINT; // 本次发布已经生成的路由投递数量stRouteFrame : ST_MqttBrokerPublishFrame; // 路由器返回的单条投递任务END_VAR// === IMPLEMENTATION ===IF NOT stPublish.xValid THENM_RoutePublishNow := FALSE;RETURN;END_IFIF stPublish.xRetain THENIF fbRouter.M_UpdateRetain(stPublish := stPublish, udiNowMs := udiNowMs) THENsLastRetainTopic := stPublish.sTopic;uiLastRetainPayloadLen := stPublish.uiPayloadLen;uiRetainCount := fbRouter.uiRetainCount;IF stPublish.uiPayloadLen = 0 THENstMetrics.udiRetainCleared := stMetrics.udiRetainCleared + 1;ELSEstMetrics.udiRetainUpdated := stMetrics.udiRetainUpdated + 1;END_IFEND_IFEND_IFuiRouteCount := 0;IF fbRouter.M_FindNextRoute(stDelivery := stRouteFrame,stPublish := stPublish,xRestart := TRUE,xFound => xRouteFound) THENWHILE xRouteFound AND (uiRouteCount < GVL_MqttBroker.cnMaxRouteFanoutPerScan) DOIF (stRouteFrame.uiTargetSlot >= 1) AND (stRouteFrame.uiTargetSlot <= GVL_MqttBroker.cnMaxClientSlots) THENIF aConnections[stRouteFrame.uiTargetSlot].M_EnqueueDelivery(stDelivery := stRouteFrame) THENstMetrics.udiPublishDelivered := stMetrics.udiPublishDelivered + 1;uiRouteCount := uiRouteCount + 1;IF stRouteFrame.eQoS <> E_MqttQoS.byQoS0 THEN// M_EnqueueDelivery 会为首次 QoS>0 投递分配“目标连接内”的 PacketId。// 事务表必须登记这个已分配后的 PacketId，后续订阅者 PUBACK 才能正确清理事务。fbTxScheduler.M_RegisterPublish(stPublish := stRouteFrame, udiNowMs := udiNowMs);END_IFELSEstMetrics.udiPublishDropped := stMetrics.udiPublishDropped + 1;END_IFEND_IFfbRouter.M_FindNextRoute(stDelivery := stRouteFrame,stPublish := stPublish,xRestart := FALSE,xFound => xRouteFound);END_WHILEIF xRouteFound THEN// 路由扇出预算耗尽时保留诊断，不把它伪装成“没有订阅者”。// 当前轻量实现不缓存剩余路由任务，默认 cnMaxRouteFanoutPerScan 与客户端规模一致，正常 5~8 客户端不会丢投。M_LogDiag(uiSlot := stPublish.uiSourceSlot, eError := E_MqttBrokerError.uiQueueFull, sMessage := 'Route fanout budget exhausted');END_IFEND_IFM_RoutePublishNow := TRUE;
完整代码 05: FB_MqttBrokerRouter.M_AddSubscription.st
/// =======================================================================/// 名称 : M_AddSubscription/// 功能 : 新增或更新订阅表项/// 说明 : 同一槽位同一 Topic Filter 重复订阅时更新 QoS，而不是新增重复项。/// 编程人员 : ControlRookie/// 时间 : 2026-05-08/// 版本 : V1.0/// ======================================================================={attribute 'hide_all_locals'}METHOD M_AddSubscription : BOOLVAR_INPUTuiSlot : UINT; // 订阅所属客户端槽位编号[1..cnMaxClientSlots]sClientId : STRING; // 订阅所属 ClientID 快照sTopicFilter : STRING; // 客户端请求的 Topic FiltereMaxQoS : E_MqttQoS; // 客户端请求的最大 QoSEND_VARVARuiIndex : UINT; // 订阅表扫描索引uiFreeIndex : UINT; // 首个空闲订阅表槽位索引END_VAR// === IMPLEMENTATION ===IF (uiSlot < 1) OR (uiSlot > GVL_MqttBroker.cnMaxClientSlots) THENeLastError := E_MqttBrokerError.uiInvalidState;M_AddSubscription := FALSE;RETURN;END_IFIF NOT F_MqttIsValidTopicFilter(sFilter := sTopicFilter) THENeLastError := E_MqttBrokerError.uiInvalidTopic;M_AddSubscription := FALSE;RETURN;END_IFuiFreeIndex := 0;FOR uiIndex := 1 TO GVL_MqttBroker.cnMaxSubscriptions DOIF aSubscriptions[uiIndex].xUsed THENIF (aSubscriptions[uiIndex].uiSlot = uiSlot) AND (aSubscriptions[uiIndex].sTopicFilter = sTopicFilter) THENaSubscriptions[uiIndex].eMaxQoS := eMaxQoS;aSubscriptions[uiIndex].xActive := TRUE;eLastError := E_MqttBrokerError.uiNoError;M_AddSubscription := TRUE;RETURN;END_IFELSIF uiFreeIndex = 0 THENuiFreeIndex := uiIndex;END_IFEND_FORIF uiFreeIndex = 0 THENeLastError := E_MqttBrokerError.uiSubscriptionFull;M_AddSubscription := FALSE;RETURN;END_IFaSubscriptions[uiFreeIndex].xUsed := TRUE;aSubscriptions[uiFreeIndex].xActive := TRUE;aSubscriptions[uiFreeIndex].uiSlot := uiSlot;aSubscriptions[uiFreeIndex].sClientId := sClientId;aSubscriptions[uiFreeIndex].sTopicFilter := sTopicFilter;aSubscriptions[uiFreeIndex].uiFilterLen := TO_UINT(LEN(sTopicFilter));aSubscriptions[uiFreeIndex].eMaxQoS := eMaxQoS;uiSubscriptionCount := uiSubscriptionCount + 1;eLastError := E_MqttBrokerError.uiNoError;M_AddSubscription := TRUE;
完整代码 06: FB_MqttBrokerRouter.M_ClearSlot.st
/// =======================================================================/// 名称 : M_ClearSlot/// 功能 : 清理指定连接槽位的订阅资源/// 说明 : Clean Session 或连接释放时调用，避免断线客户端继续收到路由任务。/// 编程人员 : ControlRookie/// 时间 : 2026-05-08/// 版本 : V1.0/// ======================================================================={attribute 'hide_all_locals'}METHOD M_ClearSlot : BOOLVAR_INPUTuiSlot : UINT; // 需要清理的客户端槽位编号[1..cnMaxClientSlots]END_VARVARuiIndex : UINT; // 订阅表扫描索引END_VAR// === IMPLEMENTATION ===IF (uiSlot < 1) OR (uiSlot > GVL_MqttBroker.cnMaxClientSlots) THENeLastError := E_MqttBrokerError.uiInvalidState;M_ClearSlot := FALSE;RETURN;END_IFFOR uiIndex := 1 TO GVL_MqttBroker.cnMaxSubscriptions DOIF aSubscriptions[uiIndex].xUsed AND (aSubscriptions[uiIndex].uiSlot = uiSlot) THENaSubscriptions[uiIndex].xUsed := FALSE;aSubscriptions[uiIndex].xActive := FALSE;aSubscriptions[uiIndex].uiSlot := 0;aSubscriptions[uiIndex].sClientId := '';aSubscriptions[uiIndex].sTopicFilter := '';aSubscriptions[uiIndex].uiFilterLen := 0;aSubscriptions[uiIndex].eMaxQoS := E_MqttQoS.byQoS0;IF uiSubscriptionCount > 0 THENuiSubscriptionCount := uiSubscriptionCount - 1;END_IFEND_IFEND_FOReLastError := E_MqttBrokerError.uiNoError;M_ClearSlot := TRUE;
完整代码 07: FB_MqttBrokerRouter.M_FindNextRetain.st
/// =======================================================================/// 名称 : M_FindNextRetain/// 功能 : 查找新订阅命中的下一条 Retain 消息/// 说明 : SUBSCRIBE 成功后循环调用，生成需要补发给订阅者的 Retain 投递任务。/// 编程人员 : ControlRookie/// 时间 : 2026-05-08/// 版本 : V1.0/// ======================================================================={attribute 'hide_all_locals'}METHOD M_FindNextRetain : BOOLVAR_INPUTuiTargetSlot : UINT; // 新订阅客户端槽位编号[1..cnMaxClientSlots]sTopicFilter : STRING; // 新订阅的 Topic FiltereMaxQoS : E_MqttQoS; // 新订阅请求的最大 QoSxRestart : BOOL; // TRUE 表示从 Retain 表起点重新扫描END_VARVAR_IN_OUTstDelivery : ST_MqttBrokerPublishFrame; // 返回给调度层的 Retain 投递任务END_VARVAR_OUTPUTxFound : BOOL; // TRUE 表示本次找到一条命中 Retain 消息END_VARVARuiIndex : UINT; // Retain 表扫描索引END_VAR// === IMPLEMENTATION ===xFound := FALSE;IF xRestart THENuiScanIndex := 1;END_IFIF (uiTargetSlot < 1) OR (uiTargetSlot > GVL_MqttBroker.cnMaxClientSlots) THENeLastError := E_MqttBrokerError.uiInvalidState;M_FindNextRetain := FALSE;RETURN;END_IFIF NOT F_MqttIsValidTopicFilter(sFilter := sTopicFilter) THENeLastError := E_MqttBrokerError.uiInvalidTopic;M_FindNextRetain := FALSE;RETURN;END_IFIF uiScanIndex < 1 THENuiScanIndex := 1;END_IFFOR uiIndex := uiScanIndex TO GVL_MqttBroker.cnMaxRetainedMessages DOuiScanIndex := uiIndex + 1;// 与普通路由一样，Retain 匹配也必须先确认表项有效，再调用 Topic 匹配函数。// 这能避免运行时非短路求值导致空 Retain 主题参与匹配。IF aRetained[uiIndex].xUsed THENIF F_MqttTopicMatch(sFilter := sTopicFilter, sTopic := aRetained[uiIndex].sTopic) THENstDelivery.xValid := TRUE;stDelivery.uiSourceSlot := 0;stDelivery.uiTargetSlot := uiTargetSlot;stDelivery.uiPacketId := 0;stDelivery.eQoS := F_MqttMinQoS(ePublishQoS := aRetained[uiIndex].eQoS, eMaxQoS := eMaxQoS);stDelivery.xDup := FALSE;stDelivery.xRetain := TRUE;stDelivery.uiTopicLen := aRetained[uiIndex].uiTopicLen;stDelivery.uiPayloadLen := aRetained[uiIndex].uiPayloadLen;stDelivery.sTopic := aRetained[uiIndex].sTopic;stDelivery.sPayload := aRetained[uiIndex].sPayload;xFound := TRUE;eLastError := E_MqttBrokerError.uiNoError;M_FindNextRetain := TRUE;RETURN;END_IFEND_IFEND_FOReLastError := E_MqttBrokerError.uiNoError;M_FindNextRetain := TRUE;
完整代码 08: FB_MqttBrokerRouter.M_FindNextRoute.st
/// =======================================================================/// 名称 : M_FindNextRoute/// 功能 : 查找发布消息的下一条订阅匹配结果/// 说明 : 外部循环调用本方法生成投递任务，每次最多返回一条，避免一次扫描阻塞扫描周期。/// 编程人员 : ControlRookie/// 时间 : 2026-05-08/// 版本 : V1.0/// ======================================================================={attribute 'hide_all_locals'}METHOD M_FindNextRoute : BOOLVAR_INPUTstPublish : ST_MqttBrokerPublishFrame; // 需要路由的发布帧xRestart : BOOL; // TRUE 表示从订阅表起点重新扫描END_VARVAR_IN_OUTstDelivery : ST_MqttBrokerPublishFrame; // 返回给调度层的投递任务END_VARVAR_OUTPUTxFound : BOOL; // TRUE 表示本次找到一条匹配订阅END_VARVARuiIndex : UINT; // 订阅表扫描索引END_VAR// === IMPLEMENTATION ===xFound := FALSE;stDelivery := stPublish;IF xRestart THENuiScanIndex := 1;END_IFIF NOT stPublish.xValid THENM_FindNextRoute := FALSE;RETURN;END_IFIF uiScanIndex < 1 THENuiScanIndex := 1;END_IFFOR uiIndex := uiScanIndex TO GVL_MqttBroker.cnMaxSubscriptions DOuiScanIndex := uiIndex + 1;// IEC ST 不要求 AND 必须短路求值，Topic 匹配这类带字符串下标的函数必须放到最内层。// 这样未使用或未激活的订阅不会进入 F_MqttTopicMatch，避免空字符串/旧数据干扰路由。// MQTT 3.1.1 没有 noLocal 订阅选项，发布者如果自己也订阅了同一主题，也应该收到 Broker 回投。IF aSubscriptions[uiIndex].xUsed THENIF aSubscriptions[uiIndex].xActive THENIF F_MqttTopicMatch(sFilter := aSubscriptions[uiIndex].sTopicFilter, sTopic := stPublish.sTopic) THENstDelivery.uiTargetSlot := aSubscriptions[uiIndex].uiSlot;stDelivery.uiPacketId := 0;stDelivery.eQoS := F_MqttMinQoS(ePublishQoS := stPublish.eQoS, eMaxQoS := aSubscriptions[uiIndex].eMaxQoS);stDelivery.xDup := FALSE;xFound := TRUE;M_FindNextRoute := TRUE;RETURN;END_IFEND_IFEND_IFEND_FORM_FindNextRoute := TRUE;
完整代码 09: FB_MqttBrokerRouter.M_RemoveSubscription.st
/// =======================================================================/// 名称 : M_RemoveSubscription/// 功能 : 删除指定槽位的订阅表项/// 说明 : UNSUBSCRIBE 成功解析后调用，按槽位和 Topic Filter 精确删除。/// 编程人员 : ControlRookie/// 时间 : 2026-05-08/// 版本 : V1.0/// ======================================================================={attribute 'hide_all_locals'}METHOD M_RemoveSubscription : BOOLVAR_INPUTuiSlot : UINT; // 取消订阅所属客户端槽位编号[1..cnMaxClientSlots]sTopicFilter : STRING; // 需要删除的 Topic FilterEND_VARVARuiIndex : UINT; // 订阅表扫描索引END_VAR// === IMPLEMENTATION ===IF (uiSlot < 1) OR (uiSlot > GVL_MqttBroker.cnMaxClientSlots) THENeLastError := E_MqttBrokerError.uiInvalidState;M_RemoveSubscription := FALSE;RETURN;END_IFFOR uiIndex := 1 TO GVL_MqttBroker.cnMaxSubscriptions DOIF aSubscriptions[uiIndex].xUsedAND (aSubscriptions[uiIndex].uiSlot = uiSlot)AND (aSubscriptions[uiIndex].sTopicFilter = sTopicFilter) THENaSubscriptions[uiIndex].xUsed := FALSE;aSubscriptions[uiIndex].xActive := FALSE;aSubscriptions[uiIndex].uiSlot := 0;aSubscriptions[uiIndex].sClientId := '';aSubscriptions[uiIndex].sTopicFilter := '';aSubscriptions[uiIndex].uiFilterLen := 0;aSubscriptions[uiIndex].eMaxQoS := E_MqttQoS.byQoS0;IF uiSubscriptionCount > 0 THENuiSubscriptionCount := uiSubscriptionCount - 1;END_IFeLastError := E_MqttBrokerError.uiNoError;M_RemoveSubscription := TRUE;RETURN;END_IFEND_FOReLastError := E_MqttBrokerError.uiNoError;M_RemoveSubscription := TRUE;
完整代码 10: FB_MqttBrokerRouter.M_UpdateRetain.st
/// =======================================================================/// 名称 : M_UpdateRetain/// 功能 : 更新或清除 Retain 保留消息/// 说明 : Retain PUBLISH 载荷为空时清除对应 Topic，否则新增或覆盖保存。/// 编程人员 : ControlRookie/// 时间 : 2026-05-08/// 版本 : V1.0/// ======================================================================={attribute 'hide_all_locals'}METHOD M_UpdateRetain : BOOLVAR_INPUTstPublish : ST_MqttBrokerPublishFrame; // 收到的 Retain PUBLISH 标准发布帧udiNowMs : ULINT; // 当前系统时间戳[ms]END_VARVARuiIndex : UINT; // Retain 表扫描索引uiFreeIndex : UINT; // 首个空闲 Retain 表槽位索引END_VAR// === IMPLEMENTATION ===IF NOT stPublish.xRetain THENM_UpdateRetain := TRUE;RETURN;END_IFIF NOT F_MqttIsValidTopicName(sTopic := stPublish.sTopic) THENeLastError := E_MqttBrokerError.uiInvalidTopic;M_UpdateRetain := FALSE;RETURN;END_IFuiFreeIndex := 0;FOR uiIndex := 1 TO GVL_MqttBroker.cnMaxRetainedMessages DOIF aRetained[uiIndex].xUsed THENIF aRetained[uiIndex].sTopic = stPublish.sTopic THENIF stPublish.uiPayloadLen = 0 THENaRetained[uiIndex].xUsed := FALSE;aRetained[uiIndex].sTopic := '';aRetained[uiIndex].sPayload := '';aRetained[uiIndex].uiTopicLen := 0;aRetained[uiIndex].uiPayloadLen := 0;aRetained[uiIndex].eQoS := E_MqttQoS.byQoS0;aRetained[uiIndex].udiUpdatedAtMs := udiNowMs;IF uiRetainCount > 0 THENuiRetainCount := uiRetainCount - 1;END_IFELSEaRetained[uiIndex].eQoS := stPublish.eQoS;aRetained[uiIndex].uiTopicLen := stPublish.uiTopicLen;aRetained[uiIndex].uiPayloadLen := stPublish.uiPayloadLen;aRetained[uiIndex].udiUpdatedAtMs := udiNowMs;aRetained[uiIndex].sTopic := stPublish.sTopic;aRetained[uiIndex].sPayload := stPublish.sPayload;END_IFeLastError := E_MqttBrokerError.uiNoError;M_UpdateRetain := TRUE;RETURN;END_IFELSIF uiFreeIndex = 0 THENuiFreeIndex := uiIndex;END_IFEND_FORIF stPublish.uiPayloadLen = 0 THENeLastError := E_MqttBrokerError.uiNoError;M_UpdateRetain := TRUE;RETURN;END_IFIF uiFreeIndex = 0 THENeLastError := E_MqttBrokerError.uiRetainFull;M_UpdateRetain := FALSE;RETURN;END_IFaRetained[uiFreeIndex].xUsed := TRUE;aRetained[uiFreeIndex].eQoS := stPublish.eQoS;aRetained[uiFreeIndex].uiTopicLen := stPublish.uiTopicLen;aRetained[uiFreeIndex].uiPayloadLen := stPublish.uiPayloadLen;aRetained[uiFreeIndex].udiUpdatedAtMs := udiNowMs;aRetained[uiFreeIndex].sTopic := stPublish.sTopic;aRetained[uiFreeIndex].sPayload := stPublish.sPayload;uiRetainCount := uiRetainCount + 1;eLastError := E_MqttBrokerError.uiNoError;M_UpdateRetain := TRUE;
完整代码 11: FB_MqttBrokerRouter.st
/// =======================================================================/// 名称 : FB_MqttBrokerRouter/// 功能 : MQTT Broker 主题路由与 Retain 管理/// 说明 : 维护全局订阅表和 Retain 表，只生成投递任务，不直接执行 TCP 发送。/// 编程人员 : ControlRookie/// 时间 : 2026-05-08/// 版本 : V1.0/// ======================================================================={attribute 'hide_all_locals'}FUNCTION_BLOCK FB_MqttBrokerRouterVAR_OUTPUTuiSubscriptionCount : UINT; // 当前全局有效订阅数量uiRetainCount : UINT; // 当前 Retain 表有效消息数量eLastError : E_MqttBrokerError; // 路由器最近一次错误码END_VARVARaSubscriptions : ARRAY[1..GVL_MqttBroker.cnMaxSubscriptions] OF ST_MqttBrokerSubscription; // 全局订阅表aRetained : ARRAY[1..GVL_MqttBroker.cnMaxRetainedMessages] OF ST_MqttBrokerRetainedMessage; // 全局 Retain 表uiScanIndex : UINT; // 路由扫描时使用的订阅表索引END_VAR// === IMPLEMENTATION ===
完整代码 12: F_MqttMinQoS.st
/// =======================================================================/// 名称 : F_MqttMinQoS/// 功能 : 选择发布 QoS 与订阅最大 QoS 中较低等级/// 说明 : MQTT Broker 投递时不能超过订阅者请求的最大 QoS。/// 编程人员 : ControlRookie/// 时间 : 2026-05-08/// 版本 : V1.0/// ======================================================================={attribute 'hide_all_locals'}FUNCTION F_MqttMinQoS : E_MqttQoSVAR_INPUTePublishQoS : E_MqttQoS; // 发布者原始 PUBLISH QoS 等级eMaxQoS : E_MqttQoS; // 订阅者在 SUBSCRIBE 中请求的最大 QoS 等级END_VAR// === IMPLEMENTATION ===IF TO_BYTE(ePublishQoS) <= TO_BYTE(eMaxQoS) THENF_MqttMinQoS := ePublishQoS;ELSEF_MqttMinQoS := eMaxQoS;END_IF
完整代码 13: F_MqttTopicMatch.st
/// =======================================================================/// 名称 : F_MqttTopicMatch/// 功能 : MQTT Topic Filter 与 Topic Name 匹配/// 说明 : 支持 + 单层通配符和 # 多层通配符，用于 Broker 路由扫描订阅表。/// 编程人员 : ControlRookie/// 时间 : 2026-05-08/// 版本 : V1.1/// ======================================================================={attribute 'hide_all_locals'}FUNCTION F_MqttTopicMatch : BOOLVAR_INPUTsFilter : STRING; // 订阅表中的 Topic Filter，可包含 + 或 #sTopic : STRING; // PUBLISH 报文中的 Topic Name，不允许包含通配符END_VARVARuiFilterLen : UINT; // Topic Filter 长度[byte]uiTopicLen : UINT; // Topic Name 长度[byte]uiFilterPos : UINT; // 当前检查的 Topic Filter 字符位置，CodeSys/CODESYS STRING 下标从 0 开始[byte]uiTopicPos : UINT; // 当前检查的 Topic Name 字符位置，CodeSys/CODESYS STRING 下标从 0 开始[byte]byFilterChar : BYTE; // 当前 Topic Filter 字符byTopicChar : BYTE; // 当前 Topic Name 字符END_VAR// === IMPLEMENTATION ===IF NOT F_MqttIsValidTopicFilter(sFilter := sFilter) THENF_MqttTopicMatch := FALSE;RETURN;END_IFIF NOT F_MqttIsValidTopicName(sTopic := sTopic) THENF_MqttTopicMatch := FALSE;RETURN;END_IFuiFilterLen := TO_UINT(LEN(sFilter));uiTopicLen := TO_UINT(LEN(sTopic));uiFilterPos := 0;uiTopicPos := 0;// 关键坑位：// CodeSys/CODESYS 的 STRING 字符访问按 0 基下标工作，和很多工程师直觉中的 1 基字符串不同。// CONNECT 协议名、SUBSCRIBE Topic Filter、PUBLISH Topic/Payload 都会经过这类字符串函数。// 如果这里按 1 基访问，现象通常不是立即编译失败，而是订阅路由错位、发布后客户端被协议错误断开。WHILE (uiFilterPos < uiFilterLen) AND (uiTopicPos < uiTopicLen) DObyFilterChar := sFilter[uiFilterPos];byTopicChar := sTopic[uiTopicPos];CASE byFilterChar OF16#23:// # 匹配当前层级以及后续所有层级；合法性已由 F_MqttIsValidTopicFilter 保证。F_MqttTopicMatch := TRUE;RETURN;16#2B:// + 匹配一个完整层级：跳过 Topic 中直到下一个 / 之前的所有字符。WHILE uiTopicPos < uiTopicLen DOIF sTopic[uiTopicPos] = 16#2F THENEXIT;END_IFuiTopicPos := uiTopicPos + 1;END_WHILEuiFilterPos := uiFilterPos + 1;IF uiFilterPos < uiFilterLen THENIF sFilter[uiFilterPos] = 16#2F THENIF uiTopicPos < uiTopicLen THENIF sTopic[uiTopicPos] = 16#2F THENuiFilterPos := uiFilterPos + 1;uiTopicPos := uiTopicPos + 1;ELSEF_MqttTopicMatch := FALSE;RETURN;END_IFELSEF_MqttTopicMatch := FALSE;RETURN;END_IFEND_IFEND_IF16#2F:IF byTopicChar <> 16#2F THENF_MqttTopicMatch := FALSE;RETURN;END_IFuiFilterPos := uiFilterPos + 1;uiTopicPos := uiTopicPos + 1;ELSEIF byFilterChar <> byTopicChar THENF_MqttTopicMatch := FALSE;RETURN;END_IFuiFilterPos := uiFilterPos + 1;uiTopicPos := uiTopicPos + 1;END_CASEEND_WHILEIF (uiFilterPos = uiFilterLen) AND (uiTopicPos = uiTopicLen) THENF_MqttTopicMatch := TRUE;ELSEF_MqttTopicMatch := FALSE;// IEC ST 的 AND 在部分运行时不保证像高级语言一样短路求值。// 因此带 STRING 下标访问的边界判断必须拆成嵌套 IF，不能写成一个长布尔表达式。IF uiFilterPos < uiFilterLen THENIF sFilter[uiFilterPos] = 16#23 THENF_MqttTopicMatch := TRUE;ELSIF sFilter[uiFilterPos] = 16#2B THENIF (uiFilterPos + 1) = uiFilterLen THEN// + 可以匹配一个空层级，例如 sport/+ 应匹配 sport/。F_MqttTopicMatch := TRUE;END_IFELSIF uiTopicPos = uiTopicLen THENIF (uiFilterPos + 1) < uiFilterLen THENIF sFilter[uiFilterPos] = 16#2F THENIF sFilter[uiFilterPos + 1] = 16#23 THEN// MQTT 通配规则允许 sport/# 匹配 sport；这里处理 Topic 已结束但 Filter 还剩 /# 的场景。F_MqttTopicMatch := TRUE;END_IFEND_IFEND_IFEND_IFEND_IFEND_IF
这一篇你最该记住的几句话
Broker 源码不要按“文件夹顺序”读，要按“入口、状态、数据、报文、路由、事务”读。 CodeSys ST 工程最容易失控的不是语法，而是对象职责边界混乱。 只要你能把本篇源码对象和在线变量对应起来，后续排查连接、订阅、发布和 QoS 问题就不会乱。
系列导航
第 1 篇：源码加更01_Broker 工程入口、容量边界和数据模型 第 2 篇：源码加更02_FB_MqttBroker 顶层调度、连接池和权限边界 第 3 篇：源码加更03_单连接槽位、TCP 字节流和发送队列 第 4 篇：源码加更04_MQTT 编解码器和字节工具函数 第 5 篇：源码加更05_订阅表、Retain、PUBLISH 路由和业务事件 第 6 篇：源码加更06_QoS 事务调度、重试和生产级闭环
