为 MetaTrader 5 开发一款 MQTT 客户端：TDD 方式

MetaTrader 5 — 积分 | 5 三月 2024, 09:34

727

Jocimar Lopes

概述

"...策略是明确的：首先令其发挥作用，然后令其走上正轨，最后，令其更迅捷..."。Stephen C. Johnson 和 Brian W. Kernighan 的 “系统编程之 C语言与模型”，载于 Byte 杂志（1983 年 8 月）

在两个或多个 MetaTrader 实例之间共享实时数据是交易者和客户经理的共同需求。可以说，最常需要共享的数据与交易业务有关，即所谓的“交易跟单者”所求。但人们能很容易地找到帐户信息共享、品种筛选、和机器学习所需统计数据的请求，仅举几例。此功能可以通过使用网络套接字、与命名管道的进程间通信、Web 服务、本地文件共享、以及可能已测试（和/或）开发的其它解决方案来获得。

作为软件开发中的常态，这些解决方案中的每一种在可用性、稳定性、可信度、以及开发和维护所需的资源方面都有其优点和缺点。简而言之，取决于用户的需求和预算，每一种都代表了不同的成本-效益关系。

本文汇报了 MQTT 协议客户端实现的第一步，其恰好是一种满足这种需求的技术 — 机器之间实时数据共享 — 高性能、低带宽消耗、低资源需求、和低成本。

什么是 MQTT

MQTT 是一种在客户端-服务器之间发布/订阅消息的传输协议。它轻巧、开放、简单，并且易于实施。这些特性令其非常适合在众多状况下使用，包括受限环境，譬如机器对机器（M2M）、和物联网（IoT）环境中的通信，这其中网络带宽非常宝贵，故需占用空间小的代码。

以上定义来自 OASIS，自 2013 年以来，该协议的所有者和开发者将其作为开放标准。

“2013 年，IBM 向 OASIS 规范提交了 MQTT v3.1，并附有一份章程，确保只接受对规范的微小更改。从 IBM 接管标准维护之后，OASIS 于 2014 年 10 月 29 日发布了 3.1.1 版。2019 年 3 月 7 日发布了对 MQTT 版本 5，其有更实质性升级，增加了一些新功能。”（维基百科)

IBM 自 1999 年起开发协议，定位是满足工业需求，即使用传感器监控输油管道，并通过卫星将数据发送到远程控制中心。根据 Arlen Nipper的说法，他与 Andy Stanford-Clark 博士共同创建该协议，目标是为这些控制中心提供实时数据流。

“我们当时试图做的是，IBM 的中间件 MQ Integrator，正是我当时在通信行业所做的工作，讨论 1,200-波特拨号线路和 300-波特拨号线路，以及带宽非常受限的 VSAT，并将这两者绑缚在一起。”

尽管事实上由于技术堆栈的限制和昂贵的网络成本，它被设计为强大、快速和廉价，但它需要提供具有持续会话感知的数据交付服务品质，这令其能够应对不可靠，甚至间断性的互联网连接。

作为一种二进制协议，MQTT 在内存和处理需求方面非常高效。更令人惊奇的是，最小的 MQTT 数据包只有两个字节！

鉴于 MQTT 基于发布/订阅模型（pub/sub），取代了“请求/响应”，故 MQTT 是双向的。也就是说，一旦客户端/服务器连接建立好，数据就可以随时从客户端流向服务器，以及从服务器流向客户端，而无需事先请求，譬如 HTTP 的 WebRequest 的情况。一旦数据到达，服务器会立即将其转发给接收人。此特征是实时数据交换的基石，因为它允许端点之间的最小延迟。一些赞助商广告会有毫秒级的延迟。

类型、格式、编解码器、或有关数据的其它任何方面都无关紧要。MQTT 与数据无关。用户可以发送/接收的数据从原始字节到文本格式（XML、JSON 对象）、协议缓冲区、图片、视频片段、等等。

