乐于分享
好东西不私藏
夜雨聆风 > > 办公文件 > 源码加更06_QoS 事务调度、重试和生产级闭环

源码加更06_QoS 事务调度、重试和生产级闭环

当前时间： 2026-08-17 18:04:34 分类：办公文件 评论(0)
源码加更06_QoS 事务调度、重试和生产级闭环

源码加更06_QoS 事务调度、重试和生产级闭环

这一组源码加更只有一个目标：把 MqttBroker 的真实 ST 源码按工程阅读顺序讲完整。不是再补几段“看起来像源码”的片段，而是让读者能沿着源码对象理解这个 Broker 怎么组织、怎么运行、怎么排障。

适合谁收藏

  • 已经读过 MqttBroker 主线教程，想继续看真实源码实现的工程师。
  • 想学习 CodeSys ST 工程如何拆分 Broker、连接池、编解码、路由和 QoS 调度的人。
  • 想把 MQTT Broker 移植到 PLC、边缘控制器或教学工程里的开发者。

先给结论

这一篇看 Tx/Rx 两套调度器如何保存 Inflight 事务、推进 PUBACK/PUBREC/PUBREL/PUBCOMP、处理重试，并由顶层周期服务兜底。

QoS1/QoS2 的稳定性不是靠一次 TCP_Write。只要涉及 ACK、PacketId、Inflight、重试和清理，就必须把它当成事务闭环来做。

这篇覆盖 15 个源码文件，合计约 688 行 ST 代码。为了保持公开教程可读性，正文先讲源码阅读路径，再给完整源码。读代码时建议不要从第一个代码块一路机械读到底，而是按本篇的“读代码顺序”来抓主线。

从工程问题到代码职责

层次
本篇重点
你读源码时要抓住的判断
工程入口
程序如何启动、对象如何被实例化
先确认谁是入口，谁只是被调度的对象
数据边界
容量、状态、错误、缓冲区和表结构
先知道边界，后面排障才不会乱猜
协作关系
各 FB、函数和结构体如何互相传递数据
不按文件夹读，按数据流和状态流读
验证路径
在线观察应该看哪些变量
代码最终要能落到现场排障，而不是只停在源码阅读

本篇源码覆盖表

序号
源码对象
行数
1
FB_MqttBroker.M_ServiceRetries.st
68
2
FB_MqttBrokerRxScheduler.M_ClearSlot.st
35
3
FB_MqttBrokerRxScheduler.M_CompletePublish.st
41
4
FB_MqttBrokerRxScheduler.M_CountSlot.st
25
5
FB_MqttBrokerRxScheduler.M_RecordPublish.st
78
6
FB_MqttBrokerRxScheduler.M_TakePublishForPubRel.st
52
7
FB_MqttBrokerRxScheduler.st
18
8
FB_MqttBrokerTxScheduler.M_AckPublish.st
42
9
FB_MqttBrokerTxScheduler.M_ClearSlot.st
35
10
FB_MqttBrokerTxScheduler.M_CountSlot.st
25
11
FB_MqttBrokerTxScheduler.M_FindRetry.st
89
12
FB_MqttBrokerTxScheduler.M_ReceivePubComp.st
42
13
FB_MqttBrokerTxScheduler.M_ReceivePubRec.st
42
14
FB_MqttBrokerTxScheduler.M_RegisterPublish.st
78
15
FB_MqttBrokerTxScheduler.st
18

推荐阅读顺序

  • 先看 TxScheduler，理解 Broker 向客户端投递消息后的等待和重试。
  • 再看 RxScheduler，理解客户端向 Broker 发布 QoS2 时的接收闭环。
  • 最后看 M_ServiceRetries 如何周期性推动未完成事务。

验证和排障边界

  • 使用 QoS1/QoS2 高频发布订阅，观察是否重复、丢失或卡住。
  • 断开客户端后重连，检查 Inflight 清理和重试边界是否符合预期。

本篇完整开源代码

下面代码来自对应 .st 源文件的连续完整内容。为方便公开阅读，只保留源码对象名，不放本机工程路径。

完整代码 01: FB_MqttBroker.M_ServiceRetries.st

