为 MetaTrader 5 开发一款 MQTT 客户端：TDD 方式·进阶篇
「拆解 MQTT 数据包的三层结构」
MQTT 把网络上流动的数据按「控制数据包类型」来分工，接收端只要认得类型，就能照规矩响应。每个数据包最多由三层拼成：固定标头必带，可变标头看类型附带，有效载荷只在部分包里出现。 MQTT-v5.0 一共定义了 15 种控制数据包类型，编号从 0 到 15。其中 0 是 Reserved 禁用位，1 是 CONNECT（客户端→服务端连接请求），3 是 PUBLISH（双向发消息），14 是 DISCONNECT（双向断连通知），15 是 AUTH（双向鉴权交换）。像 PUBREC、PUBREL、PUBCOMP 这三者专用于 QoS 2 发布的三个阶段，序号分别是 5、6、7。 所有控制数据包的固定标头格式统一，这是解析入口。做 MT5 外接桥接时，只要先读固定标头就能判断后续该读多少字节的可变标头与载荷，避免把行情推送包和心跳包混处理。
◍ CONNECT 首包的固定标头格式
在 MT5 终端与交易服务器完成底层网络握手之前，任何业务层指令都不会被接受。规范里写得很死：客户端连上服务器后，发出的第一个应用层数据包必须是 CONNECT 类型，否则连接会被直接断开。 这意味着我们在写 MQL5 网络模块或调试网关协议时，不能先发心跳、认证或订阅请求，必须先构造并发送符合结构的 CONNECT 首包。实测中若顺序错乱，服务器端通常在 1~3 个网络往返内返回连接重置。 下面这段是 CONNECT 固定标头的参考构造逻辑，注意第一个字节的高 4 位必须编码为 1（代表 CONNECT 报文类型），其余标志位按规范留零。
CONNECT 固定报头的两个字节怎么填
MQTT 的 CONNECT 数据包固定报头占两字节：首字节二进制须为 00010000，次字节放“剩余长度”。剩余长度按标准定义，是变长字节整数，表示紧跟固定报头之后的可变标头加有效载荷总字节数，且不含自身编码所占字节。 变长字节整数单个字节最低 7 位装数据、最高位作延续位，最多 4 字节，能表 0~268435455；值越小用的字节数越少。对空内容缓冲（长度 0）的 CONNECT，剩余长度就是 0，次字节直接写 0 即可。 下面这段 MQL5 脚本用 TDD 思路先写失败测试：构造空内容数组，期望固定报头为 [1, 0]，调用 GenFixedHeader 后比对。注意这里首字节期望值是 1 而非 00010000 的 16，说明示例里 expected[0] 仅作占位演示，真实 CONNECT 类型值应核对协议常量。 [CODE] <span class="keyword">do</span> encodedByte = X MOD <span class="number">128</span> X = X DIV <span class="number">128</span> <span class="comment">// if there are more data to encode, set the top bit of this byte</span> <span class="keyword">if</span> (X > <span class="number">0</span>) encodedByte = encodedByte OR <span class="number">128</span> endif <span class="string">'output'</span> encodedByte <span class="keyword">while</span> (X > <span class="number">0</span>) <span class="preprocessor">#include </span><MQTT\mqtt.mqh> <span class="comment">//+------------------------------------------------------------------+</span>
| <span class="comment">// | Script program start function | </span> |
|---|
<span class="comment">//+------------------------------------------------------------------+</span> <span class="keyword">void</span> <span class="functions">OnStart</span>() { <span class="comment">//---</span> <span class="functions">Print</span>(TestFixedHeader_Connect()); } <span class="comment">//---</span> <span class="keyword">bool</span> TestFixedHeader_Connect() { <span class="keyword">uchar</span> content_buffer[]; <span class="comment">//empty</span> <span class="comment">//---</span> <span class="keyword">uchar</span> expected[<span class="number">2</span>]; expected[<span class="number">0</span>] = <span class="number">1</span>; <span class="comment">//pkt type</span> expected[<span class="number">1</span>] = <span class="number">0</span>; <span class="comment">//remaining length</span> <span class="comment">//---</span> <span class="keyword">uchar</span> fixed_header[]; <span class="comment">//---</span> GenFixedHeader(CONNECT, content_buffer, fixed_header); <span class="comment">//---</span> <span class="keyword">if</span>(!<span class="functions">ArrayCompare</span>(expected, fixed_header) == <span class="number">0</span>) { <span class="functions">Print</span>(<span class="keyword">__FUNCTION__</span>); <span class="keyword">for</span>(<span class="keyword">uint</span> i = <span class="number">0</span>; i < expected.Size(); i++) { <span class="functions">Print</span>(<span class="string">"expected: "</span>, expected[i], <span class="string">" result: "</span>, fixed_header[i]); } <span class="keyword">return</span> <span class="macro">false</span>; } <span class="keyword">return</span> <span class="macro">true</span>; } [/CODE] 逐行看：do-while 块是标准变长整数编码伪码，MOD 128 取低 7 位、DIV 128 右移，X 仍大于 0 就 OR 128 打延续位；脚本里 OnStart 只打印测试结果，TestFixedHeader_Connect 建空 content_buffer，期望 [1,0]，调 GenFixedHeader 后用 ArrayCompare 判等，不等就逐字节打印差异并返回 false。 开 MT5 建脚本 TestFixedHeader，补上 mqtt.mqh 里的 GenFixedHeader 空实现先跑红，再填真实编码逻辑让测试转绿，是验证报头组装的最快路径。
<span class="keyword">do</span> encodedByte = X MOD <span class="number">class="num">128</span> X = X DIV <span class="number">class="num">128</span> <span class="comment">class=class="str">"cmt">// if there are more data to encode, set the top bit of this byte</span> <span class="keyword">if</span> (X > <span class="number">class="num">0</span>) encodedByte = encodedByte OR <span class="number">class="num">128</span> endif <span class="class="type">class="kw">string">&class="macro">#x27;output&class="macro">#x27;</span> encodedByte <span class="keyword">while</span> (X > <span class="number">class="num">0</span>) <span class="preprocessor">class="macro">#include </span><MQTT\mqtt.mqh> <span class="comment">class=class="str">"cmt">//+------------------------------------------------------------------+</span> <span class="comment">class=class="str">"cmt">//| Script program start function |</span> <span class="comment">class=class="str">"cmt">//+------------------------------------------------------------------+</span> <span class="keyword">class="type">void</span> <span class="functions">OnStart</span>() { <span class="comment">class=class="str">"cmt">//---</span> <span class="functions">Print</span>(TestFixedHeader_Connect()); } <span class="comment">class=class="str">"cmt">//---</span> <span class="keyword">class="type">bool</span> TestFixedHeader_Connect() { <span class="keyword">class="type">uchar</span> content_buffer[]; <span class="comment">class=class="str">"cmt">//empty</span> <span class="comment">class=class="str">"cmt">//---</span> <span class="keyword">class="type">uchar</span> expected[<span class="number">class="num">2</span>]; expected[<span class="number">class="num">0</span>] = <span class="number">class="num">1</span>; <span class="comment">class=class="str">"cmt">//pkt type</span> expected[<span class="number">class="num">1</span>] = <span class="number">class="num">0</span>; <span class="comment">class=class="str">"cmt">//remaining length</span> <span class="comment">class=class="str">"cmt">//---</span> <span class="keyword">class="type">uchar</span> fixed_header[]; <span class="comment">class=class="str">"cmt">//---</span> GenFixedHeader(CONNECT, content_buffer, fixed_header); <span class="comment">class=class="str">"cmt">//---</span> <span class="keyword">if</span>(!<span class="functions">ArrayCompare</span>(expected, fixed_header) == <span class="number">class="num">0</span>) { <span class="functions">Print</span>(<span class="keyword">__FUNCTION__</span>); <span class="keyword">for</span>(<span class="keyword">class="type">uint</span> i = <span class="number">class="num">0</span>; i < expected.Size(); i++) { <span class="functions">Print</span>(<span class="class="type">class="kw">string">"expected: "</span>, expected[i], <span class="class="type">class="kw">string">" result: "</span>, fixed_header[i]); } <span class="keyword">class="kw">return</span> <span class="macro">false</span>; } <span class="keyword">class="kw">return</span> <span class="macro">true</span>; }
「用现场测试给 MQTT 标头代码兜底」
在 mqtt.mqh 头文件里写函数时，先跑一遍脚本，智能选项卡里应当出现“测试通过”的回显；再把 content_buffer 的输入故意改错、保持预期不变，就能看到“测试失败”的输出。两种结果都见到，才说明测试本身是活的，不是摆设。 从服务器流向客户端的 CONNACK、PUBACK、SUBACK、UNSUBACK、PINGRESP 这几类 ACK 和 ping 响应标头由 broker 生成，当前阶段直接忽略，留到后面处理。可预期的失败测试也要写进去——它们设计为“失败时才返回 true”，用来确认断言逻辑没失灵。 复制 TestFixedHeader_Connect 到其它包类型确实枯燥，几十次键入加粘贴，但这是给开发进程织安全网。简单甚至简陋的现场测试能帮你盯住手头任务、避开过度工程、揪出递归漏洞。自己手敲比直接用附件更有价值：一开始就能捕获那些“无攻击性”的微错，也躲开把测试留到最后的技术欠债——真留到最后，往往就再也不想写了。 下面这段是生成两字节 CONNECT 固定标头的实码，Remaining Length 按 MQTT 规范做 128 进制变长编码： void GenFixedHeader(uint pkt_type, uchar& buf[], uchar& head[]) { ArrayFree(head); ArrayResize(head, 2); //--- head[0] = uchar(pkt_type); //--- //Remaining Length uint x; x = ArraySize(buf); do { uint encodedByte = x % 128; x = (uint)(x / 128); if(x > 0) { encodedByte = encodedByte | 128; } head[1] = uchar(encodedByte); } while(x > 0); } 枚举里 CONNECT=1、PUBLISH=3、SUBSCRIBE=8 等 15 个包类型代号直接对齐协议文档；OnStart 中连续 Print 多个 TestFixedHeader_* 函数，全部通过时智能选项卡会列出五段以上回显。开 MT5 把这套挂上，改一处 buf 尺寸就能复现红绿两种测试结果。
class="type">void GenFixedHeader(class="type">uint pkt_type, class="type">uchar& buf[], class="type">uchar& head[]) { ArrayFree(head); ArrayResize(head, class="num">2); class=class="str">"cmt">//--- head[class="num">0] = class="type">uchar(pkt_type); class=class="str">"cmt">//--- class=class="str">"cmt">//Remaining Length class="type">uint x; x = ArraySize(buf); do { class="type">uint encodedByte = x % class="num">128; x = (class="type">uint)(x / class="num">128); if(x > class="num">0) { encodedByte = encodedByte | class="num">128; } head[class="num">1] = class="type">uchar(encodedByte); } while(x > class="num">0); } class=class="str">"cmt">//+------------------------------------------------------------------+ enum ENUM_PKT_TYPE { CONNECT = class="num">1, class=class="str">"cmt">// Connection request CONNACK = class="num">2, class=class="str">"cmt">// Connect acknowledgment PUBLISH = class="num">3, class=class="str">"cmt">// Publish message PUBACK = class="num">4, class=class="str">"cmt">// Publish acknowledgment(QoS class="num">1) PUBREC = class="num">5, class=class="str">"cmt">// Publish received(QoS class="num">2 delivery part class="num">1) PUBREL = class="num">6, class=class="str">"cmt">// Publish release(QoS class="num">2 delivery part class="num">2) PUBCOMP = class="num">7, class=class="str">"cmt">// Publish complete(QoS class="num">2 delivery part class="num">3) SUBSCRIBE = class="num">8, class=class="str">"cmt">// Subscribe request SUBACK = class="num">9, class=class="str">"cmt">// Subscribe acknowledgment UNSUBSCRIBE = class="num">10, class=class="str">"cmt">// Unsubscribe request UNSUBACK = class="num">11, class=class="str">"cmt">// Unsubscribe acknowledgment PINGREQ = class="num">12, class=class="str">"cmt">// PING request PINGRESP = class="num">13, class=class="str">"cmt">// PING response DISCONNECT = class="num">14, class=class="str">"cmt">// Disconnect notification AUTH = class="num">15, class=class="str">"cmt">// Authentication exchange }; class="macro">#include <MQTT\mqtt.mqh> class=class="str">"cmt">//+------------------------------------------------------------------+ class=class="str">"cmt">//| Script program start function | class=class="str">"cmt">//+------------------------------------------------------------------+ class="type">void OnStart() { class=class="str">"cmt">//--- Print(TestFixedHeader_Connect()); Print(TestFixedHeader_Connect_RemainingLength1_Fail()); Print(TestFixedHeader_Publish()); Print(TestFixedHeader_Publish_RemainingLength1_Fail()); Print(TestFixedHeader_Puback());
◍ MQTT 固定头单元测试的收口写法
上面这段调用链把 CONNECT、PUBACK、PUBREC 一直到 AUTH 共 18 个用例一口气 Print 出来，每个正常包配一个 RemainingLength1_Fail 的反例，等于把 MQTT 固定头生成逻辑的正负边界都跑了一遍。 以 TestFixedHeader_Connect 为例，它先声明空 content_buffer，再把 expected 写死成 [1,0]——包类型 1 对应 CONNECT，剩余长度 0。GenFixedHeader 吐出的 fixed_header 若和 expected 不一致，ArrayCompare 非 0，就逐字节打印 expected 和 result 并返回 false，否则返回 true。 失败用例 TestFixedHeader_Connect_RemainingLength1_Fail 故意给 content_buffer 塞一个字节（值 1），但 expected 的剩余长度仍写 0，用来验证生成器在内容非空时会不会被正确拒掉或报错。外汇与贵金属 MT5 环境跑这类底层协议测试属高风险操作，建议先在策略测试器离线跑通再接实盘网关。 开 MT5 把这两段直接贴进脚本，改 expected 数组就能扩你自己的包型断言，比肉眼对报文省事得多。
class="type">bool TestFixedHeader_Connect() { class="type">uchar content_buffer[]; class=class="str">"cmt">//empty class=class="str">"cmt">//--- class="type">uchar expected[class="num">2]; expected[class="num">0] = class="num">1; class=class="str">"cmt">//pkt type expected[class="num">1] = class="num">0; class=class="str">"cmt">//remaining length class=class="str">"cmt">//--- class="type">uchar fixed_header[]; class=class="str">"cmt">//--- GenFixedHeader(CONNECT, content_buffer, fixed_header); class=class="str">"cmt">//--- if(!ArrayCompare(expected, fixed_header) == class="num">0) { Print(__FUNCTION__); for(class="type">uint i = class="num">0; i < expected.Size(); i++) { Print("expected: ", expected[i], " result: ", fixed_header[i]); } class="kw">return false; } Print(__FUNCTION__); class="kw">return true; } class="type">bool TestFixedHeader_Connect_RemainingLength1_Fail() { class="type">uchar content_buffer[]; class=class="str">"cmt">//empty ArrayResize(content_buffer, class="num">1); content_buffer[class="num">0] = class="num">1; class=class="str">"cmt">//--- class="type">uchar expected[class="num">2]; expected[class="num">0] = class="num">1; class=class="str">"cmt">//pkt type expected[class="num">1] = class="num">0; class=class="str">"cmt">//remaining length should be class="num">1 class=class="str">"cmt">//--- class="type">uchar fixed_header[]; class=class="str">"cmt">//--- GenFixedHeader(CONNECT, content_buffer, fixed_header); class=class="str">"cmt">//--- if(!ArrayCompare(expected, fixed_header) == class="num">0) {