客户端和服务器之间的大多数交互都可以是异步的，也就是说 MQTT 是可伸缩的。在物联网行业中，话说数千甚至数百万台设备实时连接和交换数据的情况并不少见。

它的消息可以、而且通常是在端点之间加密的，因为该协议与 TLS 兼容，并内置了身份验证和授权机制。

毫不奇怪，MQTT 不仅是一套高标准的规范，而且是多个行业广泛采用的技术。

“如今，MQTT 被广泛应用于各种行业，诸如汽车、制造、电信、石油和天然气、等等。”（mqtt.org)

主要组件

发布/订阅（pub/sub）是一套非常著名的消息交换模型。客户端连接到服务器，并发布有关主题的消息。此后，订阅该主题的所有客户端都会收到消息。这是模型的基本机制。

服务器充当代理，站在客户端之间代收代发。TCP/IP 是底层传输协议，客户端是任何理解 TCP/IP 和 MQTT 的设备。消息通常是 JSON 或 XML 有效载荷，但也可以是任何内容，包括原始字节序列。

该主题是一个 UTF-8 编码的字符串，用于描述类似命名空间的分层结构

office/machine01/account123456
office/machine02/account789012
home/machine01/account345678

我们在订阅主题时还可以使用哈希（#）作为通配符。例如，订阅 home 之下 machine01 中的所有帐户：

home/machine01/#

或者订阅 office 之下所有机器：

office/#

好的，所以 MQTT 是为机器对机器之间对话而开发的，它在物联网环境中被广泛使用，而且它健壮、快速、且廉价。但您也许会问：这个东西能给交易环境带来什么样的益处或改善？MQTT 在 MetaTrader 中的用例是什么？

如上所述，“交易跟单者”是 MQTT 在交易环境中最显见的用例。但人们也许会考虑用实时数据投喂机器学习管道，根据从 Web 服务中抽取的实时数据改变 EA 行为，或者从任意设备远程控制您的 MetaTrader 应用程序。

对于任何需要在机器之间实时数据流的场景，我们也许会把 MQTT 纳入考察。

如何在 MetaTrader 中使用 MQTT

对于最流行的通用语言，有免费和开源的 MQTT 客户端函数库，包括对应移动和嵌入式设备的变体。因此，为了从 MQL5 使用 MQTT，我们可以自 C、C++、或 C# 层面生成并导入相应的 DLL。

如果要共享的数据仅限于业务/账户信息，并且可以接受相对较大的延迟，则另一种选择是使用 Python MQTT 客户端函数库，和 MQL5 Python 模块作为“桥梁”。

但正如我们所知，DLL 的使用对于 MQL5 生态系统有一些负面影响，最值得注意的是市场不接受依赖 DLL 的 EA 上架。此外，在 MQL5 云上不允许运行依赖 DLL 的 EA 进行回测优化。为了避免 DLL 依赖，和 Python 桥接，理想的解决方案是为 MetaTrader 开发一个客户端的原生 MQTT 函数库。

这就是我们将在未来几周内要做的事情：针对 MetaTrader 5 实现客户端 MQTT-v5.0 协议。

与其它网络协议相比，实现 MQTT 客户端可认为是“相对容易的”。但相对容易并非一定容易。因此，我们将按自底向上的方式开始，即由测试驱动的开发（TDD），并希望来自社区的测试和反馈。

尽管 TDD 可以（并且经常）被用作几乎任何东西的“炒作”或“流行语”，但当我们有一套正式的规范时，它却非常适合，这恰是标准化网络协议的情况。

通过采用自底向上的方式，我们面对巨型规格时可以分解它，比方说，婴儿学步。MQTT 的规格并不是庞大，与代理端相比，客户端是最简单的。但它有其自身的复杂性，特别是自 5.0 版以来，它包含了一些附加特征。

