基于 Mosquitto 的 MQTT C 客户端封装：自动重连与订阅管理实战

标题里写的是 mosquito，但真正落到代码里，这个拼写几乎所有人都会写错一次——Eclipse 官方那个 MQTT broker 项目叫 Mosquitto，多一个 t。这个细节看着不起眼，却是我排查问题时踩过的第一个坑：搜依赖包名搜不到、链接器报找不到库、apt 装完发现装了个不相干的包，全是因为少敲了一个字母。基于 Mosquitto 封装的 MQTT 客户端，说白了就是拿 libmosquitto 这个 C 库当底座，把连接管理、订阅关系、线程同步、重连退避、消息分发这些脏活累活全部收拢到一个薄壳里，让上层业务代码只看到 publish 和 subscribe 两个动作，不用关心 socket 什么时候断的、重连到第几次了、QoS 2 的握手跑到哪一步了。这套东西适合谁：写 C/C++ 的嵌入式或服务端开发者、做物联网上报通道的人、写边缘网关的人，以及那些用 Java、Python、Android 调 MQTT 但被客户端的重连逻辑折磨过的同学——接口设计的思路是通用的，换个语言照样能用。我把这套封装前后改了三版，从最早的一坨回调糊在业务里，到现在能扛住弱网反复抖动，中间踩的坑基本都在这篇里了。

1. 先搞清楚：为什么要给 libmosquitto 再套一层壳

1.1 直接用 libmosquitto 的四个硌手点

我不止一次见过这样的代码：一个几百行的 main.c，里面塞着 libmosquitto 的初始化和业务逻辑，消息回调里直接写数据库插入，连接回调里塞了二十行的重订阅代码，程序跑起来能用，但没人敢改。这不是开发者水平问题，而是 libmosquitto 本身就是一层很薄的 C 接口，它的设计目标是「给你能力」，不是「替你干活」。它的硌手点非常具体。

第一个是异步回调模型把业务逻辑切碎了。你调用mosquitto_connect()，函数立刻返回，真正的连接结果要通过mosquitto_connect_callback_set()注册的回调才知道。发布消息也是异步的，mosquitto_publish()返回成功只代表消息进了本地发送队列，不代表 broker 收到了。业务同学习惯的「调用—返回—判断结果」这条直线被打断成了回调驱动，代码可读性直接掉一档。

第二个是重连必须自己兜。libmosquitto 提供了mosquitto_reconnect_delay_set()帮你做退避，但重连成功之后要重新订阅哪些主题、内存里缓存的会话状态要不要清、正在等待确认的消息怎么处理，这些全是你的事。我见过最典型的 bug 就是断线重连后消息收不到了，查半天发现是订阅只在程序启动时做了一次，重连后压根没重新订阅。

第三个是线程模型要自己拿主意。mosquitto_loop_forever()是阻塞的，适合独占一个线程；mosquitto_loop_start()会起一个后台网络线程，主线程就能自由地调 publish。这两种模式不能混着理解，而且后者意味着你的回调函数是在另一个线程里执行的，一旦回调里碰了共享数据，锁就得自己加。

第四个是参数散落在十几个 setter 里。keepalive 在 connect 里传，遗嘱在mosquitto_will_set()里设，TLS 在mosquitto_tls_set()里配，重连退避在mosquitto_reconnect_delay_set()里调，最大飞行窗口在mosquitto_max_inflight_messages_set()里改。这些参数彼此之间有耦合关系（比如遗嘱的 QoS 和会话持久化会影响断线期间能否收到消息），散着放特别容易漏。

注意：libmosquitto 的具体函数签名在不同大版本之间有过调整，1.6 和 2.0 之间就有若干 setter 的存废变化。写封装时别照抄网上的老代码，直接用grep翻本机/usr/include/mosquitto.h最靠谱。

1.2 封装层应该解决的六件事

想明白痛点之后，接口该长什么样基本就定了。我给自己定的标准是：业务层不允许出现struct mosquitto *这个类型，所有 MQTT 相关的头文件只能在封装层内部 include。围绕这个原则，封装层要解决六件事。

连接管理是第一位。上层只关心「现在能发消息吗」，不关心底层连没连上。封装层对外暴露一个is_connected()状态查询，内部负责在断开后按指数退避重连，同时把重连次数、上次断开原因记进日志。

订阅表的集中维护是第二位。每个订阅关系记录三样东西：主题字符串、QoS 等级、回调函数指针。这张表存在内存里，断线重连之后遍历它重新订阅，业务代码写一次订阅就永远生效。这一步是解决「重连丢订阅」的根本办法。

线程安全的发布入口是第三位。上层可能在任何线程里调 publish，封装层要保证发送动作不会因为多线程竞争炸掉。稳妥的做法是每个客户端实例配一把发送锁，publish 的时候加锁，虽然有一点性能损失，但在每秒几万条以内的场景里完全感受不到。