/// =======================================================================/// 名称      : M_ServiceRetries/// 功能      : 服务 QoS1 / QoS2 出站事务重发/// 说明      : QoS1/QoS2 PUBLISH 重新进入投递队列，QoS2 PUBREL 进入协议优先队列。/// 编程人员  : ControlRookie/// 时间      : 2026-05-08/// 版本      : V1.0/// ======================================================================={attribute 'hide_all_locals'}METHOD M_ServiceRetries : BOOLVAR    uiRetryIndex : UINT; // 本周期可处理的重试事务扫描索引[1..cnMaxRetryPerScan]END_VAR// === IMPLEMENTATION ===FOR uiRetryIndex := 1 TO GVL_MqttBroker.cnMaxRetryPerScan DO    IF NOT fbTxScheduler.M_FindRetry(        stRetry := stRetry,        udiNowMs := udiNowMs,        xFound => xRetryFound,        xNeedPubRel => xRetryPubRel,        uiPubRelSlot => uiRetryPubRelSlot,        uiPubRelPacketId => uiRetryPubRelPacketId,        xRetryFailed => xRetryFailed,        uiFailedSlot => uiFailedSlot) THEN        EXIT;    END_IF    IF NOT xRetryFailed AND NOT xRetryPubRel AND NOT xRetryFound THEN        // M_FindRetry 在“扫描成功但当前没有任何到期重试事务”时也会返回 TRUE。        // 这里必须立即退出预算循环，否则每个 PLC 周期都会空扫多次 TxInflight 表，        // 在高频发布场景下会明显拉长扫描周期，表现为客户端之间消息转发延迟增大。        EXIT;    END_IF    IF xRetryFailed THEN        // 单条 QoS 出站投递重试耗尽，只丢弃该投递并记录统计，不再主动断开整个 MQTT 客户端。        // 断开连接会导致通信猫/MQTTBox 在连续发布压力测试中表现为“发送一段时间突然掉线”，        // 而工业轻量 Broker 更合适的策略是保护连接、释放事务槽、让后续消息继续流动。        stMetrics.udiPublishDropped := stMetrics.udiPublishDropped + 1;        IF uiFailedSlot <> 0 THEN            M_LogDiag(uiSlot := uiFailedSlot, eError := E_MqttBrokerError.uiQoS1RetryExceeded, sMessage := 'Outbound QoS retry exceeded, message dropped');        END_IF    ELSIF xRetryPubRel THEN        IF (uiRetryPubRelSlot >= 1) AND (uiRetryPubRelSlot <= GVL_MqttBroker.cnMaxClientSlots) THEN            IF aConnections[uiRetryPubRelSlot].M_EnqueueProtocolAck(                ePacketType := E_MqttPacketType.byPubRel,                uiPacketId := uiRetryPubRelPacketId,                byReturnCode := 0) THEN                stMetrics.udiQoS2Retries := stMetrics.udiQoS2Retries + 1;            END_IF        END_IF    ELSIF xRetryFound THEN        IF (stRetry.uiTargetSlot >= 1) AND (stRetry.uiTargetSlot <= GVL_MqttBroker.cnMaxClientSlots) THEN            IF aConnections[stRetry.uiTargetSlot].M_EnqueueDelivery(stDelivery := stRetry) THEN                IF stRetry.eQoS = E_MqttQoS.byQoS1 THEN                    stMetrics.udiQoS1Retries := stMetrics.udiQoS1Retries + 1;                ELSE                    stMetrics.udiQoS2Retries := stMetrics.udiQoS2Retries + 1;                END_IF            ELSE                stMetrics.udiPublishDropped := stMetrics.udiPublishDropped + 1;            END_IF        END_IF    END_IFEND_FORM_ServiceRetries := TRUE;

完整代码 02: FB_MqttBrokerRxScheduler.M_ClearSlot.st

/// =======================================================================/// 名称      : M_ClearSlot/// 功能      : 清理指定客户端的所有入站事务/// 说明      : 客户端断线或槽位释放时调用，避免旧 PacketId 影响新会话。/// 编程人员  : ControlRookie/// 时间      : 2026-05-08/// 版本      : V1.0/// ======================================================================={attribute 'hide_all_locals'}METHOD M_ClearSlot : BOOLVAR_INPUT    uiSlot  : UINT; // 需要清理的客户端槽位编号[1..cnMaxClientSlots]END_VARVAR    uiIndex : UINT; // 入站事务表扫描索引END_VAR// === IMPLEMENTATION ===FOR uiIndex := 1 TO GVL_MqttBroker.cnMaxRxInflight DO    IF aRxInflight[uiIndex].xUsed AND (aRxInflight[uiIndex].uiSlot = uiSlot) THEN        aRxInflight[uiIndex].xUsed := FALSE;        aRxInflight[uiIndex].uiSlot := 0;        aRxInflight[uiIndex].uiPacketId := 0;        aRxInflight[uiIndex].eDirection := E_MqttInflightDirection.iNone;        aRxInflight[uiIndex].eState := E_MqttInflightState.iFree;        aRxInflight[uiIndex].eQoS := E_MqttQoS.byQoS0;        aRxInflight[uiIndex].stPublish.xValid := FALSE;        IF uiInflightCount > 0 THEN            uiInflightCount := uiInflightCount - 1;        END_IF    END_IFEND_FOReLastError := E_MqttBrokerError.uiNoError;M_ClearSlot := TRUE;

完整代码 03: FB_MqttBrokerRxScheduler.M_CompletePublish.st

