【免费下载链接】mosquitto
Eclipse Mosquitto – An open source MQTT broker
项目地址：
https://gitcode.com/gh_mirrors/mo/mosquitto
点击查看 免费下载
本文以 2010 年 5 月 5 日发布的 Mosquitto 0.6 版本发布公告（www/posts/2010/05/version-0-6-released.md）为骨架，结合仓库当前源码逐一还原每个新特性的底层实现。读完本文，你将理解 Mosquitto 的桥接（bridging）机制、客户端工具的输入扩展、单副本消息存储设计与持久化升级路径，以及 max_inflight_messages / max_queued_messages 等队列控制参数的真实作用。
版本背景
0.6 是一个特性发布（new features release），相比前版有大量变化，完整细节收录于 man 手册 与 ChangeLog.txt 的 0.6 – 20100505 条目。发布公告同时提到 Windows 与 Ubuntu 二进制随后跟进，用户可从下载页获取。
该版本变更可划分为两大块：
- 显著的新特性：桥接、mosquitto_sub 多主题订阅、mosquitto_pub 文件与 stdin 输入、单副本消息存储、数据库 v1→v2 自动升级、零长度 retained 消息语义、max_inflight_messages / max_queued_messages；
- 不显眼的修复与增强：clean session 开关、退出时 VACUUM、store_cleanup_interval 与 persistence_file 新配置项、QoS=0 断连存储修复、零长度 payload 连接稳定性等。
桥接（Bridging）：多 Broker 互联的基础支持
0.6 为 Mosquitto 带来了"将多个 MQTT broker 连接在一起"的桥接能力，这是 Mosquitto 迈向分布式消息网络的第一步。发布公告将其列为"substantial changes"之首，并说明这是"basic support"——即基础支持，后续版本在此基础上持续演进。
从当前源码 src/bridge.c 可以反推出桥接在 0.6 奠定的核心骨架：
- 多地址连接与故障转移：broker 通过 addresses[] 数组维护一组候选服务器地址（见 src/bridge.c），连接失败时逐个尝试；配合 round_robin 选项决定轮询顺序（src/bridge.c）。
- 多种启动模式：源码中定义了 bst_lazy、bst_automatic、bst_once、bst_manual 等启动类型（见 src/bridge.c），对应按需（lazy）、自动（automatic）、一次性（once）与手动（manual）重连策略；lazy_reconnect 与 restart_t 控制自动模式下的重连间隔。
- 本地桥接优化：try_private 与 try_private_accepted（src/bridge.c）允许同一主机上的两个 broker 通过本地环回建立更高效的私有连接。
从源码结构看，0.6 的"基本桥接"奠定了 bridge 配置节、多地址轮询、自动重连与本地私有通道这些后续所有桥接功能的基石，今天 Mosquitto 的多 broker 网状部署仍沿用这一设计。
客户端工具增强：多主题订阅与三种输入模式
0.6 同时大幅增强了 mosquitto_sub 与 mosquitto_pub 两个命令行客户端。
mosquitto_sub 支持多主题订阅
mosquitto_sub 现在可以一次订阅多个主题，但受限于一个全局 QoS——所有主题共享同一条命令行上指定的 QoS 等级。这一约束在 client/sub_client.c 的选项解析中体现，帮助用户"用一个进程监控多个主题"，对应发布公告中"useful for monitoring multiple topics"的用例场景。
mosquitto_pub 的三种输入方式
mosquitto_pub 的输入来源从单一的命令行参数扩展为四种：
| -m / –message | 直接以命令行参数作为消息（原有方式） | client/client_shared.c |
| -f / –file | 读取一个文件，把文件内容作为一条消息发送 | client/client_shared.c |
| -s / –stdin-file | 读取 stdin 的全部内容，作为一条消息发送 | client/client_shared.c |
| -l / –stdin-line | 读取 stdin，并把每一行分别作为一条消息发送 | client/client_shared.c |
三种输入模式解决了脚本与自动化场景的常见需求：-f 适合发送日志文件、配置片段等已有文件内容；-s 适合把管道上游的完整输出一次性发布；-l 则把每行包装成独立消息，适合逐条上报设备数据行。这些选项沿用至今，是 mosquitto_pub 最常用的输入扩展。
数据库重构：每消息只存一份
0.6 实现了一个更高效的数据库设计：每条消息在数据库中只保存一份，而不是为每个订阅它的客户端各存一份。这在内存占用上具有数量级意义——对于 100 个订阅者，旧设计需要 100 份拷贝，新设计只需 1 份。
当前源码仍完整保留这一设计的实现，体现在 src/database.c 的引用计数机制中：
- 每个基础消息（struct mosquitto__base_msg）带 ref_count 引用计数；
- db__msg_store_ref_inc() 在新增引用时递增（src/database.c）；
- db__msg_store_ref_dec() 递减并在计数归零时真正释放消息（src/database.c）；
- db__msg_store_compact() 遍历消息存储，清理所有 ref_count < 1 的孤儿消息（src/database.c）；
- 消息还记录 dest_ids / dest_id_count，用于追踪其目标客户端（src/database.c）。
ref_count 清零即释放的语义可以在 src/retain.c 看到巧妙运用：构造临时消息时把 ref_count 预置为 10，确保它在检查/删除过程中"不会被意外释放"。
与之配套，0.6 增加了 store_cleanup_interval 配置项，用于定期清理内部消息存储中已无引用的消息。需要注意的是，该选项在 0.6 引入后并未保留到今天——在当前 src/conf.c 中已搜不到这一独立选项，说明其功能在后继版本中被吸收进其他存储管理机制；本段描述仅针对 0.6 时代的行为。
持久化：DB v1→v2 自动升级、VACUUM 与 persistence_file
0.6 在持久化存储方面有三个动作：
1. 数据库格式 v1→v2 自动升级。 配合"单消息单副本"的新存储模型，持久化数据库格式随之升到 v2。Mosquitto 在启动时检测旧格式并自动迁移，无需用户手工干预。当前仓库仍保留了新旧格式的读取路径：src/persist_read.c 与 src/persist_read_v234.c，后者正是 v2/v3/v4 时代格式的兼容读取层；src/persist_write.c 则负责写出。0.6.1（ChangeLog.txt）还专门修复了 messages 表的 DB 自动升级问题，可见这一路径在早期版本中备受关注。
2. 退出时正确运行 VACUUM。 发布公告明确"mosquitto 现在会在退出时正确地对持久化数据库运行 VACUUM"，回收因消息增删而产生的空洞空间，防止 .db 文件无限膨胀。
3. 新增 persistence_file 配置项。 此前持久化文件名不可配置，而 persistence_location 又要保持与 rsmb 兼容。0.6 引入 persistence_file 允许改变持久化数据库的文件名（见 src/conf.c），从而可以在同一目录下存放多个 Mosquitto 数据库。当前源码中该选项的默认值为 mosquitto.db（src/conf.c），persistence_location 与 persistence_file 拼接得到最终文件路径（src/conf.c），并且 persistence_location 还可以通过环境变量 MOSQUITTO_PERSISTENCE_LOCATION 覆盖（src/conf.c）。
Retained 消息语义：零长度 payload 即删除
0.6 确立了一条至今仍重要的 MQTT 语义：如果收到一个 payload 长度为 0 的 retained 消息，则删除该主题上已保存的 retained 消息。这一规则让"清除 retained 消息"有了标准手段——只需向该主题发布一条空 payload 的 retained 消息。
当前实现位于 src/retain.c 的 retain__store()：
- 当 base_msg->data.payloadlen == 0 时，将 retainhier->retained 置为 NULL，并调用 retain__clean_empty_hierarchy() 回收不再使用的层级节点（src/retain.c）；
- 只有 payloadlen != 0 时才写入新的 retained 消息（src/retain.c）；
- 注释明确写道："payloadlen == 0 意味着该消息将被移除"（src/retain.c）。
与之配套的两条零长度修复同样值得注意：
- 不再因为客户端发送零长度 payload 的 PUBLISH 而断开它（bug #573610）：此前 broker 可能误判为空消息而断连；
- 正常转发零长度消息：空 payload 是合法的 MQTT 消息，broker 应将其视为普通消息转发。
这两条修复与 retained 删除语义共同完善了 0.6 对零长度消息的完整处理闭环。
队列控制：max_inflight_messages 与 max_queued_messages
0.6 在 broker 中落地了两个重要的流量控制特性，它们在今天的 mosquitto.conf 中依然是核心配置：
max_inflight_messages
限制单个客户端连接上"已发送但尚未收到确认"的在途（inflight）消息数量，默认值为 20（src/conf.c）。配置解析时强制要求该值不大于 65535（src/conf.c）。它防止 QoS>0 的流量在慢速客户端上无限堆积，是 broker 端背压控制的基础。
max_queued_messages
限制为某个客户端排队等待发送的消息总数，默认值为 1000（src/conf.c）。在 lib/packet_mosq.c 的发送路径中，当 out_packet_count >= max_queued_messages 时，broker 会停止继续为连接追加出站包，从而防止订阅者处理速度跟不上时内存无限增长。
两个参数配合使用的场景在 mosquitto.conf 的注释中有所说明：既可以从条数（max_queued_messages）也可以从字节数（max_queued_bytes）两个维度限制队列，若两者同时指定则以先触发的为准。配套的客户端侧接口 mosquitto_max_inflight_messages_set()（lib/messages_mosq.c）允许应用程序自行调节自己的 inflight 上限。
其他增强与修复
0.6 还包含若干"不显眼但重要"的变更：
- mosquitto_sub 支持禁用 clean session：订阅客户端可关闭 clean session，配合持久化订阅在重连后继续收到积压消息；同期 man/mqtt.7.xml 手册补充了对 clean session 标志的说明。
- QoS=0 消息不为断连客户端存储（bug #572608）：0.5 中该修复不彻底，0.6 真正落实——QoS=0 本质是"尽力而为"，断连客户端无需为其保留消息，避免内存浪费。
- 对不支持的 rsmb 选项输出警告而非直接退出：提升了配置容错性，broker 可以带着告警继续运行，而不是因一个陌生选项整体崩溃。
- store_cleanup_interval 配置项：如前所述，用于周期性清理内部消息存储，是 0.6 引入后经演进吸收的配置。
结语
从 0.6 的发布公告可以清晰看到 Mosquitto 早期发展的两条主线：一是功能外延——通过桥接让多个 broker 互联成网，通过客户端输入扩展让工具更好用；二是内核加固——单副本消息存储、引用计数、持久化格式升级与队列限流，为大规模部署打下了内存与稳定性基础。今天 src/bridge.c、src/database.c、src/retain.c 中的实现，都能追溯到 0.6 确立的这些设计决策。
赞
【免费下载链接】mosquitto
Eclipse Mosquitto – An open source MQTT broker
项目地址：
https://gitcode.com/gh_mirrors/mo/mosquitto
点击查看 免费下载
相关推荐
终极iOS虚拟定位工具：iFakeLocation完全指南
基于模板匹配与状态机设计的游戏自动化框架实现
创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
