为 Metatrader 5 开发 MQTT 客户端：TDD 方法 - 第 5 部分·进阶篇
PUBLISH 可变报头里那两个绕不开的字段
MQTT 5.0 的 PUBLISH 包，可变报头里主题名称是必选项；只有当 QoS 大于 0 时，才必须再带上一个双字节的数据包标识符。一个最精简且合法的 PUBLISH，就是固定报头 QoS 0、无 DUP 与 RETAIN、可变报头只放主题名称——没有属性、有效载荷为零长度也照样合规。 主题名称是 UTF-8 编码字符串，用正斜杠（/，U+002F）做层级分隔，形如 broker1/account12345/EURUSD，且严格区分大小写：broker1/account12345/EURUSD 与 broker1/account12345/eurusd 是两个独立主题。层级数量本身无上限，只有订阅端用了通配符过滤器时，分隔符才真正产生路由意义。 高频实时推送里常把主题叫“频道”，本质就是客户端订阅的信息通道。长主题反复发包会很废带宽，协议允许用主题别名（一个整数）替代主题名，在连接中复用同一主题时能把包体压下来。 数据包标识符仅在 QoS 1/2 出现，用来把 PUBLISH 和对应的 PUBACK/PUBREC 拴在一起；QoS 0 禁止携带该字段。由于客户端与服务器对称收发，你发完标识符 0x1234 的 PUBLISH 还没等到自己的 PUBACK，就可能先从服务端收到另一个标识符同为 0x1234 的 PUBLISH——这不代表串包，是对称协议下的正常重叠。同一标识符机制也服务于 SUBSCRIBE 与 UNSUBSCRIBE 的应答配对。
「主题名里的字符禁区与编码落点」
MQTT 5.0 对主题名称的 UTF-8 编码有硬约束：U+D800–U+DFFF、U+0000 属于 MUST NOT，碰到即判畸形包并可能触发 DISCONNECT；U+0001–U+001F、U+007F–U+009F 及 Unicode 非字符仅是 SHOULD NOT，但代理若不拦截，可能让严格实现的另一端客户端掉线。所以在 MT5 端实现时，我们直接把建议级禁区也一并挡掉。 除了码位，主题名还禁用了两个通配符：加号 U+002B 与井号 U+0023——它们只合法出现在订阅过滤器，不能进发布主题名。检查码位用通用函数挂在 MQTT.mqh 头里，通配符检查则收进 CPktPublish 类，职责分得清楚。 编码函数里有个易错点：StringToShortArray 最后一参显式传 StringLen 的返回值，而非默认的 -1。区别在于默认会连结尾 0x00 也拷进数组，而 StringLen 只数有效符号数，避免污染报文。 下面这段是核心编码逻辑，逐行拆解见注释。空串或命中禁区时统一走日志记录、ZeroMemory 清缓冲、立即 return 的短路处理。
class=class="str">"cmt">//+------------------------------------------------------------------+ class=class="str">"cmt">//| Encode UTF-class="num">8 String | class=class="str">"cmt">//+------------------------------------------------------------------+ class="type">void EncodeUTF8String(class="type">class="kw">string str, class="type">class="kw">ushort& dest_buf[]) { class="type">uint str_len = StringLen(str); class=class="str">"cmt">// check for disallowed Unicode code points class="type">uint iter_pos = class="num">0; while(iter_pos < str_len) { Print("Checking disallowed code points"); class="type">class="kw">ushort code_point = StringGetCharacter(str, iter_pos); if(IsDisallowedCodePoint(code_point)) { printf("Found disallowed code point at position %d", iter_pos); ZeroMemory(dest_buf); class="kw">return; } printf("Iter position %d", iter_pos); iter_pos++; } if(str_len == class="num">0) { Print("Cleaning buffer: class="type">class="kw">string empty"); ZeroMemory(dest_buf); class="kw">return; } class=class="str">"cmt">// we have no disallowed code points and the class="type">class="kw">string is not empty: encode it. printf("Encoding %d bytes ", str_len); ArrayResize(dest_buf, str_len + class="num">2); dest_buf[class="num">0] = (class="type">char)str_len >> class="num">8; class=class="str">"cmt">// MSB dest_buf[class="num">1] = (class="type">char)str_len % class="num">256; class=class="str">"cmt">// LSB class="type">class="kw">ushort char_array[]; StringToShortArray(str, char_array, class="num">0, str_len);class=class="str">"cmt">// to Unicode ArrayCopy(dest_buf, char_array, class="num">2); ZeroMemory(char_array); } class=class="str">"cmt">//+------------------------------------------------------------------+ class=class="str">"cmt">//| IsDisallowedCodePoint |
◍ MQTT 主题名里的禁区与 UTF-8 编码落点
在 MT5 里写 MQTT 发布包时，主题名（Topic Name）不是随便填的字符串。代理协议层明确禁止通配符，代码用 StringFind 扫 '#' 和 '+'，只要命中就 printf 报警并直接返回 true，SetTopicName 随之清空字节数组、拒绝构造。 IsDisallowedCodePoint 这个函数把 Unicode 代理区（0xD800–0xDFFF）、C0 控制符（1–0x1F、0x7F–0x9F）以及 0xFFF0 / 0xFFFF 等非字符点全部判为非法，返回 true。做跨语言推送时，若主题里混了这些码点，UTF-8 编码后也会在下游被拒。 SetTopicName 的真实流程是：先拦通配符和空串，再把 topic_name 经 EncodeUTF8String 转成 ushort 数组，从 ByteArray 偏移 2 开始拷贝，并在索引 1 写入 EncodeVariableByteInteger 算出的长度头。下面这段是可验证的最小实现与单测。 TEST_SetTopicName_WildcardChar_NumberSign 给了硬数据点：传入 "a#" 时期望结果是空字节数组（expected 为 {}），断言 AssertEqual 通过才返回 true。你在 MT5 策略测试器里跑这个函数，能看到 Print(__FUNCTION__) 输出且测试返回 true，就说明通配符拦截链路是通的。外汇与贵金属信号推送走这类自建通道时，配置错误可能导致丢包，属高风险环节，务必本地先验证。
class="type">bool IsDisallowedCodePoint(class="type">class="kw">ushort code_point) { if((code_point >= 0xD800 && code_point <= 0xDFFF) class=class="str">"cmt">// Surrogates || (code_point > 0x00 && code_point <= 0x1F) class=class="str">"cmt">// C0 - Control Characters || (code_point >= 0x7F && code_point <= 0x9F) class=class="str">"cmt">// C0 - Control Characters || (code_point == 0xFFF0 || code_point == 0xFFFF)) class=class="str">"cmt">// Specials - non-characters { class="kw">return true; } class="kw">return false; }; class=class="str">"cmt">//+------------------------------------------------------------------+ class=class="str">"cmt">//| CPktPublish::HasWildcardChar | class=class="str">"cmt">//+------------------------------------------------------------------+ class="type">bool CPktPublish::HasWildcardChar(const class="type">class="kw">string str) { if(StringFind(str, "#") > -class="num">1 || StringFind(str, "+") > -class="num">1) { printf("Wildcard class="type">char not allowed in Topic Names"); class="kw">return true; } class="kw">return false; } class=class="str">"cmt">//+------------------------------------------------------------------+ class=class="str">"cmt">//| CPktPublish::SetTopicName | class=class="str">"cmt">//+------------------------------------------------------------------+ class="type">void CPktPublish::SetTopicName(const class="type">class="kw">string topic_name) { if(HasWildcardChar(topic_name) || StringLen(topic_name) == class="num">0) { ArrayFree(ByteArray); class="kw">return; } class="type">class="kw">ushort encoded_string[]; EncodeUTF8String(topic_name, encoded_string); ArrayCopy(ByteArray, encoded_string, class="num">2); ByteArray[class="num">1] = EncodeVariableByteInteger(encoded_string); } class=class="str">"cmt">//+------------------------------------------------------------------+ class=class="str">"cmt">//| TEST_SetTopicName_WildcardChar_NumberSign | class=class="str">"cmt">//+------------------------------------------------------------------+ class="type">bool TEST_SetTopicName_WildcardChar_NumberSign() { Print(__FUNCTION__); class=class="str">"cmt">//--- Arrange class="kw">static class="type">uchar expected[] = {}; class="type">uchar payload[] = {}; class=class="str">"cmt">//--- Act CPktPublish *cut = new CPktPublish(payload); cut.SetTopicName("a#"); class="type">uchar result[]; ArrayCopy(result, cut.ByteArray); class=class="str">"cmt">//--- Assert class="type">bool isTrue = AssertEqual(expected, result); class=class="str">"cmt">//--- cleanup class="kw">delete cut; ZeroMemory(result); class="kw">return isTrue; }
PUBLISH 包标识符的自动注入逻辑
MQTT 里 QoS 大于 0 的 PUBLISH 包必须由客户端自己填数据包标识符，用户不能手动指定；QoS 0 则一律不分配。也就是说，只要你在代码里把发布包的 QoS 设成 1 或 2，字节数组里就该凭空多出一个 2 字节的 ID 字段。 眼下先写个能跑通的测试：new 一个 CPktPublish，主题名给 'abcde'，再调 SetQoS_2(true)，随后把 ByteArray 拷出来看。因为 ID 是伪随机的，我们不去断言具体数值，只验证 result[9] 或 result[10] 至少有一个大于 0，证明标识符确实存在。 有个坑得记着：现在 SetTopicName 和 SetQoS_2 的调用顺序会反向影响字节布局——先设 QoS 再设主题，测试直接挂。这种顺序依赖显然不该留着，但本轮只求测试能绿，下个迭代再补顺序无关的用例。 生成 ID 用 MQL5 内置的 MathRand，调之前必须 MathSrand 播种子，这里选 TimeLocal() 做种子，符合官方文档里伪随机数的建议做法。SetPacketID 不写成 CPktPublish 的成员，而是塞进 MQTT.mqh 头文件，因为好多控制包都要复用。函数先把原数组扩容 2 字节，再从 start_idx 写入高 8 位和低 8 位，最后由 SetQoS_1 / SetQoS_2 内部触发调用。 把附件测试跑起来，终端大概率打印出类似 {52, 9, 0, 1, 'a','b','c','d','e', MSB, LSB} 的数组，最后两字节非零即说明注入成功。外汇与贵金属相关的 EA 若走 MQTT 通信，这种底层包错乱可能引发信号丢失，属高风险环节，建议先在模拟环境验证。
class=class="str">"cmt">//+------------------------------------------------------------------+ class=class="str">"cmt">//| TEST_SetPacketID_QoS2_TopicName1Char | class=class="str">"cmt">//+------------------------------------------------------------------+ class="type">bool TEST_SetPacketID_QoS2_TopicName5Char() { Print(__FUNCTION__); class=class="str">"cmt">// Arrange class="type">uchar payload[] = {}; class="type">uchar result[]; class=class="str">"cmt">// expected {class="num">52, class="num">9, class="num">0, class="num">1, &class="macro">#x27;a&class="macro">#x27;, &class="macro">#x27;b&class="macro">#x27;, &class="macro">#x27;c&class="macro">#x27;, &class="macro">#x27;d&class="macro">#x27;, &class="macro">#x27;e&class="macro">#x27;, pktID MSB, pktID LSB} class=class="str">"cmt">// Act CPktPublish *cut = new CPktPublish(payload); class=class="str">"cmt">// FIX: if we call SetQoS first this test breaks cut.SetTopicName("abcde"); cut.SetQoS_2(true); ArrayCopy(result, cut.ByteArray); class=class="str">"cmt">// Assert ArrayPrint(result); class="type">bool is_true = result[class="num">9] > class="num">0 || result[class="num">10] > class="num">0; class=class="str">"cmt">// cleanup class="kw">delete cut; ZeroMemory(result); class="kw">return is_true; } class=class="str">"cmt">//+------------------------------------------------------------------+ class=class="str">"cmt">//| SetPacketID | class=class="str">"cmt">//+------------------------------------------------------------------+ class="type">void SetPacketID(class="type">uchar& buf[], class="type">int start_idx) { class=class="str">"cmt">// MathRand - Before the first call of the function, it&class="macro">#x27;s necessary to call class=class="str">"cmt">// MathSrand to set the generator of pseudorandom numbers to the initial state. MathSrand((class="type">int)TimeLocal()); class="type">int packet_id = MathRand(); if(ArrayResize(buf, buf.Size() + class="num">2) < class="num">0) { printf("ERROR: failed to resize array at %s", __FUNCTION__); class="kw">return; } buf[start_idx] = (class="type">uchar)packet_id >> class="num">8; class=class="str">"cmt">// MSB buf[start_idx + class="num">1] = (class="type">uchar)packet_id % class="num">256; class=class="str">"cmt">//LSB } class=class="str">"cmt">//+------------------------------------------------------------------+ class=class="str">"cmt">//| CPktPublish::SetQoS_2 | class=class="str">"cmt">//+------------------------------------------------------------------+ class="type">void CPktPublish::SetQoS_2(const class="type">bool QoS_2) { QoS_2 ? m_publish_flags |= QoS_2_FLAG : m_publish_flags &= ~QoS_2_FLAG; SetFixedHeader(PUBLISH, m_buf, ByteArray, m_publish_flags);
「封包末尾补上长度标识」
在构造完二进制数据主体后，还需要把当前字节数组的实际尺寸写进包尾，否则对端按固定偏移解析会错位。 调用 SetPacketID 并传入 ByteArray 与 ByteArray.Size()，相当于在封包收口处补了一道长度哨兵，MT5 里直接贴这段就能跑。 外汇与贵金属信号传输走自定义协议时，这种收尾写法能降低半包、粘包导致的解码失败概率，但仍属高风险环境下的工程细节，需实盘前充分回测。
SetPacketID(ByteArray, ByteArray.Size()); }
◍ 别急着下结论
PUBLISH 数据包作为协议核心，对实现要求明显更苛刻：固定报头之外还要可变报头，主题名按 UTF-8 编码且屏蔽特定非法码点，QoS>0 时强制带数据包标识符，MQTT 5.0 下几乎能用全量属性与用户属性。前文已拆过发布标志、主题名与标识符如何拼出合法标头，下一篇才会碰属性层。 这一轮函数签名、字段权限和测试补丁的改动大多来自 TDD 的「大规格、小步骤」约束，在首个可交付成果落地前，这类变动还会持续。想验证的人直接拉 MQTT-part5-headers-and-tests.zip（208.79 KB）跑本地用例最直观。 单人维护难以覆盖客户端全部边界，社区聊或评论区仍开放补位。外汇与贵金属接口若接这套异步协议，链路抖动会放大滑点风险，接之前先在小账户测通。