/// =======================================================================/// 名称      : M_CompletePublish/// 功能      : 完成并清理入站事务/// 说明      : QoS1 在 PUBACK 排队成功后即可清理；QoS2 后续在 PUBCOMP 后清理。/// 编程人员  : ControlRookie/// 时间      : 2026-05-08/// 版本      : V1.0/// ======================================================================={attribute 'hide_all_locals'}METHOD M_CompletePublish : BOOLVAR_INPUT    uiSlot     : UINT; // 入站事务所属客户端槽位编号[1..cnMaxClientSlots]    uiPacketId : UINT; // 需要完成的 Packet IdentifierEND_VARVAR    uiIndex    : UINT; // 入站事务表扫描索引END_VAR// === IMPLEMENTATION ===FOR uiIndex := 1 TO GVL_MqttBroker.cnMaxRxInflight DO    IF aRxInflight[uiIndex].xUsed        AND (aRxInflight[uiIndex].uiSlot = uiSlot)        AND (aRxInflight[uiIndex].uiPacketId = uiPacketId) THEN        aRxInflight[uiIndex].xUsed := FALSE;        aRxInflight[uiIndex].uiSlot := 0;        aRxInflight[uiIndex].uiPacketId := 0;        aRxInflight[uiIndex].eDirection := E_MqttInflightDirection.iNone;        aRxInflight[uiIndex].eState := E_MqttInflightState.iFree;        aRxInflight[uiIndex].eQoS := E_MqttQoS.byQoS0;        aRxInflight[uiIndex].stPublish.xValid := FALSE;        IF uiInflightCount > 0 THEN            uiInflightCount := uiInflightCount - 1;        END_IF        eLastError := E_MqttBrokerError.uiNoError;        M_CompletePublish := TRUE;        RETURN;    END_IFEND_FOReLastError := E_MqttBrokerError.uiNoError;M_CompletePublish := TRUE;

完整代码 04: FB_MqttBrokerRxScheduler.M_CountSlot.st

/// =======================================================================/// 名称      : M_CountSlot/// 功能      : 统计指定客户端槽位的入站事务数量/// 说明      : 用于连接快照诊断，不改变事务状态。/// 编程人员  : ControlRookie/// 时间      : 2026-05-08/// 版本      : V1.0/// ======================================================================={attribute 'hide_all_locals'}METHOD M_CountSlot : UINTVAR_INPUT    uiSlot  : UINT; // 需要统计的客户端槽位编号[1..cnMaxClientSlots]END_VARVAR    uiIndex : UINT; // 入站事务表扫描索引END_VAR// === IMPLEMENTATION ===M_CountSlot := 0;FOR uiIndex := 1 TO GVL_MqttBroker.cnMaxRxInflight DO    IF aRxInflight[uiIndex].xUsed AND (aRxInflight[uiIndex].uiSlot = uiSlot) THEN        M_CountSlot := M_CountSlot + 1;    END_IFEND_FOR

完整代码 05: FB_MqttBrokerRxScheduler.M_RecordPublish.st

/// =======================================================================/// 名称      : M_RecordPublish/// 功能      : 记录入站 QoS>0 PUBLISH 事务/// 说明      : QoS1 用于诊断和重复 PacketId 防护；QoS2 后续在此扩展去重闭环。/// 编程人员  : ControlRookie/// 时间      : 2026-05-08/// 版本      : V1.0/// ======================================================================={attribute 'hide_all_locals'}METHOD M_RecordPublish : BOOLVAR_INPUT    stPublish : ST_MqttBrokerPublishFrame; // 入站发布帧    udiNowMs  : ULINT; // 当前系统时间戳[ms]END_VARVAR    uiIndex   : UINT; // 入站事务表扫描索引    uiFree    : UINT; // 首个空闲入站事务槽位索引END_VAR// === IMPLEMENTATION ===IF stPublish.eQoS = E_MqttQoS.byQoS0 THEN    M_RecordPublish := TRUE;    RETURN;END_IFIF stPublish.uiPacketId = 0 THEN    eLastError := E_MqttBrokerError.uiProtocolMalformed;    M_RecordPublish := FALSE;    RETURN;END_IFuiFree := 0;FOR uiIndex := 1 TO GVL_MqttBroker.cnMaxRxInflight DO    IF aRxInflight[uiIndex].xUsed THEN        IF (aRxInflight[uiIndex].uiSlot = stPublish.uiSourceSlot)            AND (aRxInflight[uiIndex].uiPacketId = stPublish.uiPacketId) THEN            aRxInflight[uiIndex].udiLastActionMs := udiNowMs;            aRxInflight[uiIndex].stPublish := stPublish;            eLastError := E_MqttBrokerError.uiNoError;            M_RecordPublish := TRUE;            RETURN;        END_IF    ELSIF uiFree = 0 THEN        uiFree := uiIndex;    END_IFEND_FORIF uiFree = 0 THEN    eLastError := E_MqttBrokerError.uiRxInflightFull;    M_RecordPublish := FALSE;    RETURN;END_IFaRxInflight[uiFree].xUsed := TRUE;aRxInflight[uiFree].xDupOnRetry := FALSE;aRxInflight[uiFree].xRouteDone := stPublish.eQoS <> E_MqttQoS.byQoS2;aRxInflight[uiFree].uiSlot := stPublish.uiSourceSlot;aRxInflight[uiFree].uiPacketId := stPublish.uiPacketId;aRxInflight[uiFree].uiRetryCount := 0;aRxInflight[uiFree].eDirection := E_MqttInflightDirection.iRx;aRxInflight[uiFree].eQoS := stPublish.eQoS;aRxInflight[uiFree].udiCreatedAtMs := udiNowMs;aRxInflight[uiFree].udiLastActionMs := udiNowMs;aRxInflight[uiFree].stPublish := stPublish;CASE stPublish.eQoS OF    E_MqttQoS.byQoS1:        aRxInflight[uiFree].eState := E_MqttInflightState.iRxPublishSeen;    E_MqttQoS.byQoS2:        aRxInflight[uiFree].eState := E_MqttInflightState.iWaitPubRel;ELSE        aRxInflight[uiFree].eState := E_MqttInflightState.iFault;END_CASEuiInflightCount := uiInflightCount + 1;eLastError := E_MqttBrokerError.uiNoError;M_RecordPublish := TRUE;