由于我们没有带薪时间和团队来合并技能，故婴儿学步似乎于此是最好的方式：我如何发送消息？我应当写什么样的东西？我怎样才能自有效的东西开始，如此在考虑令它工作更迅捷之前，我能改进它，令其运行更优秀？

巨型规格，婴儿学步：理解并分解 MQTT 协议

如同众多（如果不是全部）网络协议的规范，MQTT 协议的工作原理是将所谓的数据包中传输的数据分解含义。因此，如果接收方知道每种数据包的含义，它就可以根据接收到的数据包类型采取正确的操作行为。在 MQTT 术语中，数据包的种类称为控制数据包类型，每个数据包最多有三个部分：

所有数据包中都存在一个固定标头
某些数据包中存在一个可变标头
仅在某些数据包中还存在一个有效载荷

MQTT-v5.0 中有 15 种控制数据包类型：

表格 1. MQTT 控制数据包类型（表格来自 OASIS 规范）

名称	数值	流向	说明
Reserved	0	禁用	Reserved
CONNECT	1	客户端至服务端	连接请求
CONNACK	2	服务器至客户端	连接确认
PUBLISH	3	客户端到服务器或服务器到客户端	发布消息
PUBREC	5	客户端到服务器或服务器到客户端	已收到发布（QoS 2 发行第 1 部分）
PUBREL	6	客户端到服务器或服务器到客户端	发布版本（QoS 2 发行第 2 部分）
PUBCOMP	7	客户端到服务器或服务器到客户端	发布完成（QoS 2 发行第 3 部分）
SUBSCRIBE	8	客户端至服务端	订阅请求
SUBACK	9	服务器至客户端	订阅确认
UNSUBSCRIBE	10	客户端至服务端	退订请求
UNSUBACK	11	服务器至客户端	退订确认
PINGREQ	12	客户端至服务端	测试请求
PINGRESP	13	服务器至客户端	测试响应
DISCONNECT	14	客户端到服务器或服务器到客户端	断连通知
AUTH	15	客户端到服务器或服务器到客户端	身份验证交换

所有控制数据包的固定标头具有相同的格式。

图例.1 MQTT 固定标头格式

MQTT 固定标头格式

鉴于在客户端和服务器之间建立连接之前我们什么也做不了，且参考标准有一个明确的陈述，内容如下

“客户端与服务器建立网络连接后，从客户端发送到服务器的第一个数据包必须是 CONNECT 数据包”，

我们看看 CONNECT 数据包的固定标头应如何格式化。

图例 2 CONNECT 数据包的 MQTT 固定报头格式

CONNECT 数据包的 MQTT 固定报头格式

由此，我们需要用两个字节填充它：第一个字节必须具有二进制值 00010000，第二个字节必须含有所谓的“剩余长度”的值。

该标准将“剩余长度”定义为

“一个可变字节整数，表示当前控制数据包中剩余的字节数，包括可变标头和有效载荷中的数据。剩余长度不包括用来编码剩余长度的字节。数据包大小是 MQTT 控制数据包中的总字节数，这等于固定报头的长度加上剩余长度。”（强调是我们的）

该标准还定义了可变字节整数的编码方案。

“可变字节整数按编码方案进行编码，它使用单个字节表示最大 127 的值。更大的数值则按如下处理。每个字节的最低有效 7 位进行数据编码，最高有效位指示是否有后续字节存在。因此，每个字节编码 128 个值（0~127），和一个‘延续位’。可变字节整数字段中的最大字节数为 4。编码值长度必须取表示该值所需的最少字节数”。

哇！看似需要同时吸收很多信息。我们只是尝试填满第二个字节！

幸运的是，该标准提供了“将非负整数（X）编码为可变字节整数编码方案的算法”。

do
   encodedByte = X MOD 128
   X = X DIV 128
   // if there are more data to encode, set the top bit of this byte
   if (X > 0)
      encodedByte = encodedByte OR 128
   endif
   'output' encodedByte
while (X > 0)

