基于libmosquitto封装C语言MQTT客户端：断线重连与心跳保活实战

手上这个基于 mosquito 封装的 mqtt 客户端，最早是被一个设备接入项目逼出来的。当时现场有几百台小设备要往平台上报状态，还得接收平台下发的控制指令，最初图省事直接在业务代码里裸调mosquitto_publish和mosquitto_subscribe，结果连接一断整个模块就装死，线程还偶尔卡在回调里出不来。后来痛下决心抽了一层薄封装：把连接管理、心跳保活、断线重连、订阅恢复、消息分发这些事全收进一个独立模块，业务侧只留几个接口调用。

先说明一点，这个领域里正确的拼写是Eclipse Mosquitto，很多人（包括我）习惯打成 mosquito，圈内交流时基本默认是同一个东西。它是目前用得最广的 MQTT broker 之一，同时它自带的C 客户端库 libmosquitto才是我们这次的主角。所以标题里的"基于 mosquito 封装的 mqtt 客户端"，准确讲是基于 libmosquitto 那套 C API 做一层自己的客户端封装。

这篇文章适合两类人看：一类是正在用 C/C++ 写嵌入式或网关程序、需要接 MQTT 的开发者；另一类是已经用上了 Qt MQTT、Vue3 mqtt、Android mqtt 这些高层库，但想搞清楚底层到底发生了什么的人。我会把设计思路、参数取舍、完整代码、编译验证、踩坑记录全部摊开讲，你可以直接照着抄，也可以只看其中某一节。

1. 从标题到方案：为什么要封装一层 MQTT 客户端

1.1 原生 libmosquitto 用起来到底卡在哪

libmosquitto 的 API 设计其实挺克制，核心就那么二十来个函数，但它是一个事件驱动的回调式 C 库，这个特性决定了它不适合被业务代码直接使用。第一个问题是回调的上下文很敏感：on_message回调是在库的网络线程里执行的，你在里面做一次磁盘写入或者发一个同步 HTTP 请求，整个客户端的收发就被卡住了，broker 那边超过 1.5 倍 keepalive 没收到心跳包，直接把你踢下线。

第二个问题是状态管理散落。连接成功要重订阅、断开要触发业务侧的降级逻辑、重连成功要补发在线状态，这些动作如果每个业务模块各写一遍，代码里就会出现五六个版本的"重连后该干什么"，时间一长根本没人敢改。第三个问题是线程安全边界模糊。很多人不知道mosquitto_publish是可以从任意线程调用的，但mosquitto_loop_stop不行；也不知道回调收到的payload指针在回调返回后就失效了，必须自己拷贝一份。这类细节不封装起来，全靠口头约定，迟早出事。

第四个问题最隐蔽：内存所有权混乱。mosquitto_message里的topic和payload都是库内部 malloc 出来的，如果你在回调里把指针存下来异步使用，等于握着野指针；如果你自己 new 了一份忘了释放，跑一周就是内存泄漏。这些坑不踩一遍根本记不住，封装层的价值就是让它们只踩一遍。

1.2 封装层该切在哪一刀：接口边界与职责划分

我的做法是把封装层切成三块职责，边界画得很清楚。第一块是生命周期管理：创建、启动、停止、销毁，对外只暴露init/fini两个函数，内部的mosquitto_new、mosquitto_loop_start、mosquitto_destroy、mosquitto_lib_cleanup全部藏起来。这样调用方永远不会忘记调mosquitto_lib_cleanup，也不会出现两个模块各自调一次导致引用计数错乱。

第二块是连接与订阅的状态机：维护一个"期望订阅列表"，连上就全部重订阅，断开就标记状态。这个列表是封装层自己持有的，业务侧只需要在初始化时声明一次"我要订这些主题"，之后无论断线多少次都不用管。第三块是消息投递：回调里只做一件事——把 payload 拷进一块由上层管理的内存，然后通过注册的函数指针交给业务。业务侧拿到的永远是独立的内存，可以随便异步处理。