完整代码 06: FB_MqttBrokerRxScheduler.M_TakePublishForPubRel.st

/// =======================================================================/// 名称      : M_TakePublishForPubRel/// 功能      : 根据 PUBREL 取出入站 QoS2 发布帧/// 说明      : 只有未路由过的 QoS2 事务会返回发布帧，保证 exactly once 路由语义。/// 编程人员  : ControlRookie/// 时间      : 2026-05-08/// 版本      : V1.0/// ======================================================================={attribute 'hide_all_locals'}METHOD M_TakePublishForPubRel : BOOLVAR_INPUT    uiSlot     : UINT; // 收到 PUBREL 的客户端槽位编号[1..cnMaxClientSlots]    uiPacketId : UINT; // PUBREL 携带的 Packet Identifier    udiNowMs   : ULINT; // 当前系统时间戳[ms]END_VARVAR_IN_OUT    stPublish  : ST_MqttBrokerPublishFrame; // 返回给顶层路由的原始 QoS2 发布帧END_VARVAR_OUTPUT    xNeedRoute : BOOL; // TRUE 表示本次 PUBREL 首次完成，应执行一次路由END_VARVAR    uiIndex    : UINT; // 入站事务表扫描索引END_VAR// === IMPLEMENTATION ===xNeedRoute := FALSE;stPublish.xValid := FALSE;FOR uiIndex := 1 TO GVL_MqttBroker.cnMaxRxInflight DO    IF aRxInflight[uiIndex].xUsed        AND (aRxInflight[uiIndex].uiSlot = uiSlot)        AND (aRxInflight[uiIndex].uiPacketId = uiPacketId)        AND (aRxInflight[uiIndex].eQoS = E_MqttQoS.byQoS2) THEN        aRxInflight[uiIndex].udiLastActionMs := udiNowMs;        IF NOT aRxInflight[uiIndex].xRouteDone THEN            stPublish := aRxInflight[uiIndex].stPublish;            stPublish.xValid := TRUE;            aRxInflight[uiIndex].xRouteDone := TRUE;            aRxInflight[uiIndex].eState := E_MqttInflightState.iCompleted;            xNeedRoute := TRUE;        END_IF        eLastError := E_MqttBrokerError.uiNoError;        M_TakePublishForPubRel := TRUE;        RETURN;    END_IFEND_FOReLastError := E_MqttBrokerError.uiProtocolMalformed;M_TakePublishForPubRel := FALSE;

完整代码 07: FB_MqttBrokerRxScheduler.st

/// =======================================================================/// 名称      : FB_MqttBrokerRxScheduler/// 功能      : Broker 入站 QoS 事务调度器/// 说明      : 首版记录 QoS1 入站 PUBLISH 并立即允许 PUBACK，QoS2 状态位完整预留。/// 编程人员  : ControlRookie/// 时间      : 2026-05-08/// 版本      : V1.0/// ======================================================================={attribute 'hide_all_locals'}FUNCTION_BLOCK FB_MqttBrokerRxSchedulerVAR_OUTPUT    uiInflightCount : UINT; // 当前入站事务表有效数量    eLastError      : E_MqttBrokerError; // 接收事务调度器最近一次错误码END_VARVAR    aRxInflight : ARRAY[1..GVL_MqttBroker.cnMaxRxInflight] OF ST_MqttBrokerInflightMessage; // 入站 QoS>0 事务表END_VAR// === IMPLEMENTATION ===