回调分发是第四位。底层一个on_message回调进来，封装层要按主题去订阅表里做匹配（要支持+和#通配符），找到对应的业务回调，把消息拷一份传出去。注意是拷一份，不是直接传指针。

可观测性是第五位。连接、断开、重连、订阅成功、发布失败，这些事件都要有明确的日志出口，最好带客户端 ID 和主题，方便多实例环境下定位。顺带维护几个计数器：累计发送、累计接收、当前待确认、重连次数，出问题时看一眼就知道大概方向。

配置集中是第六位。所有参数收进一个结构体，从配置文件或者环境变量里读，一处修改全局生效。这一条看着最没技术含量，但它是后面调优的基础——参数散着放，你就没法系统地做压测对比。

1.3 三条路线：自封装、Paho、各语言现成客户端

有人会问，为什么不直接用 Paho 或者现成的高层库。我做过对比，结论是取决于你在哪个技术栈上。

我最后选自封装，原因很实际：目标环境是 ARM 板子，编译出来的库要控制在几百 KB，同时要求断线重连策略能按网络类型动态调（有线网和蜂窝网的退避参数不一样）。这种情况用通用客户端反而是负担。如果你只是写个服务端程序，用现成的更省事，别为了造轮子而造轮子。

2. 动手前的准备：环境搭建与工程骨架

2.1 libmosquitto 装法：包管理器和源码编译怎么选

开发机上最快的路是包管理器。Debian 系一行搞定运行库和开发头文件：apt-get install libmosquitto-dev，CentOS 系是yum install mosquitto-devel。装完先验一下，pkg-config --cflags --libs libmosquitto能输出正确的路径和库名就说明环境通了。

交叉编译到 ARM 板子就必须走源码编译。下载源码包，用 CMake 交叉编译，关键参数是关掉不需要的东西。我常用的一组配置是这样的：

cmake -B build \ -DCMAKE_C_COMPILER=arm-linux-gnueabihf-gcc \ -DCMAKE_INSTALL_PREFIX=$PWD/out \ -DWITH_TLS=ON \ -DWITH_TLS_PSK=OFF \ -DWITH_SRV=OFF \ -DDOCUMENTATION=OFF \ -DWITH_APPS=OFF \ -DWITH_BROKER=OFF cmake --build build -j4 cmake --install build

关掉WITH_BROKER很关键，你只要客户端能力，broker 那一整套会把产物撑大一倍以上。WITH_APPS是 mosquitto_pub/sub 那两个命令行工具，开发阶段建议留着，调试验证的时候能省很多事，量产裁剪时再关。WITH_SRV是 DNS SRV 记录查询，绝大多数场景用不上。

提示：如果目标板子的文件系统空间紧张，可以进一步关掉静态库只留动态库，或者反过来只留静态库省掉部署.so 的步骤。两者选一就行，别同时链。

2.2 CMake 侧怎么接：静态链接与动态链接的取舍

上层工程接这个封装库，链接方式的选择有实际影响。动态链接.so的好处是多个进程共享内存映射，升级库不用重编程序；坏处是部署时少一个文件程序就跑不起来，而且版本不对会出莫名其妙的符号错误。静态.a的好处是自包含，扔过去就能跑；坏处是程序体积变大，且库里的全局状态（比如mosquitto_lib_init()建立的引用计数）在每个进程里独立。

我的做法是：嵌入式整机固件用静态链接，服务端程序用动态链接。CMake 里的写法如下：

find_package(PkgConfig REQUIRED) pkg_check_modules(MOSQ REQUIRED libmosquitto) add_library(mqtt_wrapper STATIC src/mqtt_client.c src/mqtt_sub_table.c ) target_include_directories(mqtt_wrapper PUBLIC ${CMAKE_CURRENT_SOURCE_DIR}/include) target_link_libraries(mqtt_wrapper PUBLIC ${MOSQ_LIBRARIES}) target_include_directories(mqtt_wrapper PUBLIC ${MOSQ_INCLUDE_DIRS}) target_link_libraries(mqtt_wrapper PUBLIC pthread)

注意pthread必须显式链上。libmosquitto 在用mosquitto_loop_start()的时候内部要起线程，链接时漏了 pthread 会报一堆 undefined reference topthread_create，这个报错信息挺直白，但第一次遇到的人常常去怀疑库没装上。

2.3 目录结构和对外接口长什么样

工程骨架我习惯分三层：include/放对外头文件，src/放实现，test/放自测。对外头文件里只有配置结构体、句柄类型和若干函数声明，绝不 includemosquitto.h。