这么切的好处是：封装层的代码量大概四百行左右，但没有一行是业务逻辑。它只依赖libmosquitto和标准库，可以扔进任何 C/C++ 项目里复用。缺点是引入了函数指针跳转，性能上有个微秒级的开销——不过在 MQTT 这种每秒几百到几万条的场景里，这点开销完全可以忽略。

注意：如果你用的是 C++，别急着用类去包装它。C 回调 +void *user私有指针的模式在跨语言绑定、FFI 调用时更省事，C++ 里套一层static成员函数做转发就行，没必要上虚函数。

2. 核心机制拆解：连接、心跳、重连与 QoS

2.1 连接建立与认证参数的取舍

mosquitto_new的第一个参数是client_id，这个字符串在 broker 侧是用来做会话标识的，必须全局唯一。我见过最蠢的做法是用argv[0]或者getpid()当 client id，进程重启后 id 变了，broker 那边上一份会话还挂着，一下子多出几百个幽灵连接。正确的做法是用设备的物理标识：MAC 地址、序列号、IMEI，或者业务系统分配的设备编号。如果实在没有唯一标识，用主机名 + 进程号拼一个也行，但一定要保证同一台设备重启后 id 不变。

第二个参数clean_session是个分水岭。传true时，broker 不保存任何会话状态，你断线期间别人发的消息全部丢弃；传false时，broker 会替你保留订阅关系和 QoS 1/2 的离线消息，等你重连回来后推给你。这里有个反直觉的点：clean_session = false 时，client id 必须非空，因为 broker 要靠它认人。而且会话不能无限期挂着，Mosquitto 默认persistent_client_expiration是不过期的，实际部署时建议配上过期时间，否则设备报废后会话会一直占着内存。

认证这块支持三种：匿名、用户名密码、TLS 双向证书。mosquitto_username_pw_set传的是明文密码，走的是 MQTT CONNECT 报文里的 user name / password 字段。如果链路本身不加密，等于把密码贴在额头上，所以只要跨公网，就必须配 TLS。mosquitto_tls_set的五个参数分别是 CA 证书路径、CA 目录、客户端证书、客户端私钥、密码回调。只用服务端单向认证时后三个传NULL即可；要做双向认证，客户端证书和私钥都得给。

提示：私钥文件权限记得设成 600，并且不要把它打进容器镜像的公共层。我见过一个团队把私钥提交到代码仓库，半夜被人扫出来，第二天所有设备证书集体作废重签。

2.2 keepalive 心跳与断线判定的真实逻辑

keepalive这个参数容易被误解成"心跳间隔"，其实它的语义是客户端承诺的最大静默时间。MQTT 协议规定：如果 broker 在 1.5 倍 keepalive 时间内没有收到该客户端的任何报文（包括 PUBLISH、SUBSCRIBE 这些正常业务报文），就认为连接已失效并断开它。反过来，客户端如果在 keepalive 时间内没有任何报文要发，就必须主动发一个 PINGREQ。

理解了这个机制，参数就好定了。设成 60 秒意味着最坏情况下断线要 90 秒才被发现；设成 10 秒则断线 15 秒就能察觉，代价是心跳包变多。一个 PINGREQ/PINGRESP 往返是 4 字节左右的报文体，加上 TCP/IP 头也就七十来字节。按 keepalive = 30 秒算，单连接的额外流量是 70 × 2 / 30 ≈ 4.7 字节/秒，一天不到 400KB。这个开销对现在的网络环境完全可以接受，所以我一般用 30 秒作为默认值，移动网络下放宽到 60 秒。

但这里有个很坑的地方：libmosquitto 自己是不发包的，心跳包由网络循环里的mosquitto_loop根据时间戳决定何时发出。如果你用了mosquitto_loop_forever或者mosquitto_loop_start，这件事它替你做了。但如果你是自己在某个大循环里手动调mosquitto_loop(mosq, timeout, 1)，而那个循环被别的耗时操作拖住了，心跳就会延迟，broker 就会把你踢掉。这是我当年排查了三个晚上才找到的问题：心跳不准，先怀疑你的 loop 有没有按预期频率被调用。