完整代码 08: FB_MqttBrokerTxScheduler.M_AckPublish.st

/// =======================================================================/// 名称      : M_AckPublish/// 功能      : 处理订阅者返回的 PUBACK/// 说明      : QoS1 出站事务收到匹配 PacketId 后清理；QoS2 后续在此扩展 PUBREC/PUBCOMP。/// 编程人员  : ControlRookie/// 时间      : 2026-05-08/// 版本      : V1.0/// ======================================================================={attribute 'hide_all_locals'}METHOD M_AckPublish : BOOLVAR_INPUT    uiSlot     : UINT; // 返回 PUBACK 的订阅者槽位编号[1..cnMaxClientSlots]    uiPacketId : UINT; // PUBACK 携带的 Packet IdentifierEND_VARVAR    uiIndex    : UINT; // 出站事务表扫描索引END_VAR// === IMPLEMENTATION ===FOR uiIndex := 1 TO GVL_MqttBroker.cnMaxTxInflight DO    IF aTxInflight[uiIndex].xUsed        AND (aTxInflight[uiIndex].uiSlot = uiSlot)        AND (aTxInflight[uiIndex].uiPacketId = uiPacketId)        AND (aTxInflight[uiIndex].eState = E_MqttInflightState.iWaitPubAck) THEN        aTxInflight[uiIndex].xUsed := FALSE;        aTxInflight[uiIndex].uiSlot := 0;        aTxInflight[uiIndex].uiPacketId := 0;        aTxInflight[uiIndex].eDirection := E_MqttInflightDirection.iNone;        aTxInflight[uiIndex].eState := E_MqttInflightState.iFree;        aTxInflight[uiIndex].eQoS := E_MqttQoS.byQoS0;        aTxInflight[uiIndex].stPublish.xValid := FALSE;        IF uiInflightCount > 0 THEN            uiInflightCount := uiInflightCount - 1;        END_IF        eLastError := E_MqttBrokerError.uiNoError;        M_AckPublish := TRUE;        RETURN;    END_IFEND_FOReLastError := E_MqttBrokerError.uiProtocolMalformed;M_AckPublish := FALSE;

完整代码 09: FB_MqttBrokerTxScheduler.M_ClearSlot.st

/// =======================================================================/// 名称      : M_ClearSlot/// 功能      : 清理指定客户端的所有出站事务/// 说明      : 断线或槽位释放时调用，防止旧事务在新连接上误重发。/// 编程人员  : ControlRookie/// 时间      : 2026-05-08/// 版本      : V1.0/// ======================================================================={attribute 'hide_all_locals'}METHOD M_ClearSlot : BOOLVAR_INPUT    uiSlot  : UINT; // 需要清理的客户端槽位编号[1..cnMaxClientSlots]END_VARVAR    uiIndex : UINT; // 出站事务表扫描索引END_VAR// === IMPLEMENTATION ===FOR uiIndex := 1 TO GVL_MqttBroker.cnMaxTxInflight DO    IF aTxInflight[uiIndex].xUsed AND (aTxInflight[uiIndex].uiSlot = uiSlot) THEN        aTxInflight[uiIndex].xUsed := FALSE;        aTxInflight[uiIndex].uiSlot := 0;        aTxInflight[uiIndex].uiPacketId := 0;        aTxInflight[uiIndex].eDirection := E_MqttInflightDirection.iNone;        aTxInflight[uiIndex].eState := E_MqttInflightState.iFree;        aTxInflight[uiIndex].eQoS := E_MqttQoS.byQoS0;        aTxInflight[uiIndex].stPublish.xValid := FALSE;        IF uiInflightCount > 0 THEN            uiInflightCount := uiInflightCount - 1;        END_IF    END_IFEND_FOReLastError := E_MqttBrokerError.uiNoError;M_ClearSlot := TRUE;

完整代码 10: FB_MqttBrokerTxScheduler.M_CountSlot.st

/// =======================================================================/// 名称      : M_CountSlot/// 功能      : 统计指定客户端槽位的出站事务数量/// 说明      : 用于连接快照诊断，不改变事务状态。/// 编程人员  : ControlRookie/// 时间      : 2026-05-08/// 版本      : V1.0/// ======================================================================={attribute 'hide_all_locals'}METHOD M_CountSlot : UINTVAR_INPUT    uiSlot  : UINT; // 需要统计的客户端槽位编号[1..cnMaxClientSlots]END_VARVAR    uiIndex : UINT; // 出站事务表扫描索引END_VAR// === IMPLEMENTATION ===M_CountSlot := 0;FOR uiIndex := 1 TO GVL_MqttBroker.cnMaxTxInflight DO    IF aTxInflight[uiIndex].xUsed AND (aTxInflight[uiIndex].uiSlot = uiSlot) THEN        M_CountSlot := M_CountSlot + 1;    END_IFEND_FOR

