AtomMQTT Broker — 用 Rust 实现的轻量级高性能 MQTT 消息代理

特立独行的猫a

2747人浏览 · 2026-09-10 22:16:43

特立独行的猫a · 2026-09-10 22:16:43 发布

一句话介绍：AtomMQTT Broker 是一个纯 Rust 实现的 MQTT 3.1.1 / 5.0 双协议消息代理，单二进制文件即可部署，内置 Web 管理仪表盘、REST API、WebSocket 订阅、SQLite 持久化与 ACL 访问控制，面向物联网与轻量级微服务场景，开箱即用。

AtomGit 项目地址：https://atomgit.com/qq8864/atomMqtt
旋武社区：https://xuanwu.openatom.org（欢迎在旋武社区参与讨论、分享实践、共建开源）
博客介绍：AtomMQTT — 使用 Rust 实现的轻量级高性能 MQTT 服务器

一、项目介绍

AtomMQTT Broker 是一个面向生产环境的轻量级 MQTT Broker。与 Mosquitto 等成熟 Broker 相比，它用现代 Rust + Tokio 异步运行时重新实现了 MQTT 核心链路，带来以下直接收益：

内存安全：Rust 所有权系统从语言层面杜绝了野指针、数据竞争与缓冲区溢出；
零外部依赖部署：前端静态文件在编译期嵌入二进制，单个可执行文件跑起来即是完整的 Broker + Web 管理台；
透明可审计：代码量克制、模块边界清晰，协议到持久化的全链路都可读、可测、可改。

1.1 核心特性

特性	说明
双协议支持	同时支持 MQTT 3.1.1（v311）与 MQTT 5.0（含属性机制）
QoS 0/1/2	完整的服务质量级别支持
主题订阅树	基于 Trie（前缀树）的高效主题匹配，支持 `+` / `#` 通配符
保留消息	Retained Message 存储与分发
遗嘱消息	Will Message，客户端异常断开时自动发布
离线消息队列	`clean_session=false` 的客户端离线期间消息暂存，重连后补发
会话过期清理	可配置 `session_expiry_interval`，后台自动清理过期会话与延迟发布遗嘱
SQLite 持久化	会话、订阅、保留消息、遗嘱消息自动持久化，重启自动恢复
异步批量写入	持久化走独立后台线程，50 条事件或 100ms 触发一次批量事务，主流程零等待
Web 管理界面	内置 Actix-Web 仪表盘：指标监控、客户端管理、订阅查看、消息发布/订阅
REST API	完整的 HTTP API，方便脚本与自动化集成
WebSocket 订阅	浏览器实时订阅 MQTT 主题；另提供原生 WebSocket-MQTT 二进制桥接
认证与 ACL	文件密码认证（支持 argon2 哈希）+ 基于文件的 Topic 级 ACL
TLS 支持	MQTT TCP 与 Web 管理界面均可启用 TLS（rustls）
CLI 客户端	内置 `mqtt-client` 工具：发布 / 订阅 / 交互式 Shell
桌面调试客户端	附 Tauri 桌面版 MQTT 调试工具（`tools/tauri-mqtt-client`）
性能指标	连接数、消息数、字节数、包数等内置计数器，Web 端 2 秒级实时刷新

1.2 独特价值：为什么选择 AtomMQTT

单文件部署，运维成本趋近于零
cargo build --release 产出一个自包含二进制，前端、配置模板全部内置，扔到任何 Linux / macOS / Windows 机器即可运行，不需要 Nginx、不需要 Node 运行时、不需要数据库服务。
热路径纯内存，冷路径异步落盘
消息路由、订阅匹配全部在内存无锁结构中完成（DashMap + Trie 订阅树）；持久化通过 mpsc 通道交给独立后台线程批量写 SQLite。业务链路不被 I/O 拖慢，数据又不丢失。
双协议 + 全 QoS，一套引擎覆盖演进路线
同时实现 3.1.1 与 5.0，存量设备与新一代设备可接入同一个 Broker，升级无迁移成本。
管理面与数据面一体化
传统上「Broker + 监控面板 + 消息调试工具」是三样东西，AtomMQTT 把三者做进了同一个二进制：Web 仪表盘、REST API、WebSocket 调试、CLI 客户端一应俱全。
安全合规默认开启
文件认证支持 argon2 口令哈希；Topic 级 ACL 默认拒绝；默认弱密码启动时显式告警；MQTT 与 Web 均支持 TLS。