2.3 重连退避策略与重订阅

mosquitto_loop_start启动的后台线程里内置了重连逻辑，断线后会按reconnect_delay的间隔尝试重连。默认值我不太满意，因为它是一个固定间隔加简单递增，设备量大的时候很容易出现"整栋楼的设备同时重连把 broker 打爆"的惊群效应。

mosquitto_reconnect_delay_set(mosq, delay, delay_max, exponential_backoff)这个函数是我必调的。exponential_backoff传true时，重连间隔会按 2 的幂次增长：第一次等delay秒，第二次 2 倍，第三次 4 倍，直到撞上delay_max上限。以delay = 1、delay_max = 30为例，重试序列是 1、2、4、8、16、30、30……这样即使一千台设备同时掉线，重连请求也被摊平在时间轴上。

我在实测里还会叠加一个随机抖动：把delay设成1 + rand() % 3，进一步打散同时性。代价是恢复时间稍微变长，但相比于 broker 被打挂，这点代价太值了。

重连成功之后，必须重新订阅。这里要区分两种情况：clean_session = false时 broker 侧保留了订阅关系，理论上不用重订；但实践中我依然会在on_connect回调里无脑重订一遍，因为它幂等，多订一次不会有副作用，而漏订一次的后果是消息静默丢失，非常难查。另外别忘了发布在线状态，用 retain = true 发一条{"online":true}到设备专属主题，这样平台侧一订阅就能立刻知道设备状态，不用轮询。

2.4 QoS 等级与幂等设计

QoS 是 MQTT 里最容易被误用的东西。三个等级的实际语义是：QoS 0 最多一次，发出去就不管了，可能丢也可能到；QoS 1 至少一次，靠 PUBACK 确认，没收到会重发，所以业务侧可能收到重复消息；QoS 2 恰好一次，走 PUBLISH → PUBREC → PUBREL → PUBCOMP 四次握手，不重复不丢失，但开销大。

选型的判断标准不是"哪个更可靠"，而是你的业务对重复和丢失分别有多敏感。周期性上报的传感器数据，丢一两条无所谓，用 QoS 0 就够了，吞吐能高出好几倍。控制指令必须送到，用 QoS 1，然后在业务侧做幂等处理——给每条指令带一个单调递增的 seq 号，收到比当前 seq 小的直接丢弃。这个是关键：选了 QoS 1 却不做幂等，等于把重复问题推给了下游，迟早出事。

QoS 2 我一般只在计费、开关机这类"重复执行会出大事"的场景用。它的性能损失是实打实的：同样一条 100 字节的消息，QoS 0 就是一个报文，QoS 2 要走四个来回。而且当消息量大时，QoS 2 会在 broker 侧排起长队，max_inflight_messages默认才 20 条，队列一满新的 PUBLISH 就被丢弃或阻塞。

还有个容易被忽略的参数是retain。retain = true 的消息会被 broker 保存为对应主题的"最新值"，任何新订阅者一订阅立刻收到它。这玩意儿用来做"设备影子"特别方便，但滥用会导致 broker 内存膨胀——每个 retained 主题都要占一份内存，如果你用动态主题（比如带时间戳），retained 消息会无限堆积。我的原则是：只有状态类主题（在线状态、当前配置、最新读数）才用 retain，事件类主题一律不用。

3. 动手实现：一套可复用的客户端封装

3.1 环境准备与依赖编译

先装依赖。Debian/Ubuntu 系一条命令搞定：

sudo apt update sudo apt install -y libmosquitto-dev mosquitto-clients build-essential

libmosquitto-dev提供头文件<mosquitto.h>和静态/动态库，mosquitto-clients里的mosquitto_sub、mosquitto_pub是我们后面做验证的利器，强烈建议装上，它能帮你快速区分"是客户端的问题还是网络的问题"。

如果你想自己编译库源码（比如需要指定 MQTT 5 支持或者某些编译选项），从官方仓库拉下来用 CMake 构建：