/* include/mqtt_client.h —— 业务侧唯一需要看到的头文件 */ #ifndef MQTT_CLIENT_H #define MQTT_CLIENT_H #include <stdbool.h> #include <stddef.h> #include <stdint.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; int keepalive; /* 秒，0 表示关闭心跳 */ bool clean_session; /* false 时依赖 broker 缓存离线消息 */ int max_inflight; /* 同时未确认的最大消息数 */ int reconnect_delay_s; /* 首次重连等待 */ int reconnect_delay_max_s; bool exponential_backoff; const char *will_topic; /* 为 NULL 表示不设遗嘱 */ const char *will_payload; int will_qos; bool will_retain; } mqtt_config_t; typedef void (*mqtt_msg_cb)(const char *topic, const void *payload, size_t payloadlen, int qos, void *userdata); typedef void (*mqtt_state_cb)(bool connected, int reason, void *userdata); mqtt_client_t *mqtt_client_create(const mqtt_config_t *cfg); int mqtt_client_start(mqtt_client_t *c); /* 内部起线程并自动重连 */ int mqtt_client_subscribe(mqtt_client_t *c, const char *topic, int qos, mqtt_msg_cb cb, void *userdata); int mqtt_client_publish(mqtt_client_t *c, const char *topic, const void *payload, size_t len, int qos, bool retain); bool mqtt_client_is_connected(mqtt_client_t *c); void mqtt_client_set_state_cb(mqtt_client_t *c, mqtt_state_cb cb, void *userdata); void mqtt_client_stop(mqtt_client_t *c); void mqtt_client_destroy(mqtt_client_t *c); #endif

这套接口有一个刻意的设计：mqtt_client_subscribe()允许在启动前后任意时刻调用。启动前调用只是记录到订阅表，启动后调用是记录加立即发送订阅请求。这个语义统一很重要，否则业务代码就得关心「我是在连接建立前还是后调用的」，那就等于把底层的坑又漏出去了。

3. 核心实现：把连接、订阅、发布包成三个动作

3.1 连接生命周期与自动重连的落地写法

连接部分的实现关键是把 libmosquitto 的回调翻译成自己的状态机。内部结构体大概是这样：

struct mqtt_client { struct mosquitto *mosq; mqtt_config_t cfg; bool connected; bool stopping; int reconnect_count; pthread_mutex_t sub_lock; pthread_mutex_t send_lock; mqtt_sub_entry_t *subs; /* 订阅表，动态数组 */ size_t sub_count; mqtt_state_cb state_cb; void *state_ud; };

创建阶段把配置翻译成 setter 调用，顺序有讲究，参数必须先设好再连接：

mqtt_client_t *mqtt_client_create(const mqtt_config_t *cfg) { mqtt_client_t *c = calloc(1, sizeof(*c)); c->cfg = *cfg; pthread_mutex_init(&c->sub_lock, NULL); pthread_mutex_init(&c->send_lock, NULL); c->mosq = mosquitto_new(cfg->client_id, cfg->clean_session, c); if (!c->mosq) { free(c); return NULL; } if (cfg->username) mosquitto_username_pw_set(c->mosq, cfg->username, cfg->password); mosquitto_max_inflight_messages_set(c->mosq, cfg->max_inflight); mosquitto_reconnect_delay_set(c->mosq, cfg->reconnect_delay_s, cfg->reconnect_delay_max_s, cfg->exponential_backoff); if (cfg->will_topic) mosquitto_will_set(c->mosq, cfg->will_topic, (int)strlen(cfg->will_payload), cfg->will_payload, cfg->will_qos, cfg->will_retain); mosquitto_connect_callback_set(c->mosq, on_connect); mosquitto_disconnect_callback_set(c->mosq, on_disconnect); mosquitto_message_callback_set(c->mosq, on_message); mosquitto_log_callback_set(c->mosq, on_log); return c; }

连接回调是整个封装的枢纽，重连后的重订阅就在这里做：

static void on_connect(struct mosquitto *mosq, void *ud, int rc) { mqtt_client_t *c = ud; if (rc != 0) { /* rc 常见值：1 协议版本不支持，4 用户名密码错误，5 未授权 */ LOG_WARN("connect refused rc=%d client_id=%s", rc, c->cfg.client_id); return; } c->connected = true; c->reconnect_count = 0; pthread_mutex_lock(&c->sub_lock); for (size_t i = 0; i < c->sub_count; i++) { if (c->subs[i].active) mosquitto_subscribe(mosq, NULL, c->subs[i].topic, c->subs[i].qos); } pthread_mutex_unlock(&c->sub_lock); if (c->state_cb) c->state_cb(true, 0, c->state_ud); }

有两个细节值得单独说。第一，on_connect是在网络线程里执行的，在里面直接调mosquitto_subscribe()是安全的，这是 libmosquitto 的约定，因为网络线程本身就是消息循环的执行者。第二，重连计数清零放在这里而不是放到on_disconnect里，因为只有真正连上了才说明退避策略生效了，清零才有意义。