二、架构设计

2.1 四层架构

项目采用 Cargo Workspace 多 Crate 结构，按「协议 → 引擎 → 展示 → 客户端」分层：
在这里插入图片描述

2.2 核心设计模式

① BrokerState 全局共享状态

所有连接处理器通过 Arc<BrokerState> 共享同一份状态，关键并发结构各司其职：

字段	类型	用途
`sessions`	`DashMap<String, SessionState>`	客户端会话（无锁读）
`subscriptions`	`Mutex<SubscriptionTree>`	主题订阅树
`retained`	`DashMap<String, RetainedMessage>`	保留消息
`wills`	`DashMap<String, WillMessage>`	遗嘱消息
`connections`	`DashMap<String, UnboundedSender<Vec<u8>>>`	TCP 订阅者投递通道
`web_subscribers`	`DashMap<String, UnboundedSender<String>>`	WebSocket 订阅者通道
`metrics`	`Mutex<BrokerMetrics>`	性能计数器

② 后台单线程路由器

Publisher → mpsc::unbounded_channel → Background Router Loop → 各订阅者投递

所有 PUBLISH 消息统一进入后台路由器串行处理，避免并发投递的竞争与乱序；TCP 订阅者收到 V311 二进制 PUBLISH 包，WebSocket 订阅者收到 JSON 字符串，一次路由、双格式分发。

③ Trie 订阅树

TopicNode {
    children: Vec<(String, TopicNode)>,
    subscriptions: Vec<Subscription>,
}

子节点名为 # 表示多级通配符、+ 表示单级通配符。lookup(topic) 沿「精确匹配 / + / #」三条路径收集订阅者，匹配复杂度与主题层级成正比，远优于全量扫描。

④ 异步批量持久化

内存操作 → send(PersistEvent) → mpsc channel → bg_writer（50 条 或 100ms 批量事务）

持久化事件通过通道发送到独立后台任务，事务批量提交 SQLite（WAL 模式），Broker 关闭时 flush 全部待处理事件。发布/订阅主链路完全不感知 I/O。

三、新手使用教程（5 分钟上手）

跟着下面的步骤走，你就能跑起自己的 MQTT Broker 并完成第一条消息的收发。

步骤 0：环境准备

Rust 1.70+：推荐用 rustup 安装
操作系统：Windows / Linux / macOS 均可

# 安装 rustup（已有 Rust 可跳过）
curl --proto '=https' --tlsv1.2 -sSf https://sh.rustup.rs | sh
source $HOME/.cargo/env

步骤 1：克隆项目

git clone https://atomgit.com/qq8864/atomMqtt.git
cd atomMqtt

步骤 2：构建

# 构建全部 crate（首次构建会编译依赖，耐心等待）
cargo build --release

# 或只构建 Web Broker（含前端仪表盘，日常使用推荐）
cargo build --release -p mqtt-web -p mqtt-client

构建产物位于 target/release/：

二进制	作用
`atom-mqtt`	MQTT Broker + Web 管理界面（主程序）
`mqtt-client`	CLI 测试客户端（发布 / 订阅 / Shell）

步骤 3：启动 Broker

./target/release/atom-mqtt
# 开发阶段也可用 cargo 直接跑
# cargo run -p mqtt-web

首次启动会在当前目录自动生成 config.toml（默认全功能配置）。启动后：

MQTT TCP 监听：tcp://0.0.0.0:1883
Web 管理界面：http://localhost:8081（默认管理员账号 admin / admin）
数据库文件 broker.db 自动创建（WAL 模式）

步骤 4：CLI 客户端收发第一条消息

开两个终端（Broker 已在运行）：

# 终端 1：订阅主题（支持 + / # 通配符）
./target/release/mqtt-client sub 127.0.0.1:1883 "test/#" --client-id sub1

# 终端 2：发布消息
./target/release/mqtt-client pub 127.0.0.1:1883 "test/hello" "Hello AtomMQTT!" --client-id pub1 --qos 1

回到终端 1，你会看到实时打印：

