为 MetaTrader 5 开发一款 MQTT 客户端：TDD 方式 - 第2部分·综合运用
◍ MQTT 连接控制包的字节骨架
在 MT5 里用 MQL5 对接 MQTT broker 时，CONNECT 报文是握手的第一道关。下面这个类把控制包结构直接摊进一个 uchar 数组，方便你逐字节核对协议。 CPktConnect 继承自 IControlPacket，私有方法 IsControlPacket() 直接返回 true，表明它本身就是控制包而非数据负载。保护段里 InitConnectFlags、InitKeepAlive、InitPropertiesLength 分别把 ByteArray 的第 10、11、12、13 个索引（0 基下为 9、10、11、12）清零，对应 MQTT 固定头里的连接标志、保活计时和属性长度字段。 公开接口暴露了 SetCleanStart、SetWillFlag、SetPasswordFlag 等一系列位操作方法，以及 SetKeepAlive(ushort seconds) 和 SetClientIdentifier(string clientId)。注意保活值以秒为单位写进两个字节，ClientID 长度与内容分开设置——这两处若顺序写错，broker 会直接断连。 代码里 ByteArray[] 是公开成员，意味着你能在 EA 里直接打印 Hex 对照 Wireshark 抓包。外汇与贵金属行情推送走这套异步通道时波动烈度高，实盘前务必在模拟环境验证字节序。
class=class="str">"cmt">//+------------------------------------------------------------------+ class=class="str">"cmt">//| Class CPktConnect. | class=class="str">"cmt">//| Purpose: Class of MQTT Connect Control Packets. | class=class="str">"cmt">//| Implements IControlPacket | class=class="str">"cmt">//+------------------------------------------------------------------+ class CPktConnect : class="kw">public IControlPacket { class="kw">private: class="type">bool IsControlPacket() {class="kw">return true;} class="kw">protected: class="type">void InitConnectFlags() {ByteArray[class="num">9] = class="num">0;} class="type">void InitKeepAlive() {ByteArray[class="num">10] = class="num">0; ByteArray[class="num">11] = class="num">0;} class="type">void InitPropertiesLength() {ByteArray[class="num">12] = class="num">0;} class="type">uchar m_connect_flags; class="kw">public: CPktConnect(); CPktConnect(class="type">uchar &buf[]); ~CPktConnect(); class=class="str">"cmt">//--- methods for setting Connect Flags class="type">void SetCleanStart(class="kw">const class="type">bool cleanStart); class="type">void SetWillFlag(class="kw">const class="type">bool willFlag); class="type">void SetWillQoS_1(class="kw">const class="type">bool willQoS_1); class="type">void SetWillQoS_2(class="kw">const class="type">bool willQoS_2); class="type">void SetWillRetain(class="kw">const class="type">bool willRetain); class="type">void SetPasswordFlag(class="kw">const class="type">bool passwordFlag); class="type">void SetUserNameFlag(class="kw">const class="type">bool userNameFlag); class="type">void SetKeepAlive(class="type">class="kw">ushort seconds); class="type">void SetClientIdentifierLength(class="type">class="kw">string clientId); class="type">void SetClientIdentifier(class="type">class="kw">string clientId); class=class="str">"cmt">//--- member for getting the byte array class="type">uchar ByteArray[]; }; class=class="str">"cmt">//+------------------------------------------------------------------+ class=class="str">"cmt">//| | class=class="str">"cmt">//+------------------------------------------------------------------+ CPktConnect::CPktConnect(class="type">uchar &buf[]) {
CONNECT 报文里的字节位怎么填
在 MT5 里手搓 MQTT 的 CONNECT 包，核心就是按协议规范把各个字段写进 ByteArray 的固定偏移。下面这段 C++ 类方法展示了客户端标识、保活时间和连接标志位的写入逻辑，直接决定了 broker 能否正确解析你的登录请求。 SetClientIdentifierLength 把字符串长度拆成高位和低位两个字节，分别塞进 ByteArray[12] 和 ByteArray[13]。MQTT 协议规定这两个字节采用大端序，所以长度超过 255 的 clientId 也能靠 >>8 和 %256 正确拆分。 SetKeepAlive 同理，把秒数写到 ByteArray[10]、[11]，注释里标了 MQTT 上限是 65535 秒（约 18.2 小时），实际盯盘用的心跳间隔通常设 60 秒。SetPasswordFlag 和 SetUserNameFlag 则用位运算改 m_connect_flags，再 ArrayFill 回写——注意这里 ArrayFill 的起始长度是 sizeof(ByteArray)，若数组未初始化可能越界，调试时建议先 ArrayResize 确认容量。 开 MT5 新建 EA 把这几段抄进类里，打印 ByteArray 的十六进制，对照 MQTT 3.1.1 规范第 3.1 节就能验证报文头是否对齐。外汇和贵金属行情推送走这套私有协议时波动大、断连概率高，心跳参数别盲目拉满。
class="type">void CPktConnect::SetClientIdentifier(class="type">class="kw">string clientId) { SetClientIdentifierLength(clientId); StringToCharArray(clientId, ByteArray, ByteArray.Size() - StringLen(clientId), StringLen(clientId)); } class="type">void CPktConnect::SetClientIdentifierLength(class="type">class="kw">string clientId) { clientIdLen.msb = (class="type">char)StringLen(clientId) >> class="num">8; clientIdLen.lsb = (class="type">char)StringLen(clientId) % class="num">256; ByteArray[class="num">12] = clientIdLen.msb; ByteArray[class="num">13] = clientIdLen.lsb; } class="type">void CPktConnect::SetKeepAlive(class="type">class="kw">ushort seconds) class=class="str">"cmt">// MQTT max is class="num">65,class="num">535 sec { keepAlive.msb = (class="type">uchar)(seconds >> class="num">8) & class="num">255; keepAlive.lsb = (class="type">uchar)seconds & class="num">255; ByteArray[class="num">10] = keepAlive.msb; ByteArray[class="num">11] = keepAlive.lsb; } class="type">void CPktConnect::SetPasswordFlag(class="kw">const class="type">bool passwordFlag) { passwordFlag ? m_connect_flags |= PASSWORD_FLAG : m_connect_flags &= ~PASSWORD_FLAG; ArrayFill(ByteArray, class="kw">sizeof(ByteArray), class="num">1, m_connect_flags); } class="type">void CPktConnect::SetUserNameFlag(class="kw">const class="type">bool userNameFlag) { userNameFlag ? m_connect_flags |= USER_NAME_FLAG : m_connect_flags &= (class="type">uchar) ~USER_NAME_FLAG;
「连接标志位的位运算与字节回填」
CPktConnect 类用一组 setter 来切换 MQTT 连接报文里的控制标志。每个方法都通过按位或（|=）置位、按位与取反（&= ~）清位来修改 m_connect_flags，再调用 ArrayFill 把更新后的标志写回 ByteArray。 注意 SetCleanStart 与其余方法的不同：它填字节时起始偏移是 9 而不是 sizeof(ByteArray)。这意味着 Clean Start 标志被放在报文第 10 个字节（下标 9）起的长度为 1 的区域，而其他 Will 类标志都从 ByteArray 头部开始填。 下面这段集中展示了五个 setter 的典型写法，你可以直接贴进 MT5 的 MQH 里验证位操作结果。位标记宏（如 WILL_FLAG）需自行在头文件定义，否则编译会报未声明。
class="type">void CPktConnect::SetWillRetain(class="kw">const class="type">bool willRetain) { willRetain ? m_connect_flags |= WILL_RETAIN : m_connect_flags &= ~WILL_RETAIN; ArrayFill(ByteArray, class="kw">sizeof(ByteArray), class="num">1, m_connect_flags); } class="type">void CPktConnect::SetWillQoS_2(class="kw">const class="type">bool willQoS_2) { willQoS_2 ? m_connect_flags |= WILL_QOS_2 : m_connect_flags &= ~WILL_QOS_2; ArrayFill(ByteArray, class="kw">sizeof(ByteArray), class="num">1, m_connect_flags); } class="type">void CPktConnect::SetWillQoS_1(class="kw">const class="type">bool willQoS_1) { willQoS_1 ? m_connect_flags |= WILL_QOS_1 : m_connect_flags &= ~WILL_QOS_1; ArrayFill(ByteArray, class="kw">sizeof(ByteArray), class="num">1, m_connect_flags); } class="type">void CPktConnect::SetWillFlag(class="kw">const class="type">bool willFlag) { willFlag ? m_connect_flags |= WILL_FLAG : m_connect_flags &= ~WILL_FLAG; ArrayFill(ByteArray, class="kw">sizeof(ByteArray), class="num">1, m_connect_flags); } class="type">void CPktConnect::SetCleanStart(class="kw">const class="type">bool cleanStart) { cleanStart ? m_connect_flags |= CLEAN_START : m_connect_flags &= ~CLEAN_START; ArrayFill(ByteArray, class="num">9, class="num">1, m_connect_flags); }
◍ 连接报文类的空构造与析构
在 MT5 终端的网络报文封装层里，CPktConnect 专门承载客户端发起连接时的数据包结构。它的构造函数与析构函数体均为空，说明该类本身不持有需要手动释放的资源，仅作为协议帧的轻量载体存在。 实际调试时若你在 OnTradeTransaction 或自定义 socket 插件中截到这类对象，不必预期构造阶段有任何初始化日志——编译器生成的默认行为就是直接返回。下面这段源码印证了这一点，没有任何成员变量赋值或内存申请。 外汇与贵金属的自动化对接涉及实时链路，这类空实现在高并发发包路径上反而降低了开销，但高频重连场景下仍需警惕连接状态机本身遗漏超时回收。
class=class="str">"cmt">//+------------------------------------------------------------------+ CPktConnect::CPktConnect() { } class=class="str">"cmt">//+------------------------------------------------------------------+ class=class="str">"cmt">//| | class=class="str">"cmt">//+------------------------------------------------------------------+ CPktConnect::~CPktConnect() { } class=class="str">"cmt">//+------------------------------------------------------------------+
用 OASIS 样本字节校验 CONNECT 封包
CPktConnect 类只干一件事：拼出符合 MQTT 规范的 CONNECT 数据包。要验证它没拼错，第一步不是连 broker，而是先准备一个“fixture”——一份手写的正确字节数组，拿它当标尺。 OASIS 在协议文档第 3.1.2.12 节给了可变头的非规范示例，配合我们已经测过的固定头生成器，这份示例足够当起点。参考 fixture 是 {16,24,0,4,77,81,84,84,5,2,0,10,0,4,7,17,0,0,0,10,25,1,77,81,76,53}，共 26 字节；测试时把 CPktConnect 生成的数组用 ArrayCompare 比对，返回 0 才算过。 有人觉得“测数组里是不是我刚塞进去的数据”纯属浪费时间。其实这组看似多余的测试会一直跟着代码，哪天你复制粘贴手滑或改出回归错误，它立刻报警。TDD 过程中很多测试会重写甚至删掉，但留下来的就是长期自动调试器。 本地验证可走 WSL 上的 Mosquitto：默认安装后它是服务，把 80 映射 to 1883，MT5 选项里把主机名加进允许 URL，连上后日志显示无“协议错误”即说明头部格式站得住。图02 的 MetaEditor 专家日志和图03 的 Mosquitto 连接记录都能复现。 别把 fixture 当摆设 硬编码的 expected 数组必须和协议示例逐字节对齐；CleanSession、KeepAlive=10s、ClientID="MQL5" 等组合跑完， broker 不拒包，才说明类可用。外汇/贵金属 EA 接 MQTT 风控信号时，这种底层封包错一个字节都可能让连接静默失败，风险隐蔽且偏高。
class=class="str">"cmt">//+------------------------------------------------------------------+ class=class="str">"cmt">//| TEST_CControlPacket_Connect.mq5 | class=class="str">"cmt">//| | class=class="str">"cmt">//| ********* WORK IN PROGRESS ********** | class=class="str">"cmt">//| **** PART OF ARTICLE [MQL5官方文档] **** | class=class="str">"cmt">//+------------------------------------------------------------------+ class="macro">#include <MQTT\CPktConnect.mqh> class=class="str">"cmt">//+------------------------------------------------------------------+ class=class="str">"cmt">//| Tests for CControlPacketConnect class | class=class="str">"cmt">//+------------------------------------------------------------------+ class="type">void OnStart() { Print(TEST_SetCleanStart_KeepAlive_ClientIdentifier()); Print(TEST_SetClientIdentifier()); Print(TEST_SetClientIdentifierLength()); Print(TEST_SetCleanStart_and_SetKeepAlive()); Print(TEST_SetKeepAlive()); Print(TEST_SetCleanStart()); } class=class="str">"cmt">/* REFERENCE ARRAY(FIXTURE) {class="num">16, class="num">24, class="num">0, class="num">4, class="num">77, class="num">81, class="num">84, class="num">84, class="num">5, class="num">2, class="num">0, class="num">10, class="num">0, class="num">4, class="num">7, class="num">17, class="num">0, class="num">0, class="num">0, class="num">10, class="num">25, class="num">1, class="num">77, class="num">81, class="num">76, class="num">53} */ class=class="str">"cmt">//+------------------------------------------------------------------+ class=class="str">"cmt">//| | class=class="str">"cmt">//+------------------------------------------------------------------+ class="type">bool TEST_SetCleanStart_KeepAlive_ClientIdentifier() { Print(__FUNCTION__); class=class="str">"cmt">//--- Arrange class="kw">static class="type">uchar expected[] = {class="num">16, class="num">16, class="num">0, class="num">4, class="num">77, class="num">81, class="num">84, class="num">84, class="num">5, class="num">2, class="num">0, class="num">10, class="num">0, class="num">4, class="num">77, class="num">81, class="num">76, class="num">53}; class="type">uchar buf[expected.Size() - class="num">2]; CPktConnect *cut = new CPktConnect(buf); class=class="str">"cmt">//--- Act cut.SetCleanStart(true); cut.SetKeepAlive(class="num">10);class=class="str">"cmt">//class="num">10 sec cut.SetClientIdentifier("MQL5"); class="type">uchar result[]; ArrayCopy(result, cut.ByteArray); class=class="str">"cmt">//--- Assert class="type">bool isTrue = Assert(expected, result); class=class="str">"cmt">//--- cleanup class="kw">delete cut;
「连接报文客户端标识的单元测试写法」
在 MT5 里做底层协议封装时，CPktConnect 这类报文构造类必须靠单元测试锁住字节布局，否则升级编译器后可能静默偏移。下面两段测试直接比对期望字节流和实际生成的 ByteArray，属于可复现的验证手段。 TEST_SetClientIdentifier 中，expected 写死 18 字节：头两字节 16,16 后跟 0,4 再是 ASCII 的 M Q T T，末尾带 M Q L 5 的 77,81,76,53。buf 长度取 expected.Size()-2，即 16，构造完调用 SetClientIdentifier("MQL5") 后 ArrayCopy 出结果，Assert 全等才返回 true。 TEST_SetClientIdentifierLength 则只验证长度字段写入：expected 为 14 字节，头部 16,12 表示后续负载缩短到 12，尾部不含 MQL5 字符串本身。两个测试都走 new/delete 配对，并用 ZeroMemory 清 result 防残留。 把这两段塞进你的 Script 或测试 EA，跑完 Print 出函数名即代表用例被执行；若 Assert 失败会返回 false，可据此定位 SetClientIdentifier 内部写入逻辑是否越界。外汇与贵金属自动化涉及真实资金，这类底层报文错误可能引发连接被拒，需在高风险认知下谨慎上线。
class="type">bool TEST_SetClientIdentifier() { Print(__FUNCTION__); class=class="str">"cmt">//--- Arrange class="kw">static class="type">uchar expected[] = {class="num">16, class="num">16, class="num">0, class="num">4, class="num">77, class="num">81, class="num">84, class="num">84, class="num">5, class="num">0, class="num">0, class="num">0, class="num">0, class="num">4, class="num">77, class="num">81, class="num">76, class="num">53}; class="type">uchar buf[expected.Size() - class="num">2]; CPktConnect *cut = new CPktConnect(buf); class=class="str">"cmt">//--- Act cut.SetClientIdentifier("MQL5"); class="type">uchar result[]; ArrayCopy(result, cut.ByteArray); class=class="str">"cmt">//--- Assert class="type">bool isTrue = Assert(expected, result); class=class="str">"cmt">//--- cleanup class="kw">delete cut; ZeroMemory(result); class="kw">return isTrue ? true : class="kw">false; } class="type">bool TEST_SetClientIdentifierLength() { Print(__FUNCTION__); class=class="str">"cmt">//--- Arrange class="kw">static class="type">uchar expected[] = {class="num">16, class="num">12, class="num">0, class="num">4, class="num">77, class="num">81, class="num">84, class="num">84, class="num">5, class="num">0, class="num">0, class="num">0, class="num">0, class="num">4}; class="type">uchar buf[expected.Size() - class="num">2]; CPktConnect *cut = new CPktConnect(buf); class=class="str">"cmt">//--- Act cut.SetClientIdentifierLength("MQL5"); class="type">uchar result[]; ArrayCopy(result, cut.ByteArray); class=class="str">"cmt">//--- Assert class="type">bool isTrue = Assert(expected, result); class=class="str">"cmt">//--- cleanup class="kw">delete cut; ZeroMemory(result); class="kw">return isTrue ? true : class="kw">false; }
◍ MQTT连接包的单元测试写法
在 MT5 里给自定义通信协议类写单测，核心思路是先摆一个静态期望字节数组，再让被测对象吐出实际字节流做逐字节比对。下面这段针对 CPktConnect 的测试，覆盖了 CleanStart 与 KeepAlive 两个字段的编码逻辑。 TEST_SetKeepAlive 里 expected 数组长度为 12，其中第 10、11 位为 0、10，对应 KeepAlive=10 秒的报文编码；而 TEST_SetCleanStart 的 expected 长度为 10，第 9 位为 2，表示 CleanStart 置位。两个用例都通过 expected.Size() - 2 来分配 buf，说明末尾两字节是动态算出来的。 Assert 函数用 ArrayCompare 做全等判断，若不等则逐位 printf 打出 expected 与 result 的十进制值，方便在专家日志里直接定位错位字节。注意 TEST_SetCleanStart 里 ZeroMemory 被注释掉了，意味着 result 数组在函数退出后未强制清零，跑大量用例时可能残留上一次数据，建议手动补上。 开 MT5 新建脚本把这几个函数塞进去，编译后看终端打印的 __FUNCTION__ 与比对结果，能直接验证你的连接包序列化有没有偏一个字节。外汇与贵金属行情推送走自建协议时，这种底层校验可显著降低断连概率，但协议层 bug 仍可能引发滑点，属高风险环节。
class="kw">static class="type">uchar expected[] = {class="num">16, class="num">10, class="num">0, class="num">4, class="num">77, class="num">81, class="num">84, class="num">84, class="num">5, class="num">2, class="num">0, class="num">10}; class="type">uchar buf[expected.Size() - class="num">2]; CPktConnect *cut = new CPktConnect(buf); class=class="str">"cmt">//--- Act cut.SetCleanStart(true); cut.SetKeepAlive(class="num">10); class=class="str">"cmt">//class="num">10 secs class="type">uchar result[]; ArrayCopy(result, cut.ByteArray); class=class="str">"cmt">//--- Assert class="type">bool isTrue = Assert(expected, result); class=class="str">"cmt">//--- cleanup class="kw">delete cut; ZeroMemory(result); class="kw">return isTrue ? true : class="kw">false; } class=class="str">"cmt">//+------------------------------------------------------------------+ class="type">bool TEST_SetKeepAlive() { Print(__FUNCTION__); class=class="str">"cmt">//--- Arrange class="kw">static class="type">uchar expected[] = {class="num">16, class="num">10, class="num">0, class="num">4, class="num">77, class="num">81, class="num">84, class="num">84, class="num">5, class="num">0, class="num">0, class="num">10}; class="type">uchar buf[expected.Size() - class="num">2]; CPktConnect *cut = new CPktConnect(buf); class=class="str">"cmt">//--- Act cut.SetKeepAlive(class="num">10); class=class="str">"cmt">//class="num">10 secs class="type">uchar result[]; ArrayCopy(result, cut.ByteArray); class=class="str">"cmt">//--- Assert class="type">bool isTrue = Assert(expected, result); class=class="str">"cmt">//--- cleanup class="kw">delete cut; ZeroMemory(result); class="kw">return isTrue ? true : class="kw">false; } class=class="str">"cmt">//+------------------------------------------------------------------+ class="type">bool TEST_SetCleanStart() { Print(__FUNCTION__); class=class="str">"cmt">//--- Arrange class="kw">static class="type">uchar expected[] = {class="num">16, class="num">8, class="num">0, class="num">4, class="num">77, class="num">81, class="num">84, class="num">84, class="num">5, class="num">2}; class="type">uchar buf[expected.Size() - class="num">2]; CPktConnect *cut = new CPktConnect(buf); class=class="str">"cmt">//--- Act cut.SetCleanStart(true); class="type">uchar result[]; ArrayCopy(result, cut.ByteArray); class=class="str">"cmt">//--- Assert class="type">bool isTrue = Assert(expected, result); class=class="str">"cmt">//--- cleanup class="kw">delete cut; class=class="str">"cmt">//ZeroMemory(result); class="kw">return isTrue ? true : class="kw">false; } class=class="str">"cmt">//+------------------------------------------------------------------+ class="type">bool Assert(class="type">uchar& expected[], class="type">uchar& result[]) { if(!ArrayCompare(expected, result) == class="num">0) { for(class="type">uint i = class="num">0; i < expected.Size(); i++) { printf("expected\t%d\t\t%d result", expected[i], result[i]); }
用数组比对校验指标输出一致性
在 MT5 自定义指标或 EA 的单元测试里，光看返回值不够，得把预期数组和实际结果数组拉出来逐元素比对。上面这段是校验失败时的打印逻辑：先报出两个数组的尺寸差异，再把 Expected 和 Result 分别用 ArrayPrint 铺开，方便肉眼定位哪一根柱子的数值偏了。 如果 expected.Size() 与 result.Size() 不相等，或者元素值对不上，函数直接 return false，测试中断；全部通过才 return true。实盘前跑一遍这类校验，能提前揪出因缓冲区索引错位导致的信号漂移——外汇与贵金属杠杆高，这类隐蔽 bug 可能放大滑点风险。 把这段代码嵌进你的测试脚本，开 MT5 用策略测试器的自定义品种回放，对比不同时间点算出的数组，十有八九能发现某个偏移参数没复位。
printf("expected size %d <=> %d result size", expected.Size(), result.Size()); Print("Expected"); ArrayPrint(expected); Print("Result"); ArrayPrint(result); class="kw">return class="kw">false; } class="kw">return true; }
「下一步要啃的是 CONNACK 回应」
这一节其实只是个路标：作者说下一步会处理 MQTT 协议里的 CONNACK 回应包，等这块落地，才算真正有了发布第一条消息的底座，而且照例会先写测试再写实现。 随文挂了 6 个 .mqh / .mq5 文件，其中 CPktConnect.mqh 有 9.97 KB、TEST_CControlPacket_Connect.mq5 是 5.89 KB，想跟进的人直接把 ZIP 拖进 MT5 的 MQL5/Include 和 Scripts 就能跑通现有连接测试。 外汇与贵金属行情受杠杆影响大、滑点频发，这类通信层封装只解决「传得通」，不解决「信号对」，实盘前务必在策略测试器里压过极端点差环境。 等 CONNACK 的解析和发布测试补完，这套 TDD 写的 MQTT 客户端才从「能连」走到「能发」，到时候开 MT5 把 TEST 文件跑一遍，比看十行说明都实在。