“其中 MOD 是模运算符（C 中的 %），DIV 是整数除法（C 中的 /），OR 是按位或（C 中的 |）。”

好了。现在我们有：

所有控制数据包类型的列表，
具有两个字节的 CONNECT 数据包的固定标头的格式，
第一个字节的值，
以及填充可变字节整数第二个字节的编码算法。

我们可以开始编写我们的第一个测试了。

注意：鉴于我们采用的是自底向而上的 TDD 方式，故我们将在实现之前编写测试。我们可以从一开始就假设我们将 1）编写失败的测试，然后我们将 2）仅实现通过测试所需的代码，然后我们将 3）在需要时重构代码。无论初始实现是否幼稚、丑陋，或者它看似性能不佳，都无关紧要。一旦我们有了能工作的代码，我们就会应对这些问题。性能排在我们任务列表的末尾。

事不宜迟，我们打开我们的 MetaEditor，并创建一个名为 TestFixedHeader 的脚本，包含以下内容。

#include <MQTT\mqtt.mqh>
//+------------------------------------------------------------------+
//| Script program start function                                    |
//+------------------------------------------------------------------+
void OnStart()
  {
//---
   Print(TestFixedHeader_Connect());
  }
//---
bool TestFixedHeader_Connect()
  {
   uchar content_buffer[]; //empty
//---
   uchar expected[2];
   expected[0] = 1; //pkt type
   expected[1] = 0; //remaining length
//---
   uchar fixed_header[];
//---
   GenFixedHeader(CONNECT, content_buffer, fixed_header);
//---
   if(!ArrayCompare(expected, fixed_header) == 0)
     {
      Print(__FUNCTION__);
      for(uint i = 0; i < expected.Size(); i++)
        {
         Print("expected: ", expected[i], " result: ", fixed_header[i]);
        }
      return false;
     }
   return true;
  }

另外，创建 mqtt.mqh 头文件，我们开始在其内开发函数，并用下面的代码填充它。

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);
  }
//+------------------------------------------------------------------+
enum ENUM_PKT_TYPE
  {
   CONNECT      =  1, // Connection request
   CONNACK      =  2, // Connect acknowledgment
   PUBLISH      =  3, // Publish message
   PUBACK       =  4, // Publish acknowledgment (QoS 1)
   PUBREC       =  5, // Publish received (QoS 2 delivery part 1)
   PUBREL       =  6, // Publish release (QoS 2 delivery part 2)
   PUBCOMP      =  7, // Publish complete (QoS 2 delivery part 3)
   SUBSCRIBE    =  8, // Subscribe request
   SUBACK       =  9, // Subscribe acknowledgment
   UNSUBSCRIBE 	=  10, // Unsubscribe request
   UNSUBACK     =  11, // Unsubscribe acknowledgment
   PINGREQ      =  12, // PING request
   PINGRESP     =  13, // PING response
   DISCONNECT  	=  14, // Disconnect notification
   AUTH         =  15, // Authentication exchange
  };

通过运行脚本，您应当能在智能选项卡中看到以下内容。

图例 3 输出测试固定标头 - 测试通过

输出测试固定标头 - 通过

为了确保我们的测试正常工作，我们还要看到它失败的情况。因此，强烈建议您修改 content_buffer 变量表示的输入，同时保持预期变量不变。您应该在智能选项卡输出中看到如下内容。

图例 4 输出测试固定标头 - 测试失败

输出测试固定标头 - 测试失败

无论如何，我们能假设我们的测试在这一点上是脆弱的，如同我们在 mqtt.mqh 头文件中的代码。没问题。我们才刚刚起步，随着我们向前迈进，我们会有机会让它们变得更好，从错误中吸取教训，从而提高我们的技能。

现在，我们可以将 TestFixedHeader_Connect 函数复制到其它数据包类型。从服务器流向客户端时，我们将忽略这些。它们是 CONNACK、PUBACK、SUBACK、UNSUBACK 和 PINGRESP。这些 ACK(S) 和 ping 响应数据包标头将由服务器生成，我们稍后将处理它们。