完整代码 11: FB_MqttBrokerTxScheduler.M_FindRetry.st

/// =======================================================================/// 名称      : M_FindRetry/// 功能      : 查找需要重发的 QoS1 出站事务/// 说明      : QoS1 重发 PUBLISH；QoS2 在等待 PUBREC 时重发 PUBLISH，在等待 PUBCOMP 时重发 PUBREL。/// 编程人员  : ControlRookie/// 时间      : 2026-05-08/// 版本      : V1.0/// ======================================================================={attribute 'hide_all_locals'}METHOD M_FindRetry : BOOLVAR_INPUT    udiNowMs     : ULINT; // 当前系统时间戳[ms]END_VARVAR_IN_OUT    stRetry      : ST_MqttBrokerPublishFrame; // 返回给发送队列的重发发布帧END_VARVAR_OUTPUT    xFound       : BOOL; // TRUE 表示找到一条需要重发的事务    xNeedPubRel  : BOOL; // TRUE 表示本次需要重发 PUBREL，而不是重发 PUBLISH    uiPubRelSlot : UINT; // 需要重发 PUBREL 的目标槽位编号[1..cnMaxClientSlots]    uiPubRelPacketId : UINT; // 需要重发 PUBREL 的 Packet Identifier    xRetryFailed : BOOL; // TRUE 表示有事务达到最大重试次数    uiFailedSlot : UINT; // 重试失败事务所属槽位编号[1..cnMaxClientSlots]END_VARVAR    uiIndex      : UINT; // 出站事务表扫描索引    udiElapsedMs : ULINT; // 距离该事务上次发送已经过去的时间[ms]END_VAR// === IMPLEMENTATION ===xFound := FALSE;xNeedPubRel := FALSE;uiPubRelSlot := 0;uiPubRelPacketId := 0;xRetryFailed := FALSE;uiFailedSlot := 0;FOR uiIndex := 1 TO GVL_MqttBroker.cnMaxTxInflight DO    IF aTxInflight[uiIndex].xUsed        AND ((aTxInflight[uiIndex].eState = E_MqttInflightState.iWaitPubAck)            OR (aTxInflight[uiIndex].eState = E_MqttInflightState.iWaitPubRec)            OR (aTxInflight[uiIndex].eState = E_MqttInflightState.iWaitPubComp)) THEN        udiElapsedMs := udiNowMs - aTxInflight[uiIndex].udiLastActionMs;        IF udiElapsedMs >= TO_ULINT(GVL_MqttBroker.cnQoS2RetryIntervalMs) THEN            IF aTxInflight[uiIndex].uiRetryCount >= GVL_MqttBroker.cnQoS2MaxRetry THEN                xRetryFailed := TRUE;                uiFailedSlot := aTxInflight[uiIndex].uiSlot;                // 工业轻量 Broker 不能因为单条出站 QoS 投递重试耗尽就长期占住事务槽。                // 这里释放失败事务，由顶层记录丢弃计数和诊断，但不强制断开整个客户端连接。                aTxInflight[uiIndex].xUsed := FALSE;                aTxInflight[uiIndex].uiSlot := 0;                aTxInflight[uiIndex].uiPacketId := 0;                aTxInflight[uiIndex].uiRetryCount := 0;                aTxInflight[uiIndex].eDirection := E_MqttInflightDirection.iNone;                aTxInflight[uiIndex].eState := E_MqttInflightState.iFree;                aTxInflight[uiIndex].eQoS := E_MqttQoS.byQoS0;                aTxInflight[uiIndex].stPublish.xValid := FALSE;                IF uiInflightCount > 0 THEN                    uiInflightCount := uiInflightCount - 1;                END_IF                eLastError := E_MqttBrokerError.uiQoS1RetryExceeded;                M_FindRetry := TRUE;                RETURN;            END_IF            aTxInflight[uiIndex].uiRetryCount := aTxInflight[uiIndex].uiRetryCount + 1;            aTxInflight[uiIndex].udiLastActionMs := udiNowMs;            aTxInflight[uiIndex].xDupOnRetry := TRUE;            IF aTxInflight[uiIndex].eState = E_MqttInflightState.iWaitPubComp THEN                xNeedPubRel := TRUE;                uiPubRelSlot := aTxInflight[uiIndex].uiSlot;                uiPubRelPacketId := aTxInflight[uiIndex].uiPacketId;            ELSE                stRetry := aTxInflight[uiIndex].stPublish;                stRetry.xDup := TRUE;                xFound := TRUE;            END_IF            eLastError := E_MqttBrokerError.uiNoError;            M_FindRetry := TRUE;            RETURN;        END_IF    END_IFEND_FOReLastError := E_MqttBrokerError.uiNoError;M_FindRetry := TRUE;