退避参数怎么设，我的经验值是：首次 1 秒，上限 30 秒，开启指数退避。这样做的结果是重连间隔按 1、2、4、8、16、30、30……增长。上限压在 30 秒的原因是，MQTT 场景下消息时效性通常要求秒级到分钟级，退避太长会让恢复延迟变得不可接受。如果设备量很大（比如上万台同时掉线），上限反而要放长一点到 60 秒以上，避免恢复瞬间把 broker 打爆——这就是所谓的惊群效应，broker 刚重启完，几万台设备在同一个两秒窗口里全部重连上来，连接数还没建完又被压垮。

3.2 线程安全的发布与订阅

发布接口看着简单，实际要考虑的是队列和锁。libmosquitto 在开启网络线程后，本身对mosquitto_publish()这类操作有内部保护，但不同版本的实现细节有过调整，我不想把正确性押在版本行为上，所以自己加了一把递归锁：

int mqtt_client_publish(mqtt_client_t *c, const char *topic, const void *payload, size_t len, int qos, bool retain) { if (!c || !c->connected) return -1; pthread_mutex_lock(&c->send_lock); int rc = mosquitto_publish(c->mosq, NULL, topic, (int)len, payload, qos, retain); pthread_mutex_unlock(&c->send_lock); if (rc != MOSQ_ERR_SUCCESS) LOG_WARN("publish failed topic=%s rc=%d", topic, rc); return rc == MOSQ_ERR_SUCCESS ? 0 : -1; }

要不要在没连接的时候也允许发布，这是个产品决策。我倾向于返回失败让上层知道，因为静默缓存会带来内存无限增长的风险。如果业务确实需要「先缓存后补发」，那就必须给缓存队列设上限，超了就丢最老的，这个策略要写进文档。

订阅接口的设计比发布更绕一点，原因是订阅表要在多个线程里读写：业务线程调 subscribe 往里加，网络线程在 on_connect 里遍历、在 on_message 里匹配。所以订阅表的每一次访问都在sub_lock保护下。加新订阅的处理逻辑是：

int mqtt_client_subscribe(mqtt_client_t *c, const char *topic, int qos, mqtt_msg_cb cb, void *ud) { pthread_mutex_lock(&c->sub_lock); /* 先查重：同一主题重复订阅则覆盖回调，避免表膨胀 */ mqtt_sub_entry_t *e = sub_find(c, topic); if (!e) { e = sub_append(c); } if (!e) { pthread_mutex_unlock(&c->sub_lock); return -1; } snprintf(e->topic, sizeof(e->topic), "%s", topic); e->qos = qos; e->cb = cb; e->ud = ud; e->active = true; bool need_send = c->connected; pthread_mutex_unlock(&c->sub_lock); if (need_send) return mosquitto_subscribe(c->mosq, NULL, topic, qos) == MOSQ_ERR_SUCCESS ? 0 : -1; return 0; /* 未连接时只登记，on_connect 时统一补发 */ }

查重这一步是实际项目里很重要的一个小优化。我之前写的一个网关，业务代码在配置热更新时反复调 subscribe，主题字符串完全一样，结果订阅表涨到了几千条，每次重连都要发几千个 SUBSCRIBE 报文，broker 那边直接爆了。加了查重之后，表大小就等于实际业务的主题数量，重连开销可控。

3.3 回调分发：让业务代码完全看不到 mosquitto

消息回调进来之后，要做的第一件事是立刻把 payload 拷出来。原因很实在：mosquitto_message里的payload指针在回调返回后就不保证有效了，如果你把指针丢进另一个线程的队列里慢慢处理，等着你的就是随机崩溃或者读到垃圾数据。这是我的封装第二版修过的一个真实 bug，现象是压测跑了十分钟左右程序莫名其妙挂掉。

static void on_message(struct mosquitto *mosq, void *ud, const struct mosquitto_message *msg) { mqtt_client_t *c = ud; if (!msg->topic || msg->payloadlen <= 0) return; pthread_mutex_lock(&c->sub_lock); mqtt_sub_entry_t *best = NULL; for (size_t i = 0; i < c->sub_count; i++) { if (!c->subs[i].active) continue; if (!topic_match(c->subs[i].topic, msg->topic)) continue; /* 多主题都能匹配时，优先选层级更多的那条，避免 # 抢走精确订阅 */ if (!best || topic_depth(c->subs[i].topic) > topic_depth(best->topic)) best = &c->subs[i]; } if (!best) { pthread_mutex_unlock(&c->sub_lock); return; } mqtt_msg_cb cb = best->cb; void *cb_ud = best->ud; pthread_mutex_unlock(&c->sub_lock); if (cb) cb(msg->topic, msg->payload, (size_t)msg->payloadlen, msg->qos, cb_ud); /* 回调内部约定为同步处理或自行拷贝 */ }

