为 MetaTrader 5 开发一款 MQTT 客户端：TDD 方式 - 第2部分·进阶篇
（2/3）· 硬编码头能连上却秒断，变量头与协议级才是MQTT握手过关的关键
MQTT服务端拒绝连接的错误码区间
在 MT5 里用 MQL5 接 MQTT 做行情转发或 AIGC 信号推送时，服务端回的 CONNACK/ DISCONNECT 原因码落在 0x86–0x9B 这一段，基本都代表连接层被拒或会话异常，而不是业务数据问题。 下面这组宏把十六进制码和十进制注释对齐了：0x86 是用户名密码错（134），0x87 未授权（135），0x88 服务不可用（136），0x89 服务忙（137），0x8A 被封禁（138），0x8B 服务关闭中（139）。 继续往后看，0x8C–0x9B 覆盖了鉴权方法不对、保活超时、会话被抢、主题非法、包过大、限速超限等情形，例如 0x95 包太大（149）、0x96 消息速率过高（150）、0x9B QoS 不支持（155）。 实盘接 broker 外的 MQTT 中继时，外汇和贵金属本身杠杆高、滑点凶，若 EA 日志频繁刷出 0x88 / 0x89，倾向说明中继不稳，应先断链重连逻辑再做单，别让信号延迟放大风险。
class="macro">#define MQTT_REASON_CODE_BAD_USER_NAME_OR_PASSWORD 0x86 class=class="str">"cmt">// (class="num">134) class="macro">#define MQTT_REASON_CODE_NOT_AUTHORIZED 0x87 class=class="str">"cmt">// (class="num">135) class="macro">#define MQTT_REASON_CODE_SERVER_UNAVAILABLE 0x88 class=class="str">"cmt">// (class="num">136) class="macro">#define MQTT_REASON_CODE_SERVER_BUSY 0x89 class=class="str">"cmt">// (class="num">137) class="macro">#define MQTT_REASON_CODE_BANNED 0x8A class=class="str">"cmt">// (class="num">138) class="macro">#define MQTT_REASON_CODE_SERVER_SHUTTING_DOWN 0x8B class=class="str">"cmt">// (class="num">139) class="macro">#define MQTT_REASON_CODE_BAD_AUTHENTICATION_METHOD 0x8C class=class="str">"cmt">// (class="num">140) class="macro">#define MQTT_REASON_CODE_KEEP_ALIVE_TIMEOUT 0x8D class=class="str">"cmt">// (class="num">141) class="macro">#define MQTT_REASON_CODE_SESSION_TAKEN_OVER 0x8E class=class="str">"cmt">// (class="num">142) class="macro">#define MQTT_REASON_CODE_TOPIC_FILTER_INVALID 0x8F class=class="str">"cmt">// (class="num">143) class="macro">#define MQTT_REASON_CODE_TOPIC_NAME_INVALID 0x90 class=class="str">"cmt">// (class="num">144) class="macro">#define MQTT_REASON_CODE_PACKET_IDENTIFIER_IN_USE 0x91 class=class="str">"cmt">// (class="num">145) class="macro">#define MQTT_REASON_CODE_PACKET_IDENTIFIER_NOT_FOUND 0x92 class=class="str">"cmt">// (class="num">146) class="macro">#define MQTT_REASON_CODE_RECEIVE_MAXIMUM_EXCEEDED 0x93 class=class="str">"cmt">// (class="num">147) class="macro">#define MQTT_REASON_CODE_TOPIC_ALIAS_INVALID 0x94 class=class="str">"cmt">// (class="num">148) class="macro">#define MQTT_REASON_CODE_PACKET_TOO_LARGE 0x95 class=class="str">"cmt">// (class="num">149) class="macro">#define MQTT_REASON_CODE_MESSAGE_RATE_TOO_HIGH 0x96 class=class="str">"cmt">// (class="num">150) class="macro">#define MQTT_REASON_CODE_QUOTA_EXCEEDED 0x97 class=class="str">"cmt">// (class="num">151) class="macro">#define MQTT_REASON_CODE_ADMINISTRATIVE_ACTION 0x98 class=class="str">"cmt">// (class="num">152) class="macro">#define MQTT_REASON_CODE_PAYLOAD_FORMAT_INVALID 0x99 class=class="str">"cmt">// (class="num">153) class="macro">#define MQTT_REASON_CODE_RETAIN_NOT_SUPPORTED 0x9A class=class="str">"cmt">// (class="num">154) class="macro">#define MQTT_REASON_CODE_QOS_NOT_SUPPORTED 0x9B class=class="str">"cmt">// (class="num">155)
「MQTT报文类型与重连原因码拆解」
在 MT5 里用 MQL5 接 MQTT 做行情转发或跨端告警时，先得认全协议层的控制报文类型。报文类型占据固定头首字节的高 4 位，是一个 4 位无符号值，合法区间从 0x01 到 0x0F，共 15 种。 下面这段枚举把 CONNECT(0x01) 到 AUTH(0x0F) 全部列明，其中 PUBLISH 系 QoS 2 的三段式交付由 PUBREC(0x05)、PUBREL(0x06)、PUBCOMP(0x07) 完成，PINGREQ(0x0C) 与 PINGRESP(0x0D) 负责保活。打开 MT5 的 MQEditor，把枚举原样贴进 MQTT.mqh 就能直接编译验证。 另一组服务端返回的原因码集中在 0x9C–0xA2（十进制 156–162）。例如 0x9C 表示建议换服务器、0x9F 是连接频率超限、0xA0 为最大连接时长到期。实盘环境下若 EA 频繁掉线，优先排查 0x9F 与 0xA0 这两个码，外汇与贵金属波动时段连接压力偏高，属高风险场景，断线概率可能明显上升。
class="macro">#define MQTT_REASON_CODE_USE_ANOTHER_SERVER 0x9C class=class="str">"cmt">// (class="num">156) class="macro">#define MQTT_REASON_CODE_SERVER_MOVED 0x9D class=class="str">"cmt">// (class="num">157) class="macro">#define MQTT_REASON_CODE_SHARED_SUBSCRIPTIONS_NOT_SUPPORTED 0x9E class=class="str">"cmt">// (class="num">158) class="macro">#define MQTT_REASON_CODE_CONNECTION_RATE_EXCEEDED 0x9F class=class="str">"cmt">// (class="num">159) class="macro">#define MQTT_REASON_CODE_MAXIMUM_CONNECT_TIME 0xA0 class=class="str">"cmt">// (class="num">160) class="macro">#define MQTT_REASON_CODE_SUBSCRIPTION_IDENTIFIERS_NOT_SUPPORTED 0xA1 class=class="str">"cmt">// (class="num">161) class="macro">#define MQTT_REASON_CODE_WILDCARD_SUBSCRIPTIONS_NOT_SUPPORTED 0xA2 class=class="str">"cmt">// (class="num">162) class=class="str">"cmt">//+------------------------------------------------------------------+ class=class="str">"cmt">//| MQTT.mqh | class=class="str">"cmt">//| ********* WORK IN PROGRESS ********** | class=class="str">"cmt">//| **** PART OF ARTICLE [MQL5官方文档] **** | class=class="str">"cmt">//+------------------------------------------------------------------+ class="macro">#include "Defines.mqh" class=class="str">"cmt">//+------------------------------------------------------------------+ class=class="str">"cmt">//| MQTT - CONTROL PACKET - TYPES | class=class="str">"cmt">//+------------------------------------------------------------------+ class=class="str">"cmt">/* Position: byte class="num">1, bits class="num">7-class="num">4. Represented as a class="num">4-bit unsigned value, the values are shown below. */ enum ENUM_PKT_TYPE { CONNECT = 0x01, class=class="str">"cmt">// Connection request CONNACK = 0x02, class=class="str">"cmt">// Connection Acknowledgment PUBLISH = 0x03, class=class="str">"cmt">// Publish message PUBACK = 0x04, class=class="str">"cmt">// Publish acknowledgment(QoS class="num">1) PUBREC = 0x05, class=class="str">"cmt">// Publish received(QoS class="num">2 delivery part class="num">1) PUBREL = 0x06, class=class="str">"cmt">// Publish release(QoS class="num">2 delivery part class="num">2) PUBCOMP = 0x07, class=class="str">"cmt">// Publish complete(QoS class="num">2 delivery part class="num">3) SUBSCRIBE = 0x08, class=class="str">"cmt">// Subscribe request SUBACK = 0x09, class=class="str">"cmt">// Subscribe acknowledgment UNSUBSCRIBE = 0x0A, class=class="str">"cmt">// Unsubscribe request UNSUBACK = 0x0B, class=class="str">"cmt">// Unsubscribe acknowledgment PINGREQ = 0x0C, class=class="str">"cmt">// PING request PINGRESP = 0x0D, class=class="str">"cmt">// PING response DISCONNECT = 0x0E, class=class="str">"cmt">// Disconnect notification AUTH = 0x0F, class=class="str">"cmt">// Authentication exchange };
◍ MQTT 连接标志位与可变头字段的硬解
MQTT 的 Connect Flags 字节用 8 个比特位描述连接行为，并决定 Payload 里是否携带用户名、密码或遗愿消息等字段。在 MT5 里用枚举把这些位直接映射成十六进制常量，便于后续按位或运算拼装报文头。 枚举 ENUM_CONNECT_FLAGS 中 CLEAN_START 占 0x02，WILL_FLAG 占 0x04，USER_NAME_FLAG 占最高位 0x80；若 Will Flag 为 0，则 Will QoS 必须强制为 0x00，这是协议层的硬约束，写代码时不能留活口。 遗愿消息的 QoS 等级由 Connect Flags 的 bit4 与 bit3 承载，对应 ENUM_QOS_LEVEL：AT_MOST_ONCE=0x00、AT_LEAST_ONCE=0x01、EXACTLY_ONCE=0x02。实盘推送掉线遗愿时，选 EXACTLY_ONCE 可能降低丢单概率，但会增加 broker 端重发开销。 剩下的 SetProtocolVersion、SetProtocolName、SetFixedHeader 都是直接按偏移写 dest_buf 的底层函数：协议名固定占字节 2~7，版本号写偏移 8，固定头第 0 字节左移 4 位塞包类型、第 1 字节放剩余长度。开 MT5 新建脚本把这些函数原样粘进去，能直接看出缓冲区布局有没有错位。 剩余长度从字节 2 起算，是可变字节整数，只统计可变头加 Payload，不含自身编码字节；包总长是固定头长度加剩余长度。外汇与贵金属信号转发走这套协议时波动大、断连频繁，属于高风险链路，建议先在模拟账户验证重连逻辑。
enum ENUM_CONNECT_FLAGS { RESERVED = 0x00, CLEAN_START = 0x02, WILL_FLAG = 0x04, WILL_QOS_1 = 0x08, WILL_QOS_2 = 0x10, WILL_RETAIN = 0x20, PASSWORD_FLAG = 0x40, USER_NAME_FLAG = 0x80 }; enum ENUM_QOS_LEVEL { AT_MOST_ONCE = 0x00, AT_LEAST_ONCE = 0x01, EXACTLY_ONCE = 0x02 }; class="type">void SetProtocolVersion(class="type">uchar& dest_buf[]) { dest_buf[class="num">8] = MQTT_PROTOCOL_VERSION; } class="type">void SetProtocolName(class="type">uchar& dest_buf[]) { dest_buf[class="num">2] = MQTT_PROTOCOL_NAME_LENGTH_MSB; dest_buf[class="num">3] = MQTT_PROTOCOL_NAME_LENGTH_LSB; dest_buf[class="num">4] = MQTT_PROTOCOL_NAME_BYTE_3; dest_buf[class="num">5] = MQTT_PROTOCOL_NAME_BYTE_4; dest_buf[class="num">6] = MQTT_PROTOCOL_NAME_BYTE_5; dest_buf[class="num">7] = MQTT_PROTOCOL_NAME_BYTE_6; } class="type">void SetFixedHeader(ENUM_PKT_TYPE pkt_type, class="type">uchar& buf[], class="type">uchar& dest_buf[]) { dest_buf[class="num">0] = (class="type">uchar)pkt_type << class="num">4; dest_buf[class="num">1] = GetRemainingLength(buf); } class="type">uchar GetRemainingLength(class="type">uchar &buf[]) {
用 128 进制压缩数组长度写入
在自定义指标里往缓冲区塞变长数据时，直接写 ArraySize 会占 4 字节，浪费带宽也拖慢传输。下面这段逻辑把数组长度按 128 进制拆字节：每轮取 x 对 128 的余数作为当前字节，x 自身除以 128 继续，只要还有高位就给当前字节打上 0x80 标记位。 循环终止条件是 x 被除到 0，最后返回的 rem_len 已是末尾字节（无高位标记）。例如数组长 300 时：第一轮 rem_len=44（300%128），x 变 2，打标记得 172；第二轮 rem_len=2，x 变 0，返回 2。实际写盘是两字节 172、2，比固定 4 字节省一半。 外汇与贵金属行情 tick 密集，指标缓冲区常驻数千点，这种压缩在跨终端同步时可能明显降低序列化体积，但高频重算下仍要留意 CPU 开销。
class="type">uint x; x = ArraySize(buf); class="type">uint rem_len; do { rem_len = x % class="num">128; x = (x / class="num">128); if(x > class="num">0) { rem_len = rem_len | class="num">128; } } class="kw">while(x > class="num">0); class="kw">return (class="type">uchar)rem_len; };
「MQTT控制包的对象层次起点」
在MQTT协议栈的EA端实现里，控制数据包的对象层次可以用抽象类或接口来起头。当前阶段我选了接口 IControlPacket 作为根，它只声明了一个 IsControlPacket() 方法，本质上是个占位符，等后面写操作行为部分时，可能改成带虚函数的抽象类。 CONNECT包是协议里写入负担最重的一个。MQTT 5.0相比旧版明显加料，光连接属性和用户属性就够喝一壶，不熟悉协议的人第一次写很容易在变长头（Variable Header）上踩坑。 下面这段是接口与CONNECT变长头的结构骨架，直接在MT5的MQH里建文件就能编译跑通。接口部分只做层次根；CONNECT的可变头用几个struct拆出协议名、保活、属性长度等字段，注意 session_expiry_interval 是 uint 而 receive_maximum 是 ushort，字节序别搞反。 别把接口当万能根 IControlPacket 现在只是花哨占位符，真到实现发布/订阅逻辑时，虚函数抽象类可能更顺手，别在早期就锁死设计。
class=class="str">"cmt">//+------------------------------------------------------------------+ class=class="str">"cmt">//| IControlPacket.mqh | class=class="str">"cmt">//| WORK IN PROGRESS | class=class="str">"cmt">//+------------------------------------------------------------------+ class="macro">#include "MQTT.mqh" class=class="str">"cmt">//+------------------------------------------------------------------+ class=class="str">"cmt">//| Interface IControlPacket | class=class="str">"cmt">//| The root of object hierarchy | class=class="str">"cmt">//+------------------------------------------------------------------+ interface IControlPacket { class="type">bool IsControlPacket(); }; class=class="str">"cmt">//+------------------------------------------------------------------+ class=class="str">"cmt">//+------------------------------------------------------------------+ class=class="str">"cmt">//| PktConnect.mqh | class=class="str">"cmt">//| WORK IN PROGRESS | class=class="str">"cmt">//+------------------------------------------------------------------+ class="macro">#include "MQTT.mqh" class="macro">#include "Defines.mqh" class="macro">#include "IControlPacket.mqh" class=class="str">"cmt">//+------------------------------------------------------------------+ class=class="str">"cmt">//| CONNECT VARIABLE HEADER | class=class="str">"cmt">//+------------------------------------------------------------------+ class=class="str">"cmt">/* The Variable Header for the CONNECT Packet contains the following fields in this order: Protocol Name,Protocol Level, Connect Flags, Keep Alive, and Properties. */ class="kw">struct MqttClientIdentifierLength { class="type">uchar msb; class="type">uchar lsb; } clientIdLen; class=class="str">"cmt">//--- class="kw">struct MqttKeepAlive { class="type">uchar msb; class="type">uchar lsb; } keepAlive; class=class="str">"cmt">//--- class="kw">struct MqttConnectProperties { class="type">uint prop_len; class="type">uchar session_expiry_interval_id; class="type">uint session_expiry_interval; class="type">uchar receive_maximum_id; class="type">class="kw">ushort receive_maximum;
◍ CONNECT 报文的结构体落点
在 MT5 里用 MQL5 封装 MQTT 的 CONNECT 报文，最后两块直接拆成两个结构体：连接属性 connectProps 和载荷 connectPayload。前者管协商参数，后者管客户端身份与遗愿消息。 connectProps 里 maximum_packet_size 用 ushort 承接，MQTT 5.0 规范允许服务端拒绝超过该值的包；topic_alias_maximum 同样为 ushort，决定别名映射上限。request_response_information 与 request_problem_information 均为 uchar 开关位，控制 broker 是否回带额外元信息。 connectPayload 的 correlation_data 与 will_payload、password 都声明为 ulong 数组，说明二进制段按 8 字节对齐存取，不是 string 直接塞。will_delay_interval 和 message_expiry_interval 用 uint，时间窗以秒计，超时后遗愿可能不被分发。 开 MT5 新建 mqh，把这两段原样贴入，编译看 sizeof(connectProps) 与 sizeof(connectPayload) 的字节数，能直接验证你对对齐padding的判断准不准。
class="type">uchar maximum_packet_size_id; class="type">class="kw">ushort maximum_packet_size; class="type">uchar topic_alias_maximum_id; class="type">class="kw">ushort topic_alias_maximum; class="type">uchar request_response_information_id; class="type">uchar request_response_information; class="type">uchar request_problem_information_id; class="type">uchar request_problem_information; class="type">uchar user_property_id; class="type">class="kw">string user_property_key; class="type">class="kw">string user_property_value; class="type">uchar authentication_method_id; class="type">class="kw">string authentication_method; class="type">uchar authentication_data_id; } connectProps; class=class="str">"cmt">//--- class="kw">struct MqttConnectPayload { class="type">uchar client_id_len; class="type">class="kw">string client_id; class="type">class="kw">ushort will_properties_len; class="type">uchar will_delay_interval_id; class="type">uint will_delay_interval; class="type">uchar payload_format_indicator_id; class="type">uchar payload_format_indicator; class="type">uchar message_expiry_interval_id; class="type">uint message_expiry_interval; class="type">uchar content_type_id; class="type">class="kw">string content_type; class="type">uchar response_topic_id; class=class="str">"cmt">// for request/response class="type">class="kw">string response_topic; class="type">uchar correlation_data_id; class=class="str">"cmt">// for request/response class="type">class="kw">ulong correlation_data[]; class=class="str">"cmt">// binary data class="type">uchar user_property_id; class="type">class="kw">string user_property_key; class="type">class="kw">string user_property_value; class="type">uchar will_topic_len; class="type">class="kw">string will_topic; class="type">uchar will_payload_len; class="type">class="kw">ulong will_payload[]; class=class="str">"cmt">// binary data class="type">uchar user_name_len; class="type">class="kw">string user_name; class="type">uchar password_len; class="type">class="kw">ulong password; class=class="str">"cmt">// binary data } connectPayload;