完整代码 12: FB_MqttBrokerTxScheduler.M_ReceivePubComp.st

/// =======================================================================/// 名称      : M_ReceivePubComp/// 功能      : 处理订阅者返回的 PUBCOMP/// 说明      : QoS2 出站事务收到 PUBCOMP 后完成 exactly once 投递闭环并清理事务。/// 编程人员  : ControlRookie/// 时间      : 2026-05-08/// 版本      : V1.0/// ======================================================================={attribute 'hide_all_locals'}METHOD M_ReceivePubComp : BOOLVAR_INPUT    uiSlot     : UINT; // 返回 PUBCOMP 的订阅者槽位编号[1..cnMaxClientSlots]    uiPacketId : UINT; // PUBCOMP 携带的 Packet IdentifierEND_VARVAR    uiIndex    : UINT; // 出站事务表扫描索引END_VAR// === IMPLEMENTATION ===FOR uiIndex := 1 TO GVL_MqttBroker.cnMaxTxInflight DO    IF aTxInflight[uiIndex].xUsed        AND (aTxInflight[uiIndex].uiSlot = uiSlot)        AND (aTxInflight[uiIndex].uiPacketId = uiPacketId)        AND (aTxInflight[uiIndex].eState = E_MqttInflightState.iWaitPubComp) THEN        aTxInflight[uiIndex].xUsed := FALSE;        aTxInflight[uiIndex].uiSlot := 0;        aTxInflight[uiIndex].uiPacketId := 0;        aTxInflight[uiIndex].eDirection := E_MqttInflightDirection.iNone;        aTxInflight[uiIndex].eState := E_MqttInflightState.iFree;        aTxInflight[uiIndex].eQoS := E_MqttQoS.byQoS0;        aTxInflight[uiIndex].stPublish.xValid := FALSE;        IF uiInflightCount > 0 THEN            uiInflightCount := uiInflightCount - 1;        END_IF        eLastError := E_MqttBrokerError.uiNoError;        M_ReceivePubComp := TRUE;        RETURN;    END_IFEND_FOReLastError := E_MqttBrokerError.uiProtocolMalformed;M_ReceivePubComp := FALSE;

完整代码 13: FB_MqttBrokerTxScheduler.M_ReceivePubRec.st

/// =======================================================================/// 名称      : M_ReceivePubRec/// 功能      : 处理订阅者返回的 PUBREC/// 说明      : QoS2 出站 PUBLISH 收到 PUBREC 后，事务进入等待 PUBCOMP，并由连接层优先发送 PUBREL。/// 编程人员  : ControlRookie/// 时间      : 2026-05-08/// 版本      : V1.0/// ======================================================================={attribute 'hide_all_locals'}METHOD M_ReceivePubRec : BOOLVAR_INPUT    uiSlot     : UINT; // 返回 PUBREC 的订阅者槽位编号[1..cnMaxClientSlots]    uiPacketId : UINT; // PUBREC 携带的 Packet Identifier    udiNowMs   : ULINT; // 当前系统时间戳[ms]END_VARVAR_OUTPUT    xNeedPubRel : BOOL; // TRUE 表示顶层应向该订阅者发送 PUBRELEND_VARVAR    uiIndex     : UINT; // 出站事务表扫描索引END_VAR// === IMPLEMENTATION ===xNeedPubRel := FALSE;FOR uiIndex := 1 TO GVL_MqttBroker.cnMaxTxInflight DO    IF aTxInflight[uiIndex].xUsed        AND (aTxInflight[uiIndex].uiSlot = uiSlot)        AND (aTxInflight[uiIndex].uiPacketId = uiPacketId)        AND (aTxInflight[uiIndex].eState = E_MqttInflightState.iWaitPubRec) THEN        aTxInflight[uiIndex].eState := E_MqttInflightState.iWaitPubComp;        aTxInflight[uiIndex].udiLastActionMs := udiNowMs;        aTxInflight[uiIndex].uiRetryCount := 0;        xNeedPubRel := TRUE;        eLastError := E_MqttBrokerError.uiNoError;        M_ReceivePubRec := TRUE;        RETURN;    END_IFEND_FOReLastError := E_MqttBrokerError.uiProtocolMalformed;M_ReceivePubRec := FALSE;

完整代码 14: FB_MqttBrokerTxScheduler.M_RegisterPublish.st