[pub1] test/hello => Hello AtomMQTT!

还可以用交互式 Shell，随时 pub / sub / unsub / ping / quit：

./target/release/mqtt-client shell 127.0.0.1:1883 --client-id my-shell

步骤 5：打开 Web 管理界面

浏览器访问 http://localhost:8081，用 admin / admin 登录（生产环境请务必在 config.toml 的 [web_auth] 段修改），即可看到实时仪表盘：
在这里插入图片描述

页面	功能
📊 仪表盘	在线客户端、活跃订阅、消息统计、网络流量，2 秒自动刷新
👥 客户端	查看在线客户端详情、手动断开连接
📋 订阅	全部活跃订阅（Client ID / 主题过滤器 / QoS）
💾 保留消息	查看与删除保留消息
📤 发布消息	通过 HTTP API 发布消息到任意主题
📡 订阅消息	通过 WebSocket 实时接收订阅的消息
ℹ️ 服务器信息	Broker 配置与运行状态

💡 小技巧：在「订阅消息」页订阅 test/#，再用上面的 CLI 发布一条消息，浏览器里能立刻看到消息流——无需写任何代码。

步骤 6（可选）：桌面 GUI 客户端

仓库内 tools/tauri-mqtt-client 附带一个 Tauri 桌面 MQTT 调试工具（Windows/macOS/Linux），可图形化地连接 Broker、发布/订阅消息：

在这里插入图片描述

四、REST API 速查

所有 /api/ 路由受 HTTP Basic Auth 保护（POST /api/login 免认证）。

4.1 REST 端点

方法	路径	说明
`POST`	`/api/login`	登录（JSON `{username, password}`）
`GET`	`/api/metrics`	Broker 指标快照
`GET`	`/api/broker/info`	配置与版本信息
`GET`	`/api/clients`	在线客户端列表
`GET`	`/api/clients/{client_id}`	单个客户端详情
`GET`	`/api/subscriptions`	全部活跃订阅
`GET`	`/api/retained`	全部保留消息
`DELETE`	`/api/retained/{topic}`	删除保留消息
`POST`	`/api/publish`	发布消息到主题
`POST`	`/api/clients/{client_id}/disconnect`	断开指定客户端

4.2 WebSocket 端点

路径	协议	说明
`ws://host:8081/ws/subscribe`	JSON	浏览器实时订阅 MQTT 主题
`ws://host:8081/mqtt`	二进制 MQTT 包	原生 WebSocket-MQTT 桥接

JSON 命令示例：

{"type": "subscribe", "topic_filter": "test/#", "qos": 1}
{"type": "unsubscribe", "topic_filter": "test/#"}
{"type": "ping"}

收到的消息：

{
  "type": "publish",
  "topic": "test/hello",
  "payload": "Hello MQTT!",
  "qos": 1,
  "source_client": "pub1",
  "timestamp": "2026-09-10T10:30:00+08:00"
}

4.3 脚本集成示例

# 发布一条消息
curl -u admin:admin -X POST http://localhost:8081/api/publish \
  -H "Content-Type: application/json" \
  -d '{"topic":"demo/temp","payload":"25.6","qos":1}'

# 拉取指标
curl -u admin:admin http://localhost:8081/api/metrics

五、配置说明（config.toml）

首次启动自动生成，主要段落：

[tcp]
host = "0.0.0.0"
port = 1883

# MQTT TLS（可选）
[tcp_tls]
cert_path = "certs/server.crt"
key_path  = "certs/server.key"

[web]
host = "0.0.0.0"
port = 8081

# Web HTTPS（可选）
[web_tls]
cert_path = "certs/web.crt"
key_path  = "certs/web.key"

[broker]
max_packet_size = 10485760      # 10 MB
max_qos = 2                     # 0/1/2
allow_anonymous = false
session_expiry_interval = 3600  # 会话过期秒数，0 = 永不过期

[auth]
method = "file"                 # "none" | "file"
auth_file = "passwd"

[web_auth]
enabled = true
username = "admin"
password = "admin"              # ⚠ 生产环境务必修改

[persistence]
db_path = "broker.db"           # 不配置 = 关闭持久化

[acl]
method = "file"                 # "none" | "file"
acl_file = "acl.conf"

