简介面向使用Borland C Builder 6.0BCB6开发环境的开发者这份MQTT通信案例演示了在经典C Builder项目中集成Eclipse Paho MQTT C库实现物联网设备消息收发和主题订阅的完整路径。资源包共包含15个文件整体体积仅225KB其中既有编译好的可执行程序和paho-mqtt3a动态库也提供头文件、静态库、BCB6工程文件、主程序Unit1.cpp源码及cf.txt配置说明按项目结构归置可直接在BCB6中打开工程方便查看调用关系或进行二次修改。通过案例学习可清晰掌握创建MQTT客户端、指定服务器地址与端口、连接及鉴权、发布消息、订阅主题、注册回调函数处理到达消息、断开连接并释放资源等关键环节对理解Paho C库的API使用和传统Windows桌面程序中的MQTT接入方式很有帮助。目前已有459人次学习下载适合需要在BCB6或相近版本C Builder中快速实现物联网通信的开发者和嵌入式、上位机等应用场景。 很多朋友一听BCB6这三个字第一反应都是这年头还有人用这个。说实话我也理解Borland C Builder 6是2002年的东西了连官方支持都停了不知道多少年。但现实是工业上位机、老设备配套软件、工厂数据采集程序这一堆老古董里跑着的还是BCB6写的exe。这几年物联网改造、设备上云的需求越来越多这些老程序也要接MQTT协议往外送数据。于是问题就来了在BCB6这个快二十多岁的老编译器环境里怎么搞一个能用的MQTT客户端我把整个研究和实测的过程整理出来代码和坑都在下面给还在维护老项目的兄弟们一个参考。1. 老项目接MQTT的现状不是能不能而是怎么最省事先说结论BCB6接MQTT完全能做难点不在协议本身而在于环境太老。MQTT协议本身非常轻量基于TCP/IP报文结构简单理论上只要会写Socket就能实现。但麻烦就麻烦在BCB6的库和C标准停留在C98之前的水准很多现代库根本编译不过去哪怕编译过去了跑起来也指不定哪里出问题。我在开始之前也习惯性地搜了一下别人有没有现成方案搜了一圈发现案例确实少。更多的人是在用C#、Python、Node-RED或者ESP32之类的新玩具做MQTTBCB6相关的资料寥寥无几。这也正常用BCB6写新项目的概率本来就很低大多是老程序改造。既然没有太多现成案例那就自己动手。先说几个踩过之后才明白的点大家心里有个数MQTT协议本身不复杂一个最小化的客户端核心逻辑几百行就能搞定。BCB6里没有现成的MQTT库第三方库要么太新编译不过要么依赖一堆底层库引入成本比手写还高。BCB6自带的Indy组件TIdTCPClient等虽然能连TCP但直接用Indy做MQTT会遇到一些奇奇怪怪的缓冲和超时问题不如裸Socket可控。综合判断下来最务实的路子就是用WinSock手写一个轻量MQTT客户端只在旧编译器里编译不引入外部依赖。这样代码完全可控出问题能自己查部署到客户机器上也省掉一堆DLL依赖的麻烦。1.1 现有方案对比为什么不是Indy也不是Paho我把自己实际对比过的几条路列一下方便大家参考方案优点缺点结论BCB6自带Indy组件不需要额外引入库UI集成方便老版本会有缓冲延迟、超时处理不够灵活而且Indy 9的文档早已失传不推荐排查问题太费劲Eclipse Paho C库功能完整、协议兼容性好编译环境要求较高BCB6编译需要大量改造容易遇到一堆兼容性报错可以硬试但心累手写WinSock实现完全可控代码量小无外部依赖需要自己处理协议细节QoS支持有限我最终的选择我这个手写并不是说要把完整MQTT协议从头实现一遍而是做一个够用的子集。对绝大多数数据上报和设备控制场景来说只需要实现CONNECT、PUBLISH、SUBSCRIBE、PINGREQ/PINGRESP这几个报文就足够了。QoS 0够用的前提下代码量真的不大。2. 动手前必须搞懂的MQTT报文结构如果说前面是规划那这节就是地基。手写MQTT客户端最核心的事情就是理解报文格式。MQTT的控制报文分为三部分固定报头Fixed Header、可变报头Variable Header、载荷Payload。固定报头每个报文都有可变报头和载荷看具体报文类型。固定报头由两个部分组成第一个字节是控制报文类型和标志位后面是剩余长度Remaining Length。剩余长度用变长编码表示这个编码规则很容易被忽略但恰恰是很多手写客户端出bug的高发地。剩余长度的编码规则是这样每个字节的低7位存放数据最高位作为连续标志。如果最高位为1说明后面还有字节继续表示长度。每个字节的有效负载是7位所以最多用4个字节。我第一次写的时候就忘记处理超过127字节的情况结果一发送超过127字节的payloadbroker直接断开连接报协议错误。这个细节后面讲代码的时候会重点说。再展开一下不同报文的可变头结构。CONNECT报文的可变头包含协议名固定是四个字符MQTT、协议级别v3.1.1是0x04、连接标志用户名密码标志、遗嘱标志、Clean Session标志等、KeepAlive时间单位秒2字节。PUBLISH报文的可变头包含主题名2字节长度 主题字符串和报文标识符仅QoS大于0时有。SUBSCRIBE报文包含报文标识符2字节载荷里是主题过滤器加QoS级别的组合。这些字段都是大端序网络字节序BCB6在Windows上跑的是小端所以组装报文的时候要注意高低字节转换。用Word类型直接往字节数组里填会导致顺序颠倒必须手动位移。2.1 CONNECT报文组装详解CONNECT报文是整个连接流程的第一步。客户端建立TCP连接之后要立刻发送CONNECTbroker才会回应CONNACK。如果5秒内没收到CONNACK连接就算失败。看一个最小化的CONNECT报文结构协议级0x04无遗嘱、无用户名密码固定头: 0x10 剩余长度 可变头: 00 04 4D 51 54 54 -- MQTT协议名 04 -- 协议级别4 02 -- 连接标志Clean Session1 00 3C -- KeepAlive 60秒 载荷: 00 05 63 6C 69 65 6E 74 -- Client ID client连接标志那一位二进制是0000 0010对应Clean Session。如果要用户名密码这一位要加上1000 0000用户名和0100 0000密码也就是0xC2。我在实际测试中碰到一个问题因为公共测试broker要求必须设置用户名密码而本地EMQX默认允许匿名两种场景的报文组装还不一样所以代码里要把用户名密码做成可选参数。KeepAlive这个是心跳间隔建议设60秒。设置太短会频繁发心跳包占带宽太长的话NAT超时或者网络故障时broker发现断线的时间也会变长。对工控场景设备数量多、网络不稳的环境我一般设30秒正常项目60秒够用。2.2 变长剩余长度编码最容易翻车的地方这个值得单独拿出来说。之前提过剩余长度超过127以后用单个字节就表示不了因为每个字节只有低7位有效。举一个实际例子发送一个主题test/topic、载荷100字节的PUBLISH报文剩余长度是2主题长度字段 10主题字符串长度 100载荷长度 112一个字节就能编码。但如果载荷变成200字节剩余长度变成212二进制是11010100需要两个字节编码第一字节: 11010100 0x7F 01010100 (0x54) -- 低7位最高位补1 第二字节: 11010100 7 00000001 (0x01) -- 剩余的高位最高位为0所以在代码里必须用一个循环来编码凡是写上剩余长度 payload.Length() 2然后直接往缓冲区塞的操作都是错的。尤其是那些用TBytes或者PChar手工拼包的老C代码非常容易在这里翻车。我第N次写这个逻辑的时候才意识到这个东西就跟你骑自行车一样一旦搞明白这辈子忘不了。3. 手写一个够用的MQTT客户端核心代码拆解前面铺垫了这么多终于到正题了。下面是我自己改造BCB6程序时用的一个最小客户端实现不说完整代码全贴太长了把最关键的部分拿出来逐段拆解。我用的是一个简单的类封装放在单独一个单元里UI层只调三个方法Connect、Publish、ProcessIncoming。这样做的好处是UI层代码干净出问题先查这个类。3.1 类的整体设计和连接建立//--------------------------------------------------------------------------- #ifndef MQTTClientH #define MQTTClientH #include winsock2.h class TMQTTClient { private: SOCKET FSocket; AnsiString FHost; int FPort; AnsiString FClientId; bool FConnected; WORD FPacketId; BYTE EncodeRemainingLength(int length, BYTE* buffer); bool SendRaw(BYTE* data, int len); bool ReadByte(BYTE* value); bool ReadBytes(BYTE* buffer, int len); int ReadRemainingLength(); public: TMQTTClient(AnsiString host, int port, AnsiString clientId); ~TMQTTClient(); bool Connect(AnsiString username , AnsiString password ); void Disconnect(); bool Publish(AnsiString topic, AnsiString payload, bool retain false); bool Subscribe(AnsiString topic, BYTE qos 0); void ProcessIncoming(int timeoutMs); bool IsConnected() { return FConnected; } }; #endif //---------------------------------------------------------------------------连接函数的核心逻辑是先建TCP连接再发CONNECT包最后等CONNACK。//--------------------------------------------------------------------------- bool TMQTTClient::Connect(AnsiString username, AnsiString password) { WSADATA wsaData; WSAStartup(MAKEWORD(2,2), wsaData); FSocket socket(AF_INET, SOCK_STREAM, 0); if (FSocket INVALID_SOCKET) return false; hostent* hostEntry gethostbyname(FHost.c_str()); if (!hostEntry) { closesocket(FSocket); return false; } sockaddr_in addr; addr.sin_family AF_INET; addr.sin_port htons(FPort); memcpy(addr.sin_addr, hostEntry-h_addr, hostEntry-h_length); if (connect(FSocket, (sockaddr*)addr, sizeof(addr)) ! 0) { closesocket(FSocket); return false; } // 组装CONNECT报文 BYTE payload[256]; int pos 0; payload[pos] 0x10; // CONNECT固定头 AnsiString protocol MQTT; int protoLen protocol.Length(); int clientIdLen FClientId.Length(); int usernameLen username.Length(); int passwordLen password.Length(); int variableLen 2 protoLen 1 1 2 2 clientIdLen; BYTE connectFlags 0x02; // Clean Session if (usernameLen 0) { connectFlags | 0x80; variableLen 2 usernameLen; } if (passwordLen 0) { connectFlags | 0x40; variableLen 2 passwordLen; } BYTE lenBuf[4]; int lenCount EncodeRemainingLength(variableLen, lenBuf); payload[pos] lenBuf[0]; for (int i 1; i lenCount; i) payload[pos] lenBuf[i]; // 可变头 payload[pos] 0; payload[pos] protoLen; memcpy(payload pos, protocol.c_str(), protoLen); pos protoLen; payload[pos] 0x04; // 协议级别 payload[pos] connectFlags; payload[pos] 0x00; // KeepAlive高字节 payload[pos] 0x3C; // KeepAlive低字节 60秒 // Client ID payload[pos] clientIdLen 8; payload[pos] clientIdLen 0xFF; memcpy(payload pos, FClientId.c_str(), clientIdLen); pos clientIdLen; // 用户名 if (usernameLen 0) { payload[pos] usernameLen 8; payload[pos] usernameLen 0xFF; memcpy(payload pos, username.c_str(), usernameLen); pos usernameLen; } // 密码 if (passwordLen 0) { payload[pos] passwordLen 8; payload[pos] passwordLen 0xFF; memcpy(payload pos, password.c_str(), passwordLen); pos passwordLen; } if (!SendRaw(payload, pos)) return false; // 等待CONNACK BYTE header; if (!ReadByte(header)) return false; if ((header 4) ! 2) return false; // 0x20 CONNACK int remainLen ReadRemainingLength(); BYTE ack[4]; if (!ReadBytes(ack, remainLen)) return false; if (ack[1] ! 0) return false; // 返回码非0则失败 FConnected true; return true; } //---------------------------------------------------------------------------注意这段代码里用了固定数组payload[256]实际使用时Client ID和用户名密码可能会超长建议改成动态分配。我这里是示意真实项目我会用vector 或者new一个缓冲区。**为什么用gethostbyname而不用inet_addr**因为broker地址在工控现场往往是域名而不是IP比如公司内网MQTT服务器的机器名gethostbyname能做域名解析inet_addr只能处理点分十进制IP。当然gethostbyname已经废弃了但在BCB6环境下没有更合适的选择能用就行。3.2 PUBLISH报文组装和发送连接建立之后最常用的就是发消息。PUBLISH报文的固定头第一个字节是0x30QoS0如果retain标记加0x01如果是QoS1是0x32QoS2是0x34。我的实现只处理QoS0因为做数据上报用QoS0最简单每次发完就算完事不需要等待broker回PUBACK。如果你要QoS1就得维护一个报文标识符并处理重发逻辑代码复杂度会上升不少。//--------------------------------------------------------------------------- bool TMQTTClient::Publish(AnsiString topic, AnsiString payload, bool retain) { if (!FConnected) return false; int topicLen topic.Length(); int payloadLen payload.Length(); int remainingLen 2 topicLen payloadLen; BYTE fixedHeader[5]; fixedHeader[0] 0x30 | (retain ? 0x01 : 0x00); int headerLen 1; BYTE lenBuf[4]; int lenCount EncodeRemainingLength(remainingLen, lenBuf); for (int i 0; i lenCount; i) fixedHeader[headerLen] lenBuf[i]; BYTE packet[1024]; int pos 0; memcpy(packet, fixedHeader, headerLen); pos headerLen; packet[pos] topicLen 8; packet[pos] topicLen 0xFF; memcpy(packet pos, topic.c_str(), topicLen); pos topicLen; memcpy(packet pos, payload.c_str(), payloadLen); pos payloadLen; return SendRaw(packet, pos); } //---------------------------------------------------------------------------这里的packet[1024]也是示意如果一条消息可能超过1KB务必改成动态分配或者用链式发送。工控场景里最大消息体一般超不过几百字节但万一传个图片或者配置文件就有风险了。我在真实项目里用了一个简单的动态缓冲区发送前计算总长度然后分配。3.3 SUBSCRIBE订阅报文和收消息订阅报文相对麻烦一点因为SUBSCRIBE需要携带报文标识符而且broker会回SUBACK。报文标识符从1开始递增不用处理回绕的情况只要不是一次发几百条消息Word类型足够。//--------------------------------------------------------------------------- bool TMQTTClient::Subscribe(AnsiString topic, BYTE qos) { if (!FConnected) return false; int topicLen topic.Length(); int remainingLen 2 2 topicLen 1; BYTE fixedHeader[5]; fixedHeader[0] 0x82; // SUBSCRIBE固定头 int headerLen 1; BYTE lenBuf[4]; int lenCount EncodeRemainingLength(remainingLen, lenBuf); for (int i 0; i lenCount; i) fixedHeader[headerLen] lenBuf[i]; FPacketId; BYTE packet[256]; int pos 0; memcpy(packet, fixedHeader, headerLen); pos headerLen; packet[pos] FPacketId 8; packet[pos] FPacketId 0xFF; packet[pos] topicLen 8; packet[pos] topicLen 0xFF; memcpy(packet pos, topic.c_str(), topicLen); pos topicLen; packet[pos] qos; return SendRaw(packet, pos); } //---------------------------------------------------------------------------接收消息这部分是整个客户端里最关键的。因为MQTT是长连接broker随时可能往客户端推送消息所以不能像HTTP那样发一个请求收一个响应。我在BCB6的UI层用一个TTimer定时调ProcessIncoming每10毫秒去Socket缓冲区看看有没有数据。这个思路本质上是把阻塞接收变成非阻塞轮询避免卡死VCL主界面。//--------------------------------------------------------------------------- void TMQTTClient::ProcessIncoming(int timeoutMs) { if (!FConnected || FSocket INVALID_SOCKET) return; fd_set fds; FD_ZERO(fds); FD_SET(FSocket, fds); timeval tv; tv.tv_sec 0; tv.tv_usec timeoutMs * 1000; int ret select(0, fds, NULL, NULL, tv); if (ret 0) return; // 没有数据或出错 BYTE header; if (!ReadByte(header)) { FConnected false; return; } int remainingLength ReadRemainingLength(); BYTE* buffer new BYTE[remainingLength]; if (!ReadBytes(buffer, remainingLength)) { delete[] buffer; FConnected false; return; } int msgType (header 4) 0x0F; switch (msgType) { case 3: // PUBLISH { int topicLen (buffer[0] 8) | buffer[1]; AnsiString topic(buffer 2, topicLen); AnsiString payload(buffer 2 topicLen, remainingLength - 2 - topicLen); OnMessage(topic, payload); // 回调在UI层挂接处理 break; } case 13: // PINGRESP // 心跳响应不需要做任何事 break; case 12: // PINGREQ { BYTE pingResp[2] { 0xD0, 0x00 }; SendRaw(pingResp, 2); break; } case 8: // SUBACK // 订阅确认记日志用 break; case 4: // PUBACKQoS1时用 break; } delete[] buffer; } //---------------------------------------------------------------------------这里有一个很重要的设计决定把OnMessage做成什么形式我选择给类加一个函数指针成员或者简单在UI层直接改这个类的源码把OnMessage里做的事写死。最省事。因为BCB6老项目的UI层常有一堆全局变量用回调函数指针反而不方便直接在类里发一个VCL事件或者调用一个全局函数更顺手。实际项目中我是这样做的在OnMessage里把topic和payload塞进一个线程安全的队列UI层单独处理。这样即便未来哪天消息量大也可以扛住。如果消息量很少直接在OnMessage里Update界面也行但记住不要在Socket线程里碰VCL控件。3.4 心跳和断线检测MQTT协议里KeepAlive机制是客户端定期发PINGREQbroker回PINGRESP。如果broker在1.5倍KeepAlive时间内没收到任何报文就会认为客户端掉线并清理会话。客户端这边也要做检测如果超过一定时间没收到任何数据主动断开重连。我维护一个计数器每次ProcessIncoming调用时检查距离上次收发数据的时间超过KeepAlive时间就发一次PINGREQvoid TMQTTClient::SendPing() { BYTE ping[2] { 0xC0, 0x00 }; // PINGREQ SendRaw(ping, 2); }这个心跳在TTimer的OnTimer事件里调10毫秒轮询一次数据每30秒发一次心跳简单可靠。4. 老编译器里的暗坑BCB6这个环境自带的问题代码本身写明白了接下来才是真正折磨人的部分。BCB6作为一个2002年的开发环境在配合MQTT这种需要精确处理字节流的协议时有四个坑是我觉得最值得说的。4.1 坑一AnsiString的字节长度和字符长度混为一谈BCB6的AnsiString是单字节编码一个字符就是一个字节中文在GBK编码下是两个字节。这在大多数场景下没毛病但有个隐蔽的坑当你从MQTT报文的字节流里截取topic和payload时AnsiString(buffer 2, topicLen)这种构造方法在BCB6里是可以用的但如果payload本身包含\0字符直接用AnsiString(payloadStr)会把字符串截断。因为默认构造函数遇到\0就停了。解决办法是**任何时候都不要用C字符串函数如strlen、strcpy去处理MQTT的payload必须用带长度的构造和拷贝。**老程序里一个不小心把strcpy用在阿里的JSON数据上就会丢数据而且很难排查。4.2 坑二Socket缓冲区与Winsock初始化BCB6里如果用TClientSocket或者IndyWSAStartup会被封装掉。但我们手撸WinSock就得自己调WSAStartup。更坑的是有些老项目里已经有别的代码调过wsock32.dll的函数但未必传了版本号。如果在BCB6工程里同时用了第三方通信组件可能会发生WSAStartup版本协商不一致的问题。我的做法是在TMainForm::OnCreate事件里最早的位置调用一个InitSocket()函数不管之前有没有初始化过强制MAKEWORD(2,2)再来一次WSAStartup可以被多次调用每次要对应WSACleanup。这样最稳。4.3 坑三read操作不一定能一次读完这是写Socket最容易犯的错跟语言无关。TCP是流协议recv返回的数据长度不一定等于你期望的长度。比如你发了一个1024字节的报文对方可能在物理层就拆成了两段第一段到了600字节第二段424字节后到。如果代码里简单写recv(sock, buf, 1024, 0)可能只能收到600字节然后就把后面424字节当成下一个报文的开头来处理了直接导致协议解析错乱。我代码里专门写了ReadBytes函数循环调用recv直到收满指定字节数。这个函数是整个客户端稳定性的基础不要为了省事跳过。bool TMQTTClient::ReadBytes(BYTE* buffer, int len) { int received 0; while (received len) { int ret recv(FSocket, (char*)buffer received, len - received, 0); if (ret 0) return false; received ret; } return true; }4.4 坑四Timer轮询在界面卡顿时的行为BCB6的TTimer靠VCL消息循环驱动。如果你在主线程里跑了一个耗时的同步操作比如读取串口等待超时TTimer就停了ProcessIncoming也不会被调用Socket的数据就会积压在缓冲区里。等界面恢复Timer继续跑一次把所有积压的数据读出来——幸好我们用的是select超时非阻塞读取所以不会卡死但处理积压数据时如果一次recv缓冲区不够大也可能出问题。我的建议是**凡是可能阻塞超过50毫秒的操作一律放到单独线程里主线程只做界面刷新。**如果老项目线程化改造工作量太大至少要保证ProcessIncoming里每次最多读一个报文就返回不要用while循环把缓冲区读干避免一个Timer周期里处理太多消息导致界面掉帧。5. 真实验证连上EMQX服务器跑通收发写完了代码最后来看一次完整的连调和验证过程让照着做的朋友心里有底。5.1 测试环境准备我本机的环境是Windows 10 BCB6被打过补丁的版本测试broker用的是EMQX。为了省事也可以用公共broker比如broker.emqx.io但公共broker响应慢且有频率限制调试阶段建议本地起一个。EMQX在Windows上有安装包装完默认端口1883匿名访问默认打开测试最方便。如果你手头有Mosquitto也行命令行的mosquitto_pub/mosquitto_sub可以做对端验证但这个在Windows上配起来没EMQX图形界面直观。5.2 实测过程从连接成功到数据订阅第一步我用EMQX Dashboard建了一个测试topic叫bcb6/test/data。第二步BCB6程序里调用Connect(, )也就是匿名连接然后Subscribe一个bcb6/test/data。第三步在EMQX Dashboard的WebSocket客户端里往这个topic发一条JSON数据{temp:23.5,hum:62}。第四步看BCB6程序界面上是否显示这条消息。整个过程里连接建立后我在界面上显示MQTT已连接收到消息后弹出一个Label显示最新数据的温度和湿度。跑了半小时没有断线心跳正常。5.3 一个最容易遇到的现场问题连不上本地broker测试中我发现一个容易踩的坑是防火墙。Windows防火墙默认会拦掉外部连接如果你在调试机的firewall里没放行1883端口BCB6程序会一直卡在connect超时。排查方法很简单先关掉防火墙测试如果通了再一条一条地添加允许规则。这个坑跟代码本身无关但实际部署到客户电脑上十有八九会遇到。5.4 一个完整的现场案例老式串口设备数据上云我改造过一个现场项目一台BCB6写的上位机原本从一台温控仪通过RS485读取温度数据显示在界面上。客户的新需求是要把这台设备的实时数据同步到一个工业物联网平台。方案就是在这台老的BCB6上位机里加了MQTT功能串口读到的温度每次变化时调用Publish往factory/line1/device1/temp这个topic发一条JSON。物联网平台订阅这个topic就能拿到数据。数据格式{device:device1,temp:23.5,ts:1690123456}上报频率温度变化超过0.1℃才发正常每5秒一次避免无效数据占用带宽QoS选择QoS0丢了就丢了因为本来就是周期上报下一条会带最新值上线之后跑了三周没出过一次断线重连的问题整个过程算是顶着BCB6的老胳膊老腿把活干完了。最后再分享一点个人的实操心法这套东西做完以后我最大的感触就是在BCB6这个老环境里少依赖花哨的第三方库、多从协议底层入手反而是解决问题的最短路径。MQTT这种设计得干净利落的协议手写一遍之后你对整个通信过程的理解深度是直接调库完全比不了的。如果要做功能扩展我建议下一步可以优先补上QoS1的PUBACK处理以及遗嘱消息Last Will的设置。遗嘱消息在有设备掉线监测需求的项目里很实用实现也不难就是在CONNECT报文里把遗嘱标志和遗嘱主题加进去——顺便说一句遗嘱主题千万别跟业务topic混在一起否则断线报文会把下游数据处理逻辑搅乱。就这些有问题评论区见能帮上忙的我尽量答。本文还有配套的精品资源点击获取