/// =======================================================================/// 名称      : M_RegisterPublish/// 功能      : 登记 Broker 出站 QoS1/QoS2 PUBLISH 事务/// 说明      : QoS0 不进入事务表；QoS1 等待订阅者 PUBACK；QoS2 状态预留。/// 编程人员  : ControlRookie/// 时间      : 2026-05-08/// 版本      : V1.0/// ======================================================================={attribute 'hide_all_locals'}METHOD M_RegisterPublish : BOOLVAR_INPUT    stPublish : ST_MqttBrokerPublishFrame; // 已发送或即将发送给订阅者的发布帧    udiNowMs  : ULINT; // 当前系统时间戳[ms]END_VARVAR    uiIndex   : UINT; // 出站事务表扫描索引    uiFree    : UINT; // 首个空闲出站事务槽位索引END_VAR// === IMPLEMENTATION ===IF stPublish.eQoS = E_MqttQoS.byQoS0 THEN    M_RegisterPublish := TRUE;    RETURN;END_IFIF stPublish.uiPacketId = 0 THEN    eLastError := E_MqttBrokerError.uiProtocolMalformed;    M_RegisterPublish := FALSE;    RETURN;END_IFuiFree := 0;FOR uiIndex := 1 TO GVL_MqttBroker.cnMaxTxInflight DO    IF aTxInflight[uiIndex].xUsed THEN        IF (aTxInflight[uiIndex].uiSlot = stPublish.uiTargetSlot)            AND (aTxInflight[uiIndex].uiPacketId = stPublish.uiPacketId) THEN            aTxInflight[uiIndex].udiLastActionMs := udiNowMs;            aTxInflight[uiIndex].stPublish := stPublish;            eLastError := E_MqttBrokerError.uiNoError;            M_RegisterPublish := TRUE;            RETURN;        END_IF    ELSIF uiFree = 0 THEN        uiFree := uiIndex;    END_IFEND_FORIF uiFree = 0 THEN    eLastError := E_MqttBrokerError.uiTxInflightFull;    M_RegisterPublish := FALSE;    RETURN;END_IFaTxInflight[uiFree].xUsed := TRUE;aTxInflight[uiFree].xDupOnRetry := FALSE;aTxInflight[uiFree].xRouteDone := TRUE;aTxInflight[uiFree].uiSlot := stPublish.uiTargetSlot;aTxInflight[uiFree].uiPacketId := stPublish.uiPacketId;aTxInflight[uiFree].uiRetryCount := 0;aTxInflight[uiFree].eDirection := E_MqttInflightDirection.iTx;aTxInflight[uiFree].eQoS := stPublish.eQoS;aTxInflight[uiFree].udiCreatedAtMs := udiNowMs;aTxInflight[uiFree].udiLastActionMs := udiNowMs;aTxInflight[uiFree].stPublish := stPublish;CASE stPublish.eQoS OF    E_MqttQoS.byQoS1:        aTxInflight[uiFree].eState := E_MqttInflightState.iWaitPubAck;    E_MqttQoS.byQoS2:        aTxInflight[uiFree].eState := E_MqttInflightState.iWaitPubRec;ELSE        aTxInflight[uiFree].eState := E_MqttInflightState.iFault;END_CASEuiInflightCount := uiInflightCount + 1;eLastError := E_MqttBrokerError.uiNoError;M_RegisterPublish := TRUE;

完整代码 15: FB_MqttBrokerTxScheduler.st

/// =======================================================================/// 名称      : FB_MqttBrokerTxScheduler/// 功能      : Broker 出站 QoS 事务调度器/// 说明      : 管理 Broker 到订阅者方向的 QoS1 在途消息、PUBACK 收口和重发选择。/// 编程人员  : ControlRookie/// 时间      : 2026-05-08/// 版本      : V1.0/// ======================================================================={attribute 'hide_all_locals'}FUNCTION_BLOCK FB_MqttBrokerTxSchedulerVAR_OUTPUT    uiInflightCount : UINT; // 当前出站事务表有效数量    eLastError      : E_MqttBrokerError; // 发送事务调度器最近一次错误码END_VARVAR    aTxInflight : ARRAY[1..GVL_MqttBroker.cnMaxTxInflight] OF ST_MqttBrokerInflightMessage; // 出站 QoS>0 事务表END_VAR// === IMPLEMENTATION ===

这一篇你最该记住的几句话

  • Broker 源码不要按“文件夹顺序”读，要按“入口、状态、数据、报文、路由、事务”读。
  • CodeSys ST 工程最容易失控的不是语法，而是对象职责边界混乱。
  • 只要你能把本篇源码对象和在线变量对应起来，后续排查连接、订阅、发布和 QoS 问题就不会乱。

系列导航

  • 第 1 篇：源码加更01_Broker 工程入口、容量边界和数据模型
  • 第 2 篇：源码加更02_FB_MqttBroker 顶层调度、连接池和权限边界
  • 第 3 篇：源码加更03_单连接槽位、TCP 字节流和发送队列
  • 第 4 篇：源码加更04_MQTT 编解码器和字节工具函数
  • 第 5 篇：源码加更05_订阅表、Retain、PUBLISH 路由和业务事件
  • 第 6 篇：源码加更06_QoS 事务调度、重试和生产级闭环