为了确保我们的测试按预期工作，我们需要包括一些预计失败的测试。这些测试将在失败时返回 true。

#include <MQTT\mqtt.mqh>
//+------------------------------------------------------------------+
//| Script program start function                                    |
//+------------------------------------------------------------------+
void OnStart()
  {
//---
   Print(TestFixedHeader_Connect());
   Print(TestFixedHeader_Connect_RemainingLength1_Fail());
   Print(TestFixedHeader_Publish());
   Print(TestFixedHeader_Publish_RemainingLength1_Fail());
   Print(TestFixedHeader_Puback());
   Print(TestFixedHeader_Puback_RemainingLength1_Fail());
   Print(TestFixedHeader_Pubrec());
   Print(TestFixedHeader_Pubrec_RemainingLength1_Fail());
   Print(TestFixedHeader_Pubrel());
   Print(TestFixedHeader_Pubrel_RemainingLength1_Fail());
   Print(TestFixedHeader_Pubcomp());
   Print(TestFixedHeader_Pubcomp_RemainingLength1_Fail());
   Print(TestFixedHeader_Subscribe());
   Print(TestFixedHeader_Subscribe_RemainingLength1_Fail());
   Print(TestFixedHeader_Puback());
   Print(TestFixedHeader_Puback_RemainingLength1_Fail());
   Print(TestFixedHeader_Unsubscribe());
   Print(TestFixedHeader_Unsubscribe_RemainingLength1_Fail());
   Print(TestFixedHeader_Pingreq());
   Print(TestFixedHeader_Pingreq_RemainingLength1_Fail());
   Print(TestFixedHeader_Disconnect());
   Print(TestFixedHeader_Disconnect_RemainingLength1_Fail());
   Print(TestFixedHeader_Auth());
   Print(TestFixedHeader_Auth_RemainingLength1_Fail());
  }

//+------------------------------------------------------------------+
//|                                                                  |
//+------------------------------------------------------------------+
bool TestFixedHeader_Connect()
  {
   uchar content_buffer[]; //empty
//---
   uchar expected[2];
   expected[0] = 1; //pkt type
   expected[1] = 0; //remaining length
//---
   uchar fixed_header[];
//---
   GenFixedHeader(CONNECT, content_buffer, fixed_header);
//---
   if(!ArrayCompare(expected, fixed_header) == 0)
     {
      Print(__FUNCTION__);
      for(uint i = 0; i < expected.Size(); i++)
        {
         Print("expected: ", expected[i], " result: ", fixed_header[i]);
        }
      return false;
     }
   Print(__FUNCTION__);
   return true;
  }

//+------------------------------------------------------------------+
//|                                                                  |
//+------------------------------------------------------------------+
bool TestFixedHeader_Connect_RemainingLength1_Fail()
  {
   uchar content_buffer[]; //empty
   ArrayResize(content_buffer, 1);
   content_buffer[0] = 1;
//---
   uchar expected[2];
   expected[0] = 1; //pkt type
   expected[1] = 0; //remaining length should be 1
//---
   uchar fixed_header[];
//---
   GenFixedHeader(CONNECT, content_buffer, fixed_header);
//---
   if(!ArrayCompare(expected, fixed_header) == 0)
     {
      Print(__FUNCTION__);
      for(uint i = 0; i < expected.Size(); i++)
        {
         Print("expected: ", expected[i], " result: ", fixed_header[i]);
        }
      return true;
     }
   Print(__FUNCTION__);
   return false;
  }