cmake -B build -DWITH_TLS=ON -DWITH_CJSON=ON cmake --build build -j4 sudo cmake --install build

编译完用pkg-config --modversion libmosquitto确认一下版本。版本差异必须关注：libmosquitto 1.6 和 2.0 在mosquitto_loop_start的行为、mosquitto_threaded_set的地位、以及默认的 MQTT 协议版本上都有变化。2.0 之后默认走 MQTT 3.1.1，要上 MQTT 5 得在mosquitto_new之后显式设置。我踩过的坑就是本地是 2.0、现场设备是 1.6，代码里用了一个新版本的函数，交叉编译时才发现找不到符号。

本地测试环境还需要一个 broker。直接装一个：

sudo apt install -y mosquitto

Mosquitto 2.0 之后默认只监听回环地址，要让它对外提供服务得改配置。在/etc/mosquitto/conf.d/local.conf里加：

listener 1883 0.0.0.0 allow_anonymous true

改完sudo systemctl restart mosquitto重启。注意allow_anonymous true只适合内网调试，一旦暴露到公网，任何知道 IP 的人都能连上来往设备发指令，这个风险不用我多说。

3.2 对外接口设计

我设计了一套最小可用的接口，头文件如下：

/* mqtt_client.h */ #ifndef MQTT_CLIENT_H #define MQTT_CLIENT_H #include <stddef.h> typedef struct mqtt_client mqtt_client_t; typedef struct { const char *host; int port; const char *client_id; const char *username; const char *password; const char *cafile; /* NULL 表示不启用 TLS */ int keepalive; /* 秒，建议 30 */ int reconnect_delay; /* 首次重连间隔，秒 */ int reconnect_delay_max; /* 重连间隔上限，秒 */ const char *will_topic; /* 遗嘱主题，NULL 表示不设置 */ const char *will_payload; int will_qos; const char *online_topic; /* 上线通知主题，NULL 表示不发送 */ const char *online_payload; } mqtt_cfg_t; /* 消息回调：payload 不保证以 '\0' 结尾，用 len 判断长度 */ typedef void (*mqtt_on_message_fn)(const char *topic, const void *payload, size_t len, void *user); /* 状态回调：connected=1 表示已连接，0 表示断开 */ typedef void (*mqtt_on_state_fn)(int connected, void *user); int mqtt_client_init(mqtt_client_t **out, const mqtt_cfg_t *cfg, mqtt_on_message_fn on_msg, mqtt_on_state_fn on_state, void *user); int mqtt_client_subscribe(mqtt_client_t *c, const char *topic, int qos); int mqtt_client_publish(mqtt_client_t *c, const char *topic, const void *payload, size_t len, int qos, int retain); int mqtt_client_is_connected(mqtt_client_t *c); void mqtt_client_fini(mqtt_client_t *c); #endif

接口设计的几个决策点值得说清楚。为什么用回调而不是队列？队列需要额外的线程和锁，而业务侧大概率已经有自己的任务队列了，再套一层属于重复建设。回调最轻，业务侧想在回调里投递到自己的队列也完全可以。为什么 payload 带 len 而不是给字符串？MQTT 是二进制协议，payload 里可以有\0，可以是一段 protobuf 或者 CBOR 序列化数据。用strlen处理会截断，这个坑一旦踩了就是数据错乱，非常隐蔽。

3.3 核心实现代码拆解

先看结构体，所有状态收在一处：

/* mqtt_client.c */ #include "mqtt_client.h" #include <mosquitto.h> #include <stdlib.h> #include <string.h> #include <stdio.h> #define MAX_SUBS 32 #define MAX_TOPIC_LEN 256 struct sub_entry { char topic[MAX_TOPIC_LEN]; int qos; }; struct mqtt_client { struct mosquitto *mosq; mqtt_cfg_t cfg; mqtt_on_message_fn on_msg; mqtt_on_state_fn on_state; void *user; volatile int connected; struct sub_entry subs[MAX_SUBS]; int sub_count; };