topic_match要支持 MQTT 的通配符规则：+匹配单层，#匹配多层且必须是最后一位。有三条规则特别容易漏：第一，sport/tennis/#能匹配sport/tennis本身，也能匹配sport/tennis/player1，所以#前面的斜杠处理要小心；第二，以$开头的主题（broker 自带的$SYS/#统计信息）不会被通配符#或+匹配到，这个设计是为了防止统计信息把普通消息通道淹了；第三，匹配是大小写敏感的，Sensor/A和sensor/a是两个完全不同的主题。

匹配用最细粒度优先的策略，是我在第二版加上的。原因很实际：一台设备既订阅了dev/12345/cmd，又订阅了dev/+/status，如果先匹配到通配那条，精确订阅的业务回调就永远收不到消息。规则定成「层级多的优先，层级相同按注册顺序」之后，行为就可预期了。

注意：如果业务回调是耗时操作（写数据库、发 HTTP 请求），绝不能直接在这样的同步回调里执行，会把整个网络线程堵死，导致心跳报文发不出去进而被 broker 判定掉线。正确做法是在回调里把消息拷进自己的业务队列，由工作线程消费。

3.4 QoS、retain 和遗嘱消息的落地细节

这三个参数决定了消息可靠性的天花板，封装层最好把选择理由写进注释，不然半年后接手的人会一脸茫然。

选 QoS 1 的时候一定要配套做幂等。做法很简单，消息体里带一个单调递增的序号或者业务 ID，接收端用一个最近 N 条的环形缓存去重。我在一个电表上报项目里就吃过亏：QoS 1 在网络抖动时会重发，没做去重导致后端日电量统计一天多算了十几度，查了两天才定位到。

retain 标志的作用是让 broker 为这个主题保留最后一条消息，新的订阅者一连上就能立刻收到。它特别适合「设备当前状态」这种场景，比如dev/12345/online发布一条 retain 消息，任何时候有人订阅这个主题，都能马上知道设备现在的状态，不用干等下一次上报。但 retain 有个副作用要记住：它不会自动过期（除非 broker 支持消息过期属性）。如果设备下线了没人发新的 retained 消息覆盖，订阅者会一直收到那条旧的「在线」状态。所以设备主动下线时要记得发一条覆盖消息。

遗嘱消息是 broker 帮你发的「我死了」通知。客户端在连接时声明遗嘱主题和内容，当 broker 检测到连接异常断开（心跳超时、TCP 断开且没发 DISCONNECT 报文）时，会自动以这个客户端的名义发布遗嘱内容。这跟许多应用层自己搞的心跳超时检测相比，最大的好处是通知及时且不需要额外的服务端逻辑。遗嘱的 QoS 和 retain 要单独设，我一般用 QoS 1 加 retain true，这样看板页面切进来的时候也能立刻看到设备离线状态。

4. 参数调优：心跳、超时、队列与内存

4.1 keepalive 到底设多少才合适

keepalive 的机制是这样的：客户端承诺在 keepalive 秒内有数据往来，如果这段时间内没发过任何报文，就主动发一个 PINGREQ；broker 端如果在 1.5 倍 keepalive 时间内没收到任何报文，就判定连接断开并触发遗嘱。所以这个值是双向的约定，不是单方面的超时。

设多大有两头约束。设小了，报文频率高，功耗和流量上去了，对电池设备不友好；设大了，断线检测慢，一台设备拔了网线，broker 要等 1.5 倍的时间才发现，消息堆积和状态不一致的窗口就变长。经验值是：插电设备 30 到 60 秒，电池设备 120 到 300 秒。还有个容易被忽略的因素是网络中间设备的状态表老化时间，很多家用路由器和大规模网络里的地址转换设备，空闲连接的老化时间在几分钟量级，keepalive 设得比这个时间还长，通道就可能被中间设备悄悄回收，表现出一段时间空闲后第一条消息必然超时重发的怪现象。把 keepalive 压在 60 秒以内基本可以绕开这个问题。

设成 0 表示关闭心跳，我不建议在生产环境这么干，除非你有应用层的保活机制。有些同学为了省电设 0，结果设备掉线后服务端半小时都不知道，这个代价比多发几个心跳报文大得多。

4.2 积压消息、飞行窗口和背压处理

这里有两个容易混淆的概念：飞行窗口（inflight）和离线队列。

飞行窗口指的是同时处于「已发出但未收到确认」状态的消息数量上限，由mosquitto_max_inflight_messages_set()控制，libmosquitto 里默认值是 20。这个值的含义是，QoS 1/2 的消息最多只有这么多条同时在途。设小了吞吐上不去，因为要等前面的确认回来才能发下一条；设大了内存占用上去，而且一旦网络质量差，大量消息卡在途中等确认，重连时都要重发，恢复会变慢。我的经验值是按带宽和 RTT 算：如果单条消息 200 字节，RTT 50 毫秒，窗口 20 就意味着每秒最多发 400 条左右。真要跑高吞吐，把窗口调到 50 到 100，但前提是接收端处理得过来。