//+------------------------------------------------------------------+
//|                                                                  |
//+------------------------------------------------------------------+
bool TestFixedHeader_Publish()
  {
   uchar content_buffer[]; //empty
//---
   uchar expected[2];
   expected[0] = 3; //pkt type
   expected[1] = 0; //remaining length
//---
   uchar fixed_header[];
//---
   GenFixedHeader(PUBLISH, content_buffer, fixed_header);
//---
   if(!ArrayCompare(expected, fixed_header) == 0)
     {
      Print(__FUNCTION__);
      for(uint i = 0; i < expected.Size(); i++)
        {
         Print("expected: ", expected[i], " result: ", fixed_header[i]);
        }
      return false;
     }
   Print(__FUNCTION__);
   return true;
  }

//+------------------------------------------------------------------+
//|                                                                  |
//+------------------------------------------------------------------+
bool TestFixedHeader_Publish_RemainingLength1_Fail()
  {
   uchar content_buffer[]; //empty
   ArrayResize(content_buffer, 1);
   content_buffer[0] = 1;
//---
   uchar expected[2];
   expected[0] = 3; //pkt type
   expected[1] = 0; //remaining length should be 1
//---
   uchar fixed_header[];
//---
   GenFixedHeader(PUBLISH, content_buffer, fixed_header);
//---
   if(!ArrayCompare(expected, fixed_header) == 0)
     {
      Print(__FUNCTION__);
      for(uint i = 0; i < expected.Size(); i++)
        {
         Print("expected: ", expected[i], " result: ", fixed_header[i]);
        }
      return true;
     }
   Print(__FUNCTION__);
   return false;
  }
.
.
.
(omitted for brevity)

嘿！这是大量的范本代码，数十次键入，以及复制/粘贴！

是的，这是肯定的。但从长远来看，这将有很好的回报。通过这些简单、甚至简陋的现场测试，我们正在为我们的开发建立一种安全网。它们应当帮助我们

专注于手头的任务，
避免过度工程，
以及发现递归漏洞。

注意：我强烈鼓励您自己编写它们，替代简单地使用附件。您从一开始就会看到捕捉到多少微小的、“无攻击性”的错误。随着我们对客户端操作行为的推进，这些测试（以及其它更具体的测试）将证明它们的价值。除此之外，我们还避免了常见的技术欠债：将测试留待最后编写。通常，如果您把测试留到最后，就永远不想再写了。

图例 5 输出测试固定标头 - 全部通过

输出测试固定标头 - 全部通过

好的，我们来看看两字节 CONNECT 标头是否被 MQTT 代理识别为有效的标头。

如何安装开发和测试的 MQTT 代理（及客户端）

网上有许多 MQTT 代理产品，其中大多数都提供某种“沙箱” URL，用于开发和测试目的。在您最喜欢的搜索引擎上简单地搜索 “MQTT broker” 就足以帮助您找到其中的一些。

然而，此刻我们的客户端才刚萌芽。若无数据包分析器来捕获网络流量，我们还不能接收和读取响应。这个工具以后会很实用，但就目前而言，在我们的开发机器上安装一个符合规范的 MQTT 代理就足够了，如此我们就可以检查它的日志来查看我们的交互结果。理想情况下，它应该安装在虚拟机上，以便得到一个我们客户端以外的 IP。用不同 IP 的代理进行开发和测试，我们能够更提早解决连接和身份验证问题。

再次，Windows、Linux 和 Mac 也有若干选项。我已经在 Windows 的 Linux 子系统（WSL）上安装了 Mosquitto。除了免费和开源之外，Mosquitto 还非常便捷，因为它带有两个非常适用于开发的命令行应用程序：mosquitto_pub 和 mosquitto_sub 来发布和订阅 MQTT 主题。我还把它安装在 Windows 开发机器上，如此我就可以交叉检查一些错误。

请记住，MetaTrader 要求您在工具>选项菜单的智能系统选项卡中列出任何外部 URL，并且 MetaTrader 只允许您访问端口 80 或 443。因此，如果您遵照在 WSL 上安装代理的路径，不要忘记包含其主机 IP，也不要忘记将到达端口 80 的网络流量重定向到 1883，这是默认的 MQTT（和 Mosquitto）端口。有一个名为 redir 的工具，它以简单稳定的方式执行端口重定向。