connected加了volatile，因为它会被网络线程写、被业务线程读，不加的话编译器可能把它缓存进寄存器，导致业务侧永远读到旧值。这是个小细节，但在多线程场景下是致命的。

接下来是初始化，重点在几个顺序不能乱的调用：

static void cb_connect(struct mosquitto *mosq, void *ud, int rc); static void cb_disconnect(struct mosquitto *mosq, void *ud, int rc); static void cb_message(struct mosquitto *mosq, void *ud, const struct mosquitto_message *msg); int mqtt_client_init(mqtt_client_t **out, const mqtt_cfg_t *cfg, mqtt_on_message_fn on_msg, mqtt_on_state_fn on_state, void *user) { mqtt_client_t *c = calloc(1, sizeof(*c)); if (!c) return -1; c->cfg = *cfg; c->on_msg = on_msg; c->on_state = on_state; c->user = user; mosquitto_lib_init(); /* 第三个参数 clean_session：需要离线消息就传 false */ c->mosq = mosquitto_new(cfg->client_id, false, c); if (!c->mosq) { free(c); return -2; } mosquitto_connect_callback_set(c->mosq, cb_connect); mosquitto_disconnect_callback_set(c->mosq, cb_disconnect); mosquitto_message_callback_set(c->mosq, cb_message); if (cfg->username) mosquitto_username_pw_set(c->mosq, cfg->username, cfg->password); if (cfg->cafile) mosquitto_tls_set(c->mosq, cfg->cafile, NULL, NULL, NULL, NULL); mosquitto_reconnect_delay_set(c->mosq, cfg->reconnect_delay, cfg->reconnect_delay_max, true); if (cfg->will_topic) mosquitto_will_set(c->mosq, cfg->will_topic, (int)strlen(cfg->will_payload), cfg->will_payload, cfg->will_qos, false); int rc = mosquitto_connect(c->mosq, cfg->host, cfg->port, cfg->keepalive); if (rc != MOSQ_ERR_SUCCESS) { fprintf(stderr, "[mqtt] connect failed: %s\n", mosquitto_strerror(rc)); mosquitto_destroy(c->mosq); free(c); return -3; } /* loop_start 会拉起后台线程并处理重连 */ rc = mosquitto_loop_start(c->mosq); if (rc != MOSQ_ERR_SUCCESS) { mosquitto_destroy(c->mosq); free(c); return -4; } *out = c; return 0; }

这里的顺序有讲究：回调必须在 connect 之前设置，否则第一次连接成功的事件你可能就错过了，导致重订阅没执行。遗嘱消息也必须在 connect 前设置，因为它是 CONNECT 报文的一部分。mosquitto_reconnect_delay_set同理，要在 loop 启动前调好。

连接回调是整个封装的灵魂，它负责"断线重连后世界恢复原样"这件事：

static void cb_connect(struct mosquitto *mosq, void *ud, int rc) { mqtt_client_t *c = (mqtt_client_t *)ud; if (rc != 0) { fprintf(stderr, "[mqtt] CONNACK rc=%d (%s)\n", rc, mosquitto_connack_string(rc)); c->connected = 0; if (c->on_state) c->on_state(0, c->user); return; } c->connected = 1; /* 幂等重订阅，别管 broker 是否保留了会话 */ for (int i = 0; i < c->sub_count; i++) { mosquitto_subscribe(mosq, NULL, c->subs[i].topic, c->subs[i].qos); } /* 用 retain 发在线状态，新订阅者能立刻拿到 */ if (c->cfg.online_topic) { mosquitto_publish(mosq, NULL, c->cfg.online_topic, (int)strlen(c->cfg.online_payload), c->cfg.online_payload, 1, 1); } fprintf(stdout, "[mqtt] connected to %s:%d\n", c->cfg.host, c->cfg.port); if (c->on_state) c->on_state(1, c->user); }

注意rc是 CONNACK 返回码，不是 MOSQ_ERR 系列，两者含义完全不同，这里用mosquitto_connack_string转成可读文本，排查问题时能省很多事。

消息回调只做一件事——把数据搬走：