离线队列是另一回事，它跟clean_session = false配合使用。当客户端以持久会话方式连接过之后断开，broker 会为它保留订阅关系，并把这段时间内匹配订阅的 QoS 1/2 消息缓存下来，等客户端重连后补发。缓存量由 broker 侧的max_queued_messages配置控制。这条机制能不能生效，取决于三个条件同时满足：客户端用 clean_session=false、订阅的 QoS 至少为 1、broker 侧队列没满。任缺一条，断线期间的消息就永久丢了。

这里有个我踩过的坑：开发阶段 broker 用默认配置，队列上限是 1000 条，测试环境消息少，一切正常。上线后设备批量断线，队列瞬间打满，超出部分静默丢弃——注意是静默，broker 不会告诉你丢了。后来我改成在设备侧做应用层的序号校验，接收端发现序号不连续就主动向设备请求补数据，等于加了一层兜底。

发布侧也要考虑背压。当封装层内部有发送队列时（比如自己做异步发送队列），必须设上限，超限时的策略要明确：阻塞等待、丢弃最旧、还是直接返回失败。我选的是直接返回失败并把事件记入日志，让业务层看得见压力，而不是让队列悄悄吃掉内存。嵌入式设备上一个几百 MB 的内存泄漏就足以让整机重启。

4.3 账号鉴权与 TLS 怎么接

先给一个结论：生产环境不要开匿名访问。这看起来像废话，但我在真实网络里扫描过的对外开放的 MQTT 端口，绝大多数都是允许匿名连接的，任何人都能订阅全部主题。做封装的时候把鉴权参数做成必填项，能挡住至少一半的疏忽。

账号密码的写法前面代码里已经出现过，mosquitto_username_pw_set()一行就够了。broker 侧的配置要注意 Mosquitto 2.0 之后的行为变化：默认配置只监听本地回环地址，要对外提供服务必须显式写监听配置和鉴权配置。

# mosquitto.conf 关键片段 listener 1883 allow_anonymous false password_file /etc/mosquitto/passwd # 需要网页端直连时再开 WebSocket listener 9001 protocol websockets

密码文件用mosquitto_passwd -c /etc/mosquitto/passwd 用户名生成，会把密码做单向散列后存进去，比明文配置安全得多。

TLS 是在mosquitto_username_pw_set()之外单独接的，用mosquitto_tls_set()传入 CA 证书、客户端证书和私钥的文件路径。三条经验：第一，证书路径用绝对路径，程序用相对路径加载证书在不同工作目录下行为不同，是经典的「开发机好使部署机报错」；第二，验证服务端证书的 hostname 一定要开，mosquitto_tls_opts_set()里传SSL_VERIFY_PEER和对应的 hostname，否则中间人替换证书你也发现不了；第三，客户端证书不是必须的，双向认证在设备量大的时候证书分发和轮换会非常痛苦，很多场景用「账号密码加服务端证书单向校验」就够了，安全级别取决于你的威胁模型，别一上来就上最重的方案。

提示：如果 TLS 握手失败，libmosquitto 的错误码往往只有一个笼统的值，看不出具体原因。把mosquitto_log_callback_set()注册上，日志里能打印出 OpenSSL 层的具体报错，排查效率提升好几倍。

5. 实测踩坑与排查速查表

5.1 十个高频故障现象与定位思路

我把过去两年在这个封装上处理过的问题整理成了速查表，出问题时按现象对号入座能省不少时间。

这张表里最值得展开的是「中文乱码」和「回调指针」这两条，因为它们都属于改起来简单但发现起来很费劲的类型。

中文乱码的根因是把 payload 当 C 字符串处理。MQTT 的 payload 是二进制安全的，长度由payloadlen给出，里面完全可能包含 0 字节，也可能没有结尾的 0。有些同学图省事写了printf("%s", msg->payload)，一旦消息里出现非 UTF-8 的二进制内容或者恰好没有终止符，就会读到越界内存。正确做法是永远带上长度，需要当字符串用时再单独申请一块len + 1的内存补上结尾 0。

回调指针那条我在前面提过，这里补充一个典型场景：有人为了不阻塞网络线程，在on_message里把msg->payload的指针丢进了一个队列，然后工作线程从队列里取出来处理。压测时一切正常，因为内存还没来得及被复用；跑到十几分钟内存压力上来之后，开始出现零星的解析错误，再往后就是段错误。现象随机、复现困难、日志看不出规律，是这类 bug 的标准特征。

5.2 线程与消息循环的三个经典错误

第一个错误是把mosquitto_loop_forever()和mosquitto_loop_start()混用。前者是阻塞式，调用之后当前线程就一直在跑消息循环，后面的代码根本不会执行；后者是非阻塞的，起一个后台线程跑循环，主线程继续干别的。两个都调用的结果是两条消息循环同时在跑同一个连接，行为未定义，通常表现为消息回调被调用两次或者干脆崩掉。我的封装统一用loop_start()，因为它更适合多线程场景。