5.1 认证

passwd 文件格式为 用户名:口令，一行一个用户。口令支持两种形式：

明文（兼容旧文件）：admin:admin123
argon2 哈希（推荐）：运行 cargo run -p mqtt-broker --example gen_hash 生成 PHC 格式哈希后填入

5.2 ACL 规则

acl.conf 每行一条规则，按顺序匹配、首条命中生效、默认拒绝：

user <用户名> topic <publish|subscribe|readwrite> <主题过滤器>

user admin topic readwrite #
user testuser topic publish test/#

5.3 持久化数据

数据	表	恢复时机
会话	`sessions`	启动时
订阅	`subscriptions`	启动时
保留消息	`retained_messages`	启动时
遗嘱消息	`will_messages`	启动时

六、测试与开发

6.1 单元测试

cargo test                 # 全部
cargo test -p mqtt-broker  # 仅 Broker 引擎（订阅树 / ACL / 持久化等）
cargo test -p mqtt-core    # 仅协议层（编解码 / 属性）

6.2 集成测试（Python + paho-mqtt）

test/test_mqtt.py 覆盖认证、QoS 0/1/2 发布订阅、+/# 通配符、保留消息、ACL 与离线消息队列：

pip install paho-mqtt
# 确保 Broker 已启动
python test/test_mqtt.py

6.3 调试日志

RUST_LOG=mqtt_broker=debug,mqtt_web=debug ./target/release/atom-mqtt

6.4 本地开发约定

日志统一使用 tracing（info/warn/error/debug）
核心逻辑（ACL / 订阅树 / 编解码）必须有 #[test] 单元测试
新增 REST API → mqtt-web/src/api.rs + main.rs 注册路由
新增持久化动作 → mqtt-broker/src/persistence.rs 的 PersistEvent

七、项目结构速览

atomMqtt/
├── Cargo.toml              # Workspace 根
├── mqtt-core/              # 协议层：v311 / v5 编解码
├── mqtt-broker/            # 引擎层：状态、路由、持久化、认证
├── mqtt-web/               # 展示层：REST + WebSocket + 嵌入前端
├── mqtt-client/            # CLI 客户端
├── tools/tauri-mqtt-client/# 桌面 GUI 调试客户端
├── test/                   # Python 集成测试
├── Doc/                    # 文档（本文档所在目录）
├── Knowledge/              # 知识库（架构决策 / 模式沉淀）
├── config.toml             # Broker 配置
├── acl.conf                # ACL 规则
└── CHANGELOG.md            # 更新日志

更深入的实现细节见 Doc/ 下各篇：架构设计 · 原理与实现 · 消息路由 · 协议支持 · Web API · 客户端实现。

八、许可证

本项目基于 MIT 许可证开源。

九、参与社区

AtomGit 仓库：https://atomgit.com/qq8864/atomMqtt — Issue / PR / 讨论都欢迎
旋武社区：https://xuanwu.openatom.org — 加入旋武社区，与更多开发者一起交流开源实践、分享 IoT 与 Rust 的落地经验
贡献指南：见 CONTRIBUTING.md（分支策略、提交规范、PR 流程）
更新日志：见 CHANGELOG.md

如果你也在做 IoT、边缘计算或 Rust 中间件，欢迎 Star 仓库、提 Issue，或在旋武社区找到我们！

亚马逊云科技技术品牌专区

更多推荐

基于大数据+Hadoop+深度学习的淘宝用户购物可视化与行为预测系统设计与开发

亚马逊云科技技术品牌专区

基于大数据爬虫+Hadoop+深度学习的旅游推荐系统设计与开发

亚马逊云科技技术品牌专区

学Linux选学校：一线城市还是老家？

而二线城市很多机构的老师，自己都没进过大厂，技术停留在五年前的水平，甚至对着PPT照本宣科。而二线机构缺乏企业合作资源，项目案例永远是“搭建博客系统”“配置个LAMP环境”——这些在现在的云计算岗位面试中，连初筛都过不了。在一线城市学习，你接触的同学、老师、企业内推资源，都是二线城市没法比的。而老家机构的学员，最好的结果也就是去本地小公司做网管，月薪五六千，天花板肉眼可见。在IT培训这行干了快十年