static void cb_message(struct mosquitto *mosq, void *ud, const struct mosquitto_message *msg) { mqtt_client_t *c = (mqtt_client_t *)ud; if (!c->on_msg) return; /* msg->payload 在回调返回后即失效，这里直接同步交给上层， 上层若需异步处理，请在回调内自行 memcpy 一份 */ c->on_msg(msg->topic, msg->payload, msg->payloadlen > 0 ? (size_t)msg->payloadlen : 0, c->user); }

我最开始是让业务侧直接拿指针存起来的，后来发现只要业务回调里 sleep 超过一秒，payload 就变成乱码了。根本原因是 libmosquitto 会在下一轮 loop 里复用这块缓冲区。现在的做法是在文档里写死：回调内必须完成拷贝。如果你嫌麻烦，可以在这层封装里帮你 memcpy，代价是每次消息多一次内存分配，高频场景下不建议。

发布和订阅的封装就直白多了：

int mqtt_client_publish(mqtt_client_t *c, const char *topic, const void *payload, size_t len, int qos, int retain) { if (!c || !c->mosq) return -1; /* mosquitto_publish 是线程安全的，可从业务线程直接调用。 mid 传 NULL 表示不关心消息确认。 */ return mosquitto_publish(c->mosq, NULL, topic, (int)len, payload, qos, retain); } int mqtt_client_subscribe(mqtt_client_t *c, const char *topic, int qos) { if (!c || c->sub_count >= MAX_SUBS) return -1; /* 记录到期望订阅表，重连时自动恢复 */ struct sub_entry *e = &c->subs[c->sub_count]; if (strlen(topic) >= MAX_TOPIC_LEN) return -2; strncpy(e->topic, topic, MAX_TOPIC_LEN - 1); e->topic[MAX_TOPIC_LEN - 1] = '\0'; e->qos = qos; c->sub_count++; if (c->connected) return mosquitto_subscribe(c->mosq, NULL, topic, qos); return 0; /* 未连接时先记账，连上后自动补订 */ }

最后是销毁，顺序不能错：

void mqtt_client_fini(mqtt_client_t *c) { if (!c) return; if (c->mosq) { mosquitto_disconnect(c->mosq); mosquitto_loop_stop(c->mosq, false); mosquitto_destroy(c->mosq); } mosquitto_lib_cleanup(); free(c); }

mosquitto_loop_stop的第二个参数force传false表示优雅退出（等当前操作完成），传true会强制中断。我优先用 false，因为强制中断可能导致一个正在处理的 QoS 1 消息没来得及发 PUBACK，broker 会重发。只有在退出流程被卡住时才升级到 true。另外mosquitto_lib_cleanup是进程级的，一个进程调一次就够，如果你有多个客户端实例，得自己加个引用计数，我在实际项目里就是用一个静态计数器解决的。

3.4 编译、运行与验证

编译命令很简单，关键是链接顺序：

gcc -Wall -O2 -o mqtt_demo main.c mqtt_client.c -lmosquitto -lpthread

-lmosquitto必须放在源文件后面，这是链接器的老规矩，顺序反了会报undefined reference。用 CMake 的话：

find_package(PkgConfig REQUIRED) pkg_check_modules(MOSQ REQUIRED libmosquitto) target_include_directories(demo PRIVATE ${MOSQ_INCLUDE_DIRS}) target_link_libraries(demo PRIVATE ${MOSQ_LIBRARIES} pthread)

验证环节我习惯开三个终端。第一个跑mosquitto_sub -h 127.0.0.1 -t 'dev/#' -v观察全部流量。第二个跑我们的客户端程序。第三个用mosquitto_pub往客户端订阅的主题发消息，看第一个终端和客户端是否都收到。

再测断线重连：把 broker 停掉，观察客户端日志里是否出现重连尝试；等几秒再把 broker 拉起来，看客户端是否自动恢复连接、重新订阅、并发出上线通知。这一套跑通，说明封装的核心逻辑是站得住的。实测的数据是：broker 重启到客户端恢复订阅完成，在reconnect_delay = 1的情况下大约 3 到 5 秒，具体取决于重试时机的对齐。