第二个错误是在线程还没停的时候销毁对象。正确的关闭顺序是这样的：

void mqtt_client_stop(mqtt_client_t *c) { if (!c) return; c->stopping = true; /* 先发正常的 DISCONNECT 报文，让 broker 不触发遗嘱 */ mosquitto_disconnect(c->mosq); /* 等待网络线程退出，false 表示等待完成 */ mosquitto_loop_stop(c->mosq, false); c->connected = false; } void mqtt_client_destroy(mqtt_client_t *c) { if (!c) return; if (c->mosq) { mosquitto_destroy(c->mosq); /* 内部会释放订阅和未完成的消息 */ c->mosq = NULL; } free(c->subs); pthread_mutex_destroy(&c->sub_lock); pthread_mutex_destroy(&c->send_lock); free(c); }

顺序反过来写，也就是先 destroy 再 stop，等待你的是「回调里访问了已经释放的 userdata」这类崩溃。另外注意mosquitto_lib_cleanup()是进程级调用，一个进程里多次 init 之后只需在最外层统一 cleanup 一次，在多实例场景里到处调它会导致其他实例的底层资源被提前回收。

第三个错误是在回调里加锁的顺序和自己业务代码里的顺序相反。死锁这东西一旦发生就是挂死，没有日志、没有崩溃、CPU 也不高，特别难查。我的做法是强制规定锁的获取顺序：先拿sub_lock再拿send_lock，任何地方都不许反过来。同时规定回调里不允许持有sub_lock去调用业务回调——前面代码里我是先从表里取出回调指针、解锁、再调用，就是为了避开这个坑。

5.3 断线重连丢消息的真实复现与解决

这个坑我完整复现过一次，过程值得记录。测试环境是一台设备通过可编程交换机连 broker，脚本每 30 秒断网 10 秒，模拟弱网。现象是每次断网恢复后，断网期间设备侧发布的 5 到 10 条消息服务端都收不到，而设备侧日志显示 publish 全部返回成功。

第一次分析：怀疑是 publish 在未连接状态下也返回了成功。查代码发现mqtt_client_publish里确实有if (!c->connected) return -1;的判断，但有个时序漏洞——连接断开到on_disconnect回调被触发之间有一个窗口，这个窗口里c->connected还是 true，消息发出去了，实际都进了黑洞。修复办法是把connected标志的维护改成更保守的策略：在on_disconnect里立刻置 false，同时 publish 的返回值要区分「参数错误」「未连接」「队列满」三种情况返回不同错误码。

第二次分析：修复了时序漏洞之后仍有少量消息丢失。这次抓包看到了真相：设备侧消息确实发到了 broker，但重连后没有重新订阅，所以服务端订阅者收不到。这个场景的特殊之处在于，设备侧同时也是订阅者（接收下发指令），断线重连后订阅关系丢了。修复办法就是前面 3.1 节里的那段代码，在on_connect里遍历订阅表重发 SUBSCRIBE。

第三次分析：订阅恢复了，但断线那 10 秒内 broker 缓存的消息也没补下来。原因就是 clean_session 设成了 true。改成 false 之后，broker 在设备离线期间为它缓存了 QoS 1 的消息，重连后一次性补发。这里有个副作用要提前想清楚：clean_session=false意味着 broker 侧要为每台设备维护会话状态，设备数量到十万级的时候 broker 的内存开销会明显上升，需要评估。

三次排查分别对应了三个不同层面的问题：应用层的状态标志时序、协议层的订阅重建、broker 层的会话保持。这也是我坚持把这套封装写出来的原因——如果没有封装层，这三个问题会散落在各个业务模块里，每次都得重新排查一遍。封装做对了，这三个坑只需要踩一次。

6. 封装好之后：日志、压测与跨语言复用

6.1 日志埋点该记哪些，怎么记

日志这件事，我一开始是随便记的，出问题的时候才发现关键信息全没有。后来定了一套规则，现在回头看这套规则救过我很多次。

连接相关的事件全部记 INFO 及以上级别，包含时间戳、客户端 ID、broker 地址端口、断开原因码、本次连接持续时长。断开原因这一项很多人会漏，其实 libmosquitto 在断开回调里给出的原因码能区分「自己主动断开」「网络错误」「broker 拒绝」三种情况，对定位问题帮助巨大。

消息相关的事件用 DEBUG 级别，只记主题、长度、QoS，不记 payload 内容。不记内容有两个考虑：一是流量大，二是消息体里可能有业务敏感数据，日志落到哪都不合适。调试阶段需要看内容的时候，用一个编译开关单独打开，不要把 payload 日志留到生产环境。

发布失败要单独记 WARN，带错误码和主题。这个是最容易被忽略的一类日志，因为它平时不出现，一旦出现往往就是问题开始的信号。我在一个项目里加了这条日志之后才发现，某类主题在特定时刻的失败率高得离谱，追下去发现是那类主题的消息体超过了 broker 配置的最大报文长度限制。