图例.6 MetaTrader 5 对话框 - 允许 Web 请求 URL

MetaTrader 5 对话框 - 允许 Web 请求 URL

若要获取 WSL IP，运行以下命令。

图例 7 WSL 获取主机名命令

WSL 获取主机名命令

一旦安装完毕，Mosquitto 将自行配置为在电脑启动时作为“服务”开启。因此，只需重新启动 WSL 即可在默认端口 1883 上开启 Mosquitto。

为了用 redir 将网络流量从端口 80 重定向到 1883，请运行以下命令。

图例 8 “redir” 重定向网络流量

在命令行用 “redir” 端口重定向

最后，我们可以检查我们的两个字节 CONNECT 固定标头是否被符合规范的 MQTT 代理识别为有效的 MQTT 标头。只需创建一个“临时”脚本，并粘贴以下代码即可。（不要忘记根据 get hostname -I 命令的输出更改 broker_ip 变量中的 IP 地址。

#include <MQTT\mqtt.mqh>

string broker_ip = "172.20.155.236";
int broker_port = 80;
//+------------------------------------------------------------------+
//| Script program start function                                    |
//+------------------------------------------------------------------+
void OnStart()
  {
   int socket = SocketCreate();
   if(socket != INVALID_HANDLE)
     {
      if(SocketConnect(socket, broker_ip, broker_port, 1000))
        {
         Print("Connected ", broker_ip);
         //---
         uchar fixed_header[];
         uchar content_buffer[]; //empty
         //---
         GenFixedHeader(CONNECT, content_buffer, fixed_header);
         //---
         if(SocketSend(socket, fixed_header, ArraySize(fixed_header)) < 0)
           {
            Print("Failed sending fixed header ", GetLastError());
           }
        }
     }
  }

您应当在“智能”选项卡输出中看到以下内容...

图例 9 输出本地代理连接

输出本地代理连接

...以及 Mosquitto 日志中的以下输出。

图例 10 输出本地代理连接 - Mosquitto 日志

输出本地代理连接 - Mosquitto 日志

所以，是的，我们的 CONNECT 固定标头已由 Mosquito 识别，但<未知>客户端“由于协议错误”立即断开连接。发生此错误的原因是，我们尚未包含可变标头、协议名称、协议级别、和其它相关的元数据。我们将在下一步中修复该问题。

注意：正如您在上述命令的一开头所看到的，我们用的是 tail -f {pathToLogFile} 命令。我们可以在开发过程中用它来跟踪 Mosquito 日志更新，而无需打开和重新加载文件。

在下一步中，我们将实现 CONNECT 可变标头 - 和其它标头 - 以便维护与代理的稳定连接。我们还将发布一条消息，并处理代理返回的 CONNACK 数据包，及其相关原因代码。下一步将有一些有趣的按位运算，来填充我们的连接标志。下一步还要求我们大幅改进我们的测试，从而应对因客户端-代理对话而出现的复杂性。

结束语

在本文中，我们查看了一份 MQTT 发布/订阅实时消息共享协议、其起源和主要组件。我们还指出了 MQTT 在交易环境中实时消息传递的一些可能用例，以及如何通过导入从 C、C++ 或 C# 生成的 DLL，或通过 MetaTrader 5 Python 模块调用 MQTT Python 函数库，将其用于 MetaTrader 5 中的自动化操作。

考虑到在 MetaQuotes 市场和 MetaQuotes 云端测试上使用 DLL 的限制，我们还提出并讲述了使用测试驱动开发（TDD）方式实现原生 MQL5 MQTT 客户端的第一步。

一些可能有用的参考资料

我们不需要重新发明所有的轮子。开发人员在为其它语言编写 MQTT 客户端时面临的最常见挑战，大多都以开源库/SDK 的形式提供了解决方案。