4. 踩坑记录与问题排查

4.1 错误码与常见故障速查

排查 MQTT 问题，第一件事永远是看清楚返回码属于哪一套。MOSQ_ERR_*是库自身返回的，mosquitto_connack_string(rc)对应的才是 broker 的应答。我整理了一张常用的对照表：

最后一行值得展开说。系统时间错误导致 TLS 失败这个坑我在嵌入式设备上遇到不下五次。设备没有 RTC，开机后时间是 1970 年，而证书的有效期是 2024 年起，校验必然失败。解决办法是开机后先通过 NTP 或从服务端拿一次时间，再建立 TLS 连接。

4.2 线程与内存的隐性坑

第一个坑是回调里做阻塞操作。前面提过，on_message在库的网络线程里跑，你在里面做一次 500ms 的数据库写入，心跳就被推迟 500ms。次数多了累积起来就会超时断线。我的做法是在回调里只做数据搬运，重活全部丢给业务侧的线程池。

第二个坑是mosquitto_loop_stop被回调调用。如果你在on_message里判断到某个条件就调用mqtt_client_fini，而fini里又调了mosquitto_loop_stop，那就是网络线程等自己退出，直接死锁。正确做法是置一个退出标志，让主线程去执行销毁。

第三个坑是mosquitto_message_copy的误用。库提供了这个函数来复制消息，但它复制出来的mosquitto_message需要调mosquitto_message_free释放，而且topic和payload是分开 malloc 的，只 free 结构体本身会泄漏两块内存。我在代码审查里见过这个错误，跑一个月泄漏了 200MB。

第四个坑是多线程调用mosquitto_publish时的 mid 冲突。如果你传了mid参数想追踪消息确认，在多个线程里各自维护 mid 会乱。要么统一传 NULL 不追踪，要么把发布操作收敛到单一线程。

4.3 生产环境的几条经验

设备量大之后，我总结了几个必须提前做的东西。一是连接数上限。Mosquitto 默认的max_connections是 -1（不限制），但受限于文件描述符，ulimit -n不改的话一千个连接就崩了，记得在 systemd 的 service 文件里加LimitNOFILE=65535。二是消息限流。设备端如果有突发上报，比如一次性推一万条历史数据，broker 的内存会被撑爆。我一般在设备端加一个发送队列，超过阈值就丢弃最旧的数据，宁可丢数据也不能把 broker 拖死。

三是主题设计要一次定死。我习惯的格式是产品线/设备ID/方向/业务类型，例如sensor/A1B2C3/up/telemetry、sensor/A1B2C3/down/cmd。这种层级结构的好处是平台侧可以用sensor/+/up/#一次性订阅所有设备的上行数据，做权限控制时也能按前缀划分。千万不要在主题里放时间戳或者随机数，会造成大量不重复的主题，retained 消息一旦开了就是内存黑洞。

四是 QoS 不要一刀切。我见过有人为了"可靠"把所有消息都设成 QoS 2，结果设备上报频率一上来，broker 的 inflight 队列全满，消息开始排队延迟，反而比 QoS 0 更难用。正确的做法是按业务分：遥测数据 QoS 0，控制指令 QoS 1，配置下发 QoS 1 + retain，涉及钱和安全的用 QoS 2。

最后分享一个排查小技巧。当你不确定问题出在客户端还是 broker 时，先退到原生命令行工具复现一遍。用mosquitto_sub -d打开调试输出，它会打印每一条收发的报文类型和内容。如果mosquitto_sub能收到消息而你的程序收不到，问题就在你的代码里；两边都收不到，那就是 broker 配置或者主题匹配的问题。这个二分法帮我节省的时间，比任何调试器都多。

我自己在这个封装上迭代过三四个版本，从最早的裸调用到现在的样子，最大的感受是：MQTT 客户端的难点从来不在协议本身，而在于断线、重连、重复、超时这些边界情况。把这几件事用一层薄薄的代码封死，业务代码就能回到它该有的样子。