计数器建议至少维护这四个：累计发布条数、累计接收条数、累计重连次数、当前订阅主题数。把它们做成一个可以随时查询的结构，配合定时打印，就能在客户投诉之前发现问题苗头。比如重连次数在半小时内涨了五十次，说明网络链路有问题，这时候主动去看一眼比等消息大量丢失之后再查要好得多。

6.2 压测怎么做，容量怎么估

压测这件事，最容易犯的错误是拿单条消息单次发送去测延时，然后乘以一个想当然的并发数。这个算法得出的结论和实际运行情况能差一个数量级。

我用的方法分三步。第一步是慢速基线，一秒钟发一条消息，测最干净情况下的往返延时，这个数字代表网络路径本身的开销。第二步是阶梯加压，从每秒 100 条开始，每档跑 3 分钟，逐档翻倍，同时盯着发送失败率、重连次数、broker 的 CPU 和内存。找到失败率开始上升的那一档，往回退一档就是安全容量。第三步是长稳测试，用七成安全容量跑 12 小时以上，观察内存是否稳定、有没有缓慢增长。

压测过程中要重点看几个 broker 侧指标：当前连接数、每秒入站消息数、每秒出站消息数、消息队列深度。这些指标 Mosquitto 通过$SYS/broker/...系列主题暴露出来，直接订阅$SYS/broker/#就能看到，非常方便。记住前面提过的那条规则，$SYS开头的主题不会被#通配符匹配到，得显式写$SYS/#。

容量估算上，我给的经验量级是这样的：在一般的服务器配置上，QoS 0 的消息吞吐能做到每秒几万条，QoS 1 会明显下降，QoS 2 再降一个档次。这个量级的波动取决于消息大小、订阅者数量、是否落盘。订阅者数量对性能的影响是超线性的，因为一条消息要扇出给 N 个订阅者就产生 N 次发送。做容量规划的时候，先算清楚一个主题平均有多少订阅者，这个数字比消息条数更能决定 broker 的压力。

网络带宽也要算。一条 200 字节的消息带 MQTT 协议头算 230 字节左右，每秒一万条就是 2.3 MB/s，接近 20 Mbps。如果一条消息有 20 个订阅者，出站流量还要乘 20。我见过有人只算了入站流量就选了带宽，上线之后出站把链路打满。

6.3 这套封装思路在其他语言里怎么复用

这套设计里真正有价值的部分不是 C 代码，而是接口语义和分层方式。换到别的语言，核心的几件事是一样的，只是实现手段不同。

在 Java 或 Android 上，Paho 和 HiveMQ 都提供了自动重连，但重连后的重订阅、状态标志的时序、回调到业务线程的切换、主题到处理器的映射，这些还是得自己包一层。Android 上还多一个问题：网络切换（Wi-Fi 切蜂窝）时 TCP 连接不会立刻断开，而是进入一种「假死」状态，表现为 publish 返回成功但消息发不出去。解决办法是监听系统的网络变化广播，主动触发重连，同时把 keepalive 调短一点加快检测。

在 Python 上，paho-mqtt 的loop_start()会起一个后台线程，和 C 版本的模型几乎一样，回调也是在那个线程里执行的。要特别注意 Python 的 GIL 不保护你的数据结构，订阅字典的读写照样要加锁，或者改用线程安全的容器。异步框架下用 asyncio 版本会更顺，消息处理不会阻塞网络循环。

在网页端，mqtt.js 走的是 WebSocket，broker 侧要额外开一个 WebSocket 监听。网页端的重连是自动的，但要处理的是「页面切到后台之后浏览器会限制定时器」的问题，导致心跳变慢甚至停发，切回前台时连接已经断了。做法是监听页面的可见性变化，切回前台时主动检查连接状态。前端还有个必须注意的点：WebSocket 连出去之后，broker 地址和凭据对客户端是可见的，不要把管理级别的凭据下发到浏览器，应该用后端签发短时效的临时凭据。

不管在哪个语言里，下面这套接口语义我都建议保留：创建时传配置、启动后自动维持连接、订阅一次永久生效、发布返回明确的三态结果（成功、未连接、失败）、关闭时保证资源回收完成。这套语义一旦定下来，上层业务代码在不同平台之间迁移的成本会低很多，这也是当初写这个封装最直接的收益。

我个人在实际项目中的体会是，MQTT 客户端这块代码写完第一版只要一天，让它真正扛住弱网和各种边界情况，得花上几个月边跑边改。所以别指望一次写完美，先把日志和计数器埋好，让问题自己能浮出来，比提前想象各种异常情况要高效得多。另外一个小技巧，开发阶段在包里加一个「故障注入」开关，能人为触发断连、丢包、延时，比等真实网络出问题再排查快十倍，这个开关上线时记得关掉就行。