假设你正在做一套农业监测设备。板子能采集温湿度,也能控制一台水泵。现在,你希望在电脑上看到温度,并且点一下按钮,就能让远处的水泵启动。
先别急着背 QoS、Retain、Session。我们从两个最具体的问题开始:设备怎样把数据交给后台?后台怎样把命令送回设备?
本文用同一个设备 d001 贯穿实验与代码。前 3 章完成一次双向通信,第 4、5 章处理可靠性与断网,第 6、7 章落到设备程序和项目交付,第 8 章再看系统扩展。默认先学习 MQTT 3.1.1;涉及 MQTT 5.0 的区别时单独说明。实验使用模拟设备,不连接真实水泵。
第 1 章 从一台传感器出发:物联网通信需要解决什么
先把协议名字放到一边。我们的设备要完成三件事:定期上传温湿度,接收水泵控制命令,让后台知道它是否还在线。协议选型要围绕这些需求展开。
1.1 温度能上传了,服务器怎样让水泵启动?
如果只需要上传温度,一个 HTTP 请求就能解决:
POST /devices/d001/state HTTP/1.1
Host: api.example.com
Content-Type: application/json
Content-Length: 34
{"temperature":26.3,"humidity":61}
设备发请求,服务器保存数据并回复。这个方向很自然。
现在换个方向:用户在后台点击“开泵”,但设备此时没有发送请求,服务器怎样立即把命令送过去?设备可能还藏在路由器或运营商 NAT 后面,服务器也无法直接连接它的私网地址。
一种办法是让设备不断询问:“有没有我的命令?”这就是轮询。假设每 5 秒问一次,没有命令时也要消耗通信资源;命令刚好在一次询问结束后产生,还可能等近 5 秒才被取走。轮询间隔越短,请求越密;间隔越长,控制越不及时。
长轮询、WebSocket 等方案可以改善这个问题。MQTT 则让设备先主动连接服务器,随后双方沿着已经建立的连接收发消息。后台有命令时,可以通过这条连接送到设备,无需等待下一轮询问。
这里要分清一个常见误解:HTTP/1.1 默认支持持久连接,并不是每次请求都必须重新握手。 我们比较的是不同通信模型对这个业务是否方便,不能用“HTTP 一定是短连接”来解释 MQTT 的价值。HTTP/1.1 的连接复用规则可见 RFC 9112 的持久连接说明。
1.2 多个系统都想接收温度,设备要分别发送吗?
最初只有后台保存温度。后来,告警服务也要接收温度,手机页面也要实时展示。
如果让设备分别调用三个接口,那么增加一个消费者,就可能需要修改设备程序。设备还得关心每个服务的地址、连接和失败处理。消费者越多,设备与后台之间的依赖越复杂。
换一种方式:设备只发布“这是 d001 的最新状态”,需要这份数据的服务提前登记自己的兴趣。中间的消息服务器负责把消息转交给它们。
这就是发布/订阅:
反过来,设备也可以订阅控制命令。于是同一个设备既能上报,又能接收控制,角色会随着消息方向变化。
1.3 MQTT 适合承担哪些工作?
MQTT 是一种轻量级的应用层消息协议,常运行在 TCP 之上,并通过 TLS 保护连接。它把发布/订阅、消息交付等级、会话和掉线通知等约定统一起来,让设备与后台不用各自设计一整套消息协议。
“轻量”可以从我们的温度上报来理解:业务数据很少,如果每次都附带大量重复元信息,额外开销就会很显眼。MQTT 使用紧凑的二进制报文格式,适合持续发送较小消息。但完整消息仍包含主题名等内容,不能把“固定头最小 2 字节”理解成“任何 MQTT 消息都只有 2 字节”。
在一个实际项目里,可以这样分工:
| | |
|---|
| | |
| | |
| 后台通过 WebSocket 推送;也可采用 MQTT over WebSocket | |
| | |
WebSocket 提供双向通信通道;订阅关系、消息确认和离线处理仍需要上层协议定义。MQTT 可以运行在 WebSocket 上,所以两者也可以组合。
低功耗同样要看设备怎样工作。常在线设备可以复用连接,减少反复建连的成本;每天只醒一次的电池设备,则可能更适合“唤醒、建连、上报、断开、休眠”。保持一条连接并不自动等于最低功耗。
到这里,我们选择 MQTT 的理由已经明确:同一条连接承担上报与下控,设备和多个后台消费者通过发布/订阅协作。 接下来看看一条消息究竟怎样走。
第 2 章 跟着一条消息,认识 MQTT 的通信过程
假设设备刚刚读到温度 26.3°C。我们跟着这份数据走到杭州扬森电力科技有限故事后台,再让一条开泵命令沿反方向回到设备,逐步认识连接、主题和报文。
2.1 设备把消息交给谁,后台又从哪里接收?
设备先连接一个 MQTT 消息服务器,它叫 Broker。后台服务也连接同一个 Broker。二者都是 MQTT Client,客户端可以是 MCU、Linux 程序、手机应用、网关或测试工具。
设备发送一条消息时,是发布者;后台接收这条消息时,是订阅者。后台发送控制命令时,双方的角色交换。因此,“发布者只会发,订阅者只会收”描述的是一次通信中的角色,不能据此把一个程序永久分成两类。
我们的第一条消息可以写成:
发布者:设备 d001
Topic:farm/p1/d001/state
Payload:{"temperature":26.3,"humidity":61,"pump":false}
其中 p1 是本例的产品标识,d001 是设备标识,state 表示状态。
后台提前订阅 farm/p1/d001/state。设备发布后,Broker 查找匹配的订阅关系,再把消息转给后台。如果告警服务也订阅了这个主题,它也能收到一份。
Broker 不会自动把消息广播给所有已连接客户端。 已连接只代表建立了通信关系,订阅匹配且权限允许,才有机会收到对应消息。没有订阅者时,普通发布消息也不会自动变成供未来消费者读取的历史记录。
Broker 可以承担鉴权、会话、路由和消息交付管理,但历史温度统计仍应由业务服务写入数据库。不要把 Broker 的缓存机制当作时序数据库使用。
2.2 Broker 怎么知道消息该转给谁?
靠 Topic,也就是主题名。它是消息的路由名称,/ 分隔层级;层级怎样命名由项目约定,并不要求对应磁盘目录或设备 IP。
本例先约定两个方向:
设备上报:farm/p1/d001/state
后台下控:farm/p1/d001/command
设备订阅自己的 command,后台订阅设备的 state。这里设备 ID 只是主题中的一个字段,Broker 不会因为看到 d001 就自动定位设备连接,它仍然按照订阅关系转发。
当后台需要接收同一产品下所有设备的状态,可以订阅:
farm/p1/+/state
+ 匹配完整的一层,所以它能匹配 farm/p1/d001/state 和 farm/p1/d002/state,不能匹配 farm/p1/d001/sensor/state。
如果实验时想观察 d001 下的所有消息,可以订阅:
farm/p1/d001/#
# 匹配零层或多层,必须独占最后一层;例如 farm/p1/d001/# 也能匹配 farm/p1/d001。通配符用于订阅过滤器,不能放进发布的主题名。 普通主题名区分大小写,state、State 和末尾多一个 / 都可能是不同主题。以 $ 开头的系统主题还有特殊匹配规则,入门实验先使用上述业务主题。
Topic 解决“消息去哪里”,Payload 解决“消息里装什么”。本例用 JSON,方便直接观察;以后换成二进制、CBOR 或 Protobuf,发布/订阅模型并不会因此改变。MQTT 本身不理解 temperature 或 pump 的业务含义。
2.3 从连接成功到收到命令,中间发生了什么?
设备并不是调用一次 connect(),就立刻具备全部收发能力。一次典型启动过程是:
| | |
|---|
| | |
| 发送 CONNECT,Broker 返回 CONNACK | |
| 发送 SUBSCRIBE,Broker 返回 SUBACK | |
| | |
| Broker 向设备发送匹配订阅的 PUBLISH | 从回调中取出主题和 Payload,再交给业务处理 |
CONNECT 会携带 Client ID、可选认证信息、心跳和会话设置等。Client ID 用来标识客户端会话,和 Topic 中的设备 ID 是两件事;本例可以把设备客户端命名为 device-d001,后台命名为 backend-demo。
同一 Broker 上两个活跃客户端使用相同 Client ID,后连接者会使旧连接被断开。这经常表现为“明明网络没问题,两台设备却轮流掉线”。设备 Client ID 应稳定且唯一,临时测试客户端也要使用不同 ID。
SUBACK 可能表示成功,也可能表示拒绝订阅。调用订阅 API 成功,往往只是请求已提交;程序还要观察异步结果。类似地,发布 API 返回消息编号,也不必然表示后台业务已经处理。
这些操作最终会编码成 MQTT 控制报文。报文由固定头、按类型变化的可变头和可选 Payload 组成。以 MQTT 3.1.1 的 PUBLISH 为例,主题名放在可变头中,QoS 1/2 还包含 Packet Identifier,业务数据放在 Payload 中。初学阶段先会看连接、订阅与发布日志,报文编解码交给 SDK。
连接、主题与订阅的协议规则可进一步对照 MQTT 3.1.1 标准第 3、4 章。下一章先用测试工具把这条链路跑通。
第 3 章 先动手:让两个客户端完成一次双向通信
现在用一台电脑代替传感器和后台。只要先验证消息能双向传递,后面换成真实设备时,就能区分“协议或主题问题”和“硬件程序问题”。
3.1 准备一个 Broker 和两个测试客户端
本章采用 Mosquitto 2.x + MQTTX Desktop。先安装 Mosquitto 和 MQTTX;Windows 可使用官方安装包,Linux 可以使用发行版的软件包。下面的启动命令要求 mosquitto 已加入 PATH;Windows 也可以在安装目录打开终端,执行 ./mosquitto.exe。
在工作目录新建 mosquitto-lab.conf:
listener 1883 127.0.0.1
allow_anonymous true
persistence false
然后启动:
mosquitto -c mosquitto-lab.conf -v
这个配置只允许本机访问,适合两个模拟客户端的第一轮实验。终端保持打开;若提示端口占用,先确认是否已有 Mosquitto 服务或其他 Broker 在运行,避免误连另一个实例。persistence false 表示此实验不把运行状态持久化到磁盘,重启 Broker 会清空保留消息和会话等内存状态。
打开 MQTTX,新建以下两条连接:
| | |
|---|
| | |
| mqtt://127.0.0.1 | mqtt://127.0.0.1 |
| 1883 | 1883 |
| device-d001 | backend-demo |
| 3.1.1 | 3.1.1 |
| | |
| | |
| 60 | 60 |
| | |
两条连接均应显示成功,Broker 终端也应出现两个不同的客户端标识。此时只能说明“连上了”,还没有任何业务订阅。
配置语义与客户端操作可分别查阅 Mosquitto 配置手册和 MQTTX 入门说明。
3.2 模拟设备上报温度、后台下发控制
先操作 backend-demo,添加订阅 farm/p1/d001/state,订阅 QoS 暂设为 0。再操作 device-demo,向这个主题发布以下 Payload,发布 QoS 设为 0,Retain 关闭:
{"temperature":26.3,"humidity":61,"pump":false}
预期结果是:后台连接的接收列表出现这条消息。注意区分 MQTTX 的“发送记录”和“接收记录”,只在设备窗口看到自己发送成功,还不能证明后台收到。
接着反向操作。在 device-demo 上订阅 farm/p1/d001/command,然后让 backend-demo 发布:
{"requestId":"r001","type":"set_pump","params":{"on":true},"version":1}
预期设备连接收到开泵命令。此时没有真实硬件,也没有控制程序,所以水泵不会自动启动。为了模拟执行结束,再让设备主动向 farm/p1/d001/state 发布:
{"temperature":26.3,"humidity":61,"pump":true}
后台应该看到新的状态。这一次实验实际上走完了三个不同动作:下发命令、设备接收、设备报告结果状态。 MQTTX 只是人工代替程序完成了这些动作,Broker 并不负责执行 set_pump。
再建立一个 Client ID 为 alarm-demo 的客户端,订阅 farm/p1/d001/state。设备再次上报时,后台和告警端应各收到一份。这样就验证了“设备发送一次,多个订阅者接收”。
3.3 故意改错参数,看看消息为什么收不到
把实验结果先记下来,再按下面顺序制造几个小故障。每次只改一项,观察后恢复,避免多个问题混在一起。
| | |
|---|
| | |
| | 不会补回刚才那条消息;保持 Retain 关闭、Clean Session 开启 |
| 后台改订阅 farm/p1/+/state,再发布 d001、d002 的状态 | |
| 把告警端 Client ID 改成 device-d001 后连接 | |
实验结束后断开告警端,把设备和后台恢复为第 3.1 节的配置。
如果这几次变化都能解释清楚,你已经理解了 MQTT 的基本模型。接下来再问一个更实际的问题:网络出了问题,消息会不会丢、会不会重复?
第 4 章 消息发出以后:怎样判断它有没有可靠送达
温度每分钟上报一次,少一条可能很快就被下一条替代。但“给水泵增加 10 分钟运行时间”如果重复执行,后果就不同了。理解可靠性,要先区分消息交付和业务动作。
4.1 温度偶尔少一条,控制命令也能这样处理吗?
MQTT 提供三种 QoS,决定一段发送方与接收方之间怎样完成协议交付:
| | | |
|---|
| | | |
| | | |
| PUBLISH → PUBREC → PUBREL → PUBCOMP | | |
为什么 QoS 1 可能重复?假设接收方已经收到 PUBLISH 并回复 PUBACK,但发送方还没确认结果就断线。恢复相关会话后,发送方可能重发这条尚未确认的消息。接收方已经处理过一次,现在又收到一次。
这不意味着 TCP 不可靠。TCP 负责连接内的有序字节传输;连接中断时,应用仍可能不知道对端究竟处理到了哪一步。
重传时可以见到 DUP 标志,但业务不能拿它当可靠的去重键。DUP 描述协议报文是否属于重发,无法识别后台重新构造并发送的同一业务请求;Packet Identifier 也会复用,不是全局业务 ID。
另外,一次“设备到后台”的传递包含两段:
- 设备 → Broker,按设备发布的 QoS 交付。
- Broker → 后台,按该订阅获准的最大 QoS 和原消息 QoS 决定交付等级。
通常,转发 QoS 不高于两者的较低值。设备以 QoS 1 发布,后台只订阅 QoS 0,不能据此认为后台收到的消息也有 QoS 1 保障。设备收到 Broker 的 PUBACK,也不能证明后台已经收到。
可以在第 3 章实验中同时把发布和订阅设为 QoS 1,观察接收记录及调试日志。普通网络环境未出现重复是正常的;不要把“一次实验没有重现”理解成协议保证不会重复,也不要把重传简单描述成“在线时无限定时重发”,具体行为还受协议版本和 SDK 策略影响。
4.2 收到了确认,为什么还不能认定水泵已经启动?
设想后台收到一条协议确认,但设备的继电器驱动失败,或者水泵处于保护状态。消息送到了,动作却没有完成。
因此需要分清三层结果:
- 发布请求提交成功:SDK 接受了发送请求,或将其放进本地队列。
- 协议交付完成:例如 QoS 1 收到 PUBACK,确认的是对端 MQTT 协议接收。
- 业务执行结果明确:设备报告命令执行成功或失败,并上报实际状态。
本例让设备在执行结束后,向 farm/p1/d001/command/ack 发布业务应答:
{"requestId":"r001","status":"completed","code":0}
再发布状态:
{"temperature":26.3,"humidity":61,"pump":true,"stateVersion":1024}
这里约定 completed 是执行完成;如果业务需要先排队,可以先回复 accepted,稍后再回复 completed 或 failed。后台应把“已接收”和“已完成”分开展示,并设置业务超时,不能一直等一个可能永远不到的结果。
requestId 用来把命令与应答对应起来,也用来识别重复请求。设备遇到已经处理过的 ID,应返回记录的结果,避免再次产生副作用。同一个 ID 却带了不同参数时,应视为冲突,不能悄悄当成新命令。
“设置水泵为开启状态”天然比“切换水泵状态”更容易实现幂等:连续设置两次开启,最终仍是开启;连续切换两次,反而回到了关闭。但有副作用的操作仍需要请求记录、执行状态及异常恢复设计。若要求设备重启后也能去重,记录就不能只存在 RAM 中;还要考虑掉电发生在“动作完成、结果尚未持久化”之间的情况。
QoS 2 同样不会替你把 GPIO、Flash、数据库与业务事务绑定起来。选更高 QoS,不能替代业务执行结果、去重和故障恢复设计。 QoS 的协议交付规则见 MQTT 3.1.1 标准第 4.3 节。
4.3 后台刚上线,怎样立即看到最新温度?
现在温度上报一切正常,但手机刚打开页面,距离设备下一次采样还有一分钟。用户是否必须等一分钟才能看到状态?
可以让设备发布状态时设置 Retain。Broker 为这个 Topic 保存最后一条保留消息,新订阅者订阅匹配主题后,就能获得已保存的状态。它保存的是“主题最新快照”,不是全部历史温度。
回到 MQTTX,做一次可重复的实验:
- 设备向
farm/p1/d001/state 发布温度 26.3,Retain 开启。 - 后台重新订阅,应立即收到 26.3,即使设备此时没有再次发布。
- 设备以 Retain 开启的方式发布温度 27.0;后台取消后重订阅,应获得 27.0。
- 设备向同一主题发送零字节 Payload,Retain 开启。再新建订阅,应不再获得该主题的保留快照。
最后一步是空字节串,不是 {}、null 或两个引号组成的字符串。已经在线的订阅者可能收到这条空消息,接收程序也需要能处理它。若订阅了通配符,新订阅者可能收到多个匹配主题各自保存的保留消息,并非总共一条。
有一个很隐蔽的细节:向同一 Topic 发布 retain=0 的新消息,会送给当前匹配的订阅者,但不会替换已保存的保留消息。设备平时都应按统一约定发布状态,否则刚上线的后台可能拿到旧快照。
Retain 很适合状态快照,不适合本例的瞬时控制命令。假设把“开泵”命令保留了,新设备重新订阅命令主题时,可能又拿到旧命令。更复杂的“期望状态”控制可以采用保留配置,但必须另外设计版本、有效期和状态协调,不能与一次性命令混用。
最后,保留的最新值也可能已经过时。给采样状态加时间戳或版本,页面同时显示设备在线状态与数据时间。MQTT 3.1.1 没有标准的消息过期属性;MQTT 5.0 提供 Message Expiry Interval,但设备端仍要考虑业务有效期。
第 5 章 设备断网以后:怎样发现掉线并恢复工作
一次消息可靠送达,还不足以让设备长期运行。野外网络会切换、路由器会重启、设备会断电。这一章围绕一次断网展开:先发现离线,再恢复通信,最后恢复业务状态。
5.1 设备突然断电,后台怎么知道?
如果设备能正常关机,可以先报告离线,再主动断开连接。但突然断电时,它连“我要下线了”都来不及发送。
MQTT 用 Keep Alive 帮助检测连接是否失去响应。以 MQTT 3.1.1 为例,客户端需要在约定间隔内发送 MQTT 控制报文;没有业务报文时,就发送 PINGREQ,Broker 回复 PINGRESP。它不是无论有没有业务都机械地每分钟多发一个心跳。
当 CONNECT 报文中的 Keep Alive 非零,Broker 超过其 1.5 倍时间未收到客户端的控制报文,需要断开该连接。例如设置 60 秒,对应的是约 90 秒的协议检测窗口,不是“设备断电后一定 60 秒内显示离线”。底层发现链路断开时,也可能更早检测到。报文字段设为 0 会关闭这项协议超时检测,不代表永远不会断线。注意 SDK 的配置值未必直接等于报文字段,例如第 6 章使用的 ESP-MQTT 中,配置 keepalive=0 表示采用默认间隔,关闭保活需要使用专门的配置项。
发现异常后,谁来通知后台?可以在 CONNECT 中注册一条遗嘱消息 Last Will:
| |
|---|
| farm/p1/d001/online |
| {"online":false} |
| 1 |
| |
设备正常建立业务通信后,主动向同一主题发布 {"online":true},同样开启 Retain。异常失联时,Broker 代它发布离线遗嘱;后来才上线的后台也能通过保留消息读到这份在线状态。
正常下线则需要设备自己发布 {"online":false},确认发送进展后再发送正常 DISCONNECT,因为正常 DISCONNECT 不会触发遗嘱。MQTT 5.0 还支持遗嘱延迟,并有请求发送遗嘱的断开原因码;不要把 3.1.1 的正常断开规则不加区分地套用到所有情况。
在 MQTTX 中断开模拟设备,配置遗嘱后重新连接。为了在强制结束 MQTTX 后仍能观察消息,保持 Broker 运行,并在独立终端启动观察端(同样要求 mosquitto_sub 已加入 PATH):
mosquitto_sub -h 127.0.0.1 -p 1883 -i will-observer -t farm/p1/d001/online -q 1 -v
设备先手工发布在线消息,QoS 设为 1、Retain 开启,确认观察端收到后,再用任务管理器等方式强制结束 MQTTX 进程。不要点击正常断开或退出按钮,也不要启用快速自动重连。MQTTX 中的模拟后台也会一并断开,因此这里用独立终端观察:Broker 判定设备失联后,应收到离线消息。重新打开 MQTTX、连接设备并发布在线状态,观察端应看到恢复。强制结束进程时 TCP 关闭可能较快被发现,不一定要等完整心跳窗口;只断开电脑外网不会中断本实验的 127.0.0.1 回环连接。
TCP Keepalive 也能协助检测连接问题,但它与 MQTT Keep Alive 属于不同层,配置与检测节奏也不同。至于水泵控制任务是否卡死,还需要业务健康监测;网络线程仍能回复心跳,不代表硬件控制一定正常。
5.2 重新连上后,还需要重新订阅吗?
先看默认的干净连接:断线后使用全新会话连接,之前的订阅关系没有了,设备就需要重新订阅命令主题。只恢复 TCP 和 MQTT 连接,却忘了恢复订阅,会出现“显示已连接,但再也收不到命令”。
如果希望断线后保留订阅与相应待投递消息,可以使用持久会话。两种协议版本的配置方式不同:
| | |
|---|
| 稳定的 Client ID,Clean Session=0 | 成功 CONNACK 中的 Session Present |
| 稳定的 Client ID,使用适当的 Session Expiry Interval;恢复时 Clean Start=0 | 成功 CONNACK 中的 Session Present |
MQTT 5.0 的 Clean Start 决定连接时是否丢弃旧会话,Session Expiry Interval 决定断开后保留多久。只设置 Clean Start 为 0,却没有有效的会话保留时间,并不等于已经实现持久会话。相关属性见 MQTT 5.0 标准的 CONNECT 与会话说明。
会话可以保存订阅以及协议中未完成的交付状态。对已有订阅匹配的 QoS 1/2 消息,Broker 可以在客户端离线期间按会话规则保存,等待恢复后交付;QoS 0 不应作为可靠离线补投的依据。会话过期、队列上限、消息过期和 Broker 持久化设置仍会影响最终结果。
用 MQTTX 验证时,先让模拟设备以 Clean Session=0、Client ID device-d001 建立连接,订阅一个专用于实验的主题 farm/p1/d001/test/offline,订阅 QoS 设为 1。断开设备后,后台向该主题以 QoS 1、Retain 关闭发布一条消息;设备用相同 ID 和会话设置重连,应能观察到补投。再换成新 Client ID,便无法继承原设备的会话。实验后清理该会话,避免影响后续控制测试。
到这里可以把三个容易混淆的机制放在一起:
另外,Broker 帮设备保存离线待收消息,不等于设备能在断网时把温度发出去。设备离线期间产生的采样,需要设备自己的 RAM/Flash 缓存与补发策略。SDK 的内存发送队列通常也不能保证掉电不丢。
5.3 网络恢复了,怎样避免旧命令和旧状态造成混乱?
假设设备离线了半小时,期间后台发过“开泵”,后来用户又取消了灌溉。此时补投旧命令,虽然符合消息交付规则,却可能不符合当前业务意图。
因此,重连要同时处理通信恢复和业务恢复。建议按下面的顺序组织:
- 等待 Wi-Fi、蜂窝网络等底层网络可用,再建立 MQTT 连接。
- 收到成功 CONNACK 后,检查会话是否恢复。未恢复时重新订阅;已有会话也要核对必要订阅与当前业务是否一致,并检查 SUBACK 结果。
- 恢复期间到达的消息先进入受控队列,按业务规则检查有效期、请求 ID 和顺序,不能假设它们一定等到初始化结束才到达。
- 读取设备真实状态,发布当前快照和在线状态,再进入正常运行;补投命令执行后继续发布新的应答与状态。
对于瞬时控制,本例默认不依赖离线命令积压:后台发现设备离线时提示无法执行,恢复后让用户重新确认。确实要排队时,应在命令中加入有效期、设备配置版本或其他前置条件。MQTT 5.0 的消息过期可以减少过期消息投递,但设备仍需验证已经收到或已经入队的命令。
状态快照可以带 stateVersion。如果它用于跨重启比较,就要持久化递增序号,或者使用“启动代号 + 本次启动内序号”等可解释的规则;简单地每次启动归零,会让后台误以为新状态比旧状态更旧。时间戳也要明确时钟来源,未校准的设备时间不能直接拿来判断控制指令是否过期。
连接失败后也不要每毫秒调用一次 connect()。可以采用有上限的指数退避与随机抖动:
upper = min(maxDelay, base × 2^attempt)
delay = random(upper / 2, upper)
例如从 1 秒量级开始、上限设为 60 秒,实际值按产品恢复速度和服务容量调整;计算时要限制指数,避免溢出。设备稳定在线一段时间后再重置失败计数,可以避免“刚连上就断开”时反复进入高频重试。
只让一个连接管理模块负责重连。SDK 已有自动重连时,应配置和协调它,避免业务任务再启动另一套竞争逻辑。认证失败、证书失效等问题应记录并等待修复,不能靠快速重试解决。重连状态可以概括为:等待网络 → 连接 → 恢复订阅 → 同步状态 → 就绪,任一阶段断线都回到相应等待状态。
第 6 章 接入真实设备:把通信实验变成嵌入式程序
前面已经用工具验证了主题与行为。现在把模拟设备替换成 ESP32:测试后台仍用 MQTTX,先确认板子能收发,再接入真实传感器和水泵驱动。
6.1 把实验搬到设备上,需要哪些代码?
先解决网络地址。ESP32 中的 127.0.0.1 指板子自身,不能指向电脑。第 3 章的 Broker 也只监听回环地址,需要为局域网实验调整为:
listener 1883 0.0.0.0
allow_anonymous true
persistence false
停止旧 Broker 后按同样命令重启。电脑和 ESP32 应处于互通的测试局域网;防火墙允许该测试网访问 TCP 1883,确认无线网络没有客户端隔离。查出电脑局域网 IP,例如 192.168.1.100,并把代码中的地址替换成实际值。这个匿名监听只用于受控实验,实验后关闭或恢复本机监听;正式接入方案见第 7 章。
下面是 ESP-IDF 5.5.1 风格的 MQTT 收发模块,用于加入已经完成 NVS、网络接口、事件循环和 Wi-Fi 初始化的工程。在获得 IP 后调用 mqtt_demo_start(),只调用一次。示例使用内置 ESP-MQTT,组件构建依赖需包含 mqtt;单独放在 mqtt_demo.c 时,还要把这个源文件加入组件的 SRCS。其他版本可能需要额外引入组件或调整配置字段,以所用 SDK 文档为准。
这段代码先做三件事:连接后订阅命令、在订阅事件中检查结果、复制收到的小消息用于观察。它不驱动水泵,也不宣称已经具备第 5 章的完整恢复逻辑。
#include <stdbool.h>
#include <string.h>
#include "esp_err.h"
#include "esp_event.h"
#include "esp_log.h"
#include "mqtt_client.h"
#define TOPIC_COMMAND "farm/p1/d001/command"
#define TOPIC_STATE "farm/p1/d001/state"
#define TOPIC_ONLINE "farm/p1/d001/online"
static const char *TAG = "mqtt_demo";
static esp_mqtt_client_handle_t client;
static int command_sub_id = -1;
static void mqtt_event_handler(void *arg, esp_event_base_t base,
int32_t event_id, void *event_data)
{
esp_mqtt_event_handle_t e = event_data;
switch ((esp_mqtt_event_id_t)event_id) {
case MQTT_EVENT_CONNECTED:
command_sub_id = esp_mqtt_client_subscribe(
client, TOPIC_COMMAND, 1);
ESP_LOGI(TAG, "subscribe request id=%d", command_sub_id);
break;
case MQTT_EVENT_SUBSCRIBED:
// MQTT 3.1.1 SUBACK: 此处只请求了一个主题。
if (command_sub_id >= 0 && e->msg_id == command_sub_id &&
e->data_len == 1 && e->data &&
(unsigned char)e->data[0] <= 1) {
ESP_LOGI(TAG, "command subscription ready, qos=%u",
(unsigned int)(unsigned char)e->data[0]);
} else {
ESP_LOGW(TAG, "subscription not confirmed or rejected");
}
break;
case MQTT_EVENT_DATA: {
// 教学版仅接收单次事件完整携带、且长度不超过 255 字节的命令。
// 被分片或过长的消息直接拒绝;生产实现应使用有上限的重组缓存。
if (e->current_data_offset != 0 ||
e->total_data_len != e->data_len ||
e->data_len <= 0 || e->data_len > 255 ||
e->topic_len != (int)strlen(TOPIC_COMMAND) ||
!e->topic || !e->data ||
memcmp(e->topic, TOPIC_COMMAND, strlen(TOPIC_COMMAND)) != 0) {
ESP_LOGW(TAG, "ignore fragmented, oversized or other message");
break;
}
char payload[256];
memcpy(payload, e->data, (size_t)e->data_len);
payload[e->data_len] = '\0';
// 仅实验日志;这里还没有解析 JSON 或执行硬件动作。
ESP_LOGI(TAG, "command received: %s", payload);
break;
}
case MQTT_EVENT_PUBLISHED:
ESP_LOGI(TAG, "publish acknowledged, id=%d", e->msg_id);
break;
case MQTT_EVENT_DISCONNECTED:
command_sub_id = -1;
ESP_LOGW(TAG, "disconnected");
break;
case MQTT_EVENT_ERROR:
ESP_LOGE(TAG, "MQTT error, inspect error_handle and network logs");
break;
default:
break;
}
}
void mqtt_demo_start(void)
{
const esp_mqtt_client_config_t cfg = {
.broker.address.uri = "mqtt://192.168.1.100:1883",
.credentials.client_id = "device-d001",
.session.protocol_ver = MQTT_PROTOCOL_V_3_1_1,
.session.keepalive = 60,
.session.last_will.topic = TOPIC_ONLINE,
.session.last_will.msg = "{\"online\":false}",
.session.last_will.qos = 1,
.session.last_will.retain = 1,
};
client = esp_mqtt_client_init(&cfg);
if (!client) {
ESP_LOGE(TAG, "client init failed");
return;
}
ESP_ERROR_CHECK(esp_mqtt_client_register_event(
client, ESP_EVENT_ANY_ID, mqtt_event_handler, NULL));
ESP_ERROR_CHECK(esp_mqtt_client_start(client));
}
// 从业务任务调用;先确认连接和订阅就绪。
// 固定数值仅用于收发测试,不代表真实传感器读数或硬件状态。
void mqtt_demo_publish_sample(void)
{
if (!client) {
return;
}
const char *sample =
"{\"temperature\":26.3,\"humidity\":61,\"pump\":false}";
int id = esp_mqtt_client_enqueue(
client, TOPIC_STATE, sample, 0, 1, 1, true);
ESP_LOGI(TAG, "state enqueue id=%d", id);
}
开始测试前,断开 MQTTX 中同为 device-d001 的模拟设备,保留后台连接。在设备日志确认命令订阅成功后,从业务任务调用一次 mqtt_demo_publish_sample(),后台应收到状态;再从后台发布第 3.2 节命令,串口应打印收到的 JSON。
本例使用 enqueue() 把状态交给 SDK 发送队列,返回非负消息编号只说明提交成功,负值需要处理。QoS 1 的 MQTT_EVENT_PUBLISHED 可以用于观察协议确认;发送失败、队列容量、过期和断线期间缓存仍需要业务管理。不要在采样循环中无限积压消息。示例使用默认的干净会话和 SDK 自动重连,每次连接后重新订阅;指数退避、业务就绪状态与状态重同步需要按第 5 章继续完善。
接收代码为什么没有直接对 e->data 调用 JSON 解析?SDK 交给回调的 Topic 和数据不保证以零字符结尾,一条较大消息也可能分成多次 DATA 事件。因此必须按长度处理,并在回调结束前复制需要保留的数据。正式程序应把复制后的完整命令投递到队列。接口、分片和事件字段可查阅 ESP-IDF 5.5.1 ESP-MQTT 文档。
换平台时,通信步骤不变,实现入口会变化:Linux/Paho 有同步与异步接口,要明确选择哪一种;STM32+4G 可以调用模组内置的 MQTT AT 命令,也可以由 MCU 通过 LwIP/PPP 和 MQTT 库完成协议处理。前者要核对模组的 AT 手册与异步通知,后者要自行管理更多内存、任务和网络状态,不能混用两套接口的行为假设。
6.2 收到开泵命令后,程序应该怎样执行?
串口打印命令只证明“消息到达了程序”。要真正控制水泵,还得补上解析、校验、执行、反馈。
假设你直接在 MQTT 回调中打开继电器,再等待 10 秒确认水流。回调所在的网络任务可能因此无法及时处理后续消息和心跳。设备表面上在执行控制,通信却开始超时,重连又带来更多重复请求。
更合适的职责划分是:回调负责有界复制和快速校验,控制任务负责业务和硬件操作,发送任务或 SDK 队列负责回传。下面是职责示意伪代码,不是可直接编译的 SDK 示例:
收到 MQTT 消息:
核对 Topic、分片完整性和总长度
把完整 Payload 复制到受控消息缓冲
非阻塞投递到控制队列
若队列已满,记录并按约定报告忙,不能静默当作已执行
尽快返回网络回调
控制任务:
取出并解析命令
校验 requestId、类型、参数、版本和有效期
若 requestId 已完成且参数一致,返回已保存的结果
若 requestId 正在处理,不再启动第二次动作
校验设备当前是否允许启动水泵
调用硬件驱动,等待有界的执行结果
读取与业务承诺相符的反馈,生成 completed 或 failed
保存必要的执行结果,发布 command/ack
发布新的真实 state
这里的“真实状态”要讲清楚:写 GPIO 成功,只能说明控制输出已经改变;如果系统承诺“水泵实际运行”,就需要接触器、电流或水流等反馈。没有反馈传感器时,应把状态定义为“控制输出已开启”,不能把它宣传成实际出水。
读到真实温湿度后,再替换第 6.1 节的固定测试值。正常就绪时发布 online=true;退出时执行第 5.1 节的正常下线流程。重连后的订阅恢复、快照同步和业务队列处理,也应纳入同一套状态管理。
RAM、Flash 和看门狗在这里都有具体含义:消息与队列必须设上限,Flash 缓存要考虑擦写寿命和掉电一致性,驱动等待必须有超时。看门狗能帮助恢复卡死,但不能用不断喂狗来掩盖网络或业务任务已经停止工作。
6.3 设备与后台怎样约定消息内容?
当你能收发字符串以后,接下来的困难往往不在 MQTT,而在双方是否理解同一份数据。26.3 是摄氏度还是华氏度?on 是布尔值还是字符串?一个未知字段应忽略还是拒绝?这些需要业务协议约定。
把前面用过的主题统一下来:
Topic 后缀,前缀均为 farm/p1/d001/ | | | |
|---|
state | | | QoS 1、Retain 开启;高频可替代遥测可另用 QoS 0 |
event | | | |
command | | | |
command/ack | | | |
online | | | |
这张表是案例约定,不是所有项目必须照搬的 MQTT 标准格式。主题方向也应与 Broker 权限保持一致。
一份便于入门的命令约定如下:
{
"requestId": "r002",
"type": "set_pump",
"params": {"on": false},
"version": 1
}
可以明确规定:本例命令最多 255 字节;requestId 是长度受限的非空字符串;type 只接受已实现的命令;params.on 必须是 JSON 布尔值;version 必须受支持;缺必选字段或类型错误要拒绝,未知可选字段按兼容规则处理。第 6.1 节使用 256 字节接收缓冲,为最多 255 字节的内容额外保留一个字符串结束符位置。
状态则补充温度单位、湿度范围、采样时间与时钟来源、状态版本的生成方式。命令如果允许离线排队,再增加 expiresAt 或其他有效性条件,并明确设备时钟未校准时怎样处理。MQTT 不会检查这些字段。
JSON 很适合第一版,因为 MQTTX 和日志都能直接观察。资源或流量确实成为瓶颈后,再测量并比较:
- 自定义二进制可以很紧凑,但要承担字段布局、兼容与解码工具的维护。
- Protobuf 通过 schema 约定字段,需要生成代码和运行时资源评估;不代表所有大型项目都必须采用它。
- CBOR 用二进制表达常见数据类型,可在体积与实现成本之间提供另一种选择。
无论使用哪种格式,都应保留版本和边界校验。最后,MQTT 传的是消息,温度数据是否保存历史、告警是否需要补偿,仍由后台业务系统决定。
第 7 章 从能运行到能交付:安全与故障排查
在局域网里跑通,是第一步。要让设备长期工作,还需要回答两个问题:谁有资格控制它,出问题时怎样找到断点。安全配置和可观测性都应落到前面已经跑通的链路上。
7.1 知道设备 ID,就能控制别人的水泵吗?
如果 Broker 允许匿名连接,并且任何客户端都能向 farm/p1/d001/command 发布,那么知道主题名的人就可能发送命令。把设备 ID 写得复杂,不等于建立了权限控制。
正式环境至少要解决三层问题:
| | |
|---|
| | TCP 1883 常用于明文,8883 常用于 TLS;端口号本身不提供加密 |
| | |
| | 设备只能发布、订阅被允许的主题,后台也按业务权限隔离 |
例如,d001 只应订阅自己的 command,并发布自己的 state、event、command/ack 和 online;不能给 d002 发命令,也不能订阅整个农场的数据。认证成功但没有对应主题权限时,仍可能连接正常却收发失败。
TLS 还可以使用双向证书认证,让服务端验证设备证书。无论采用哪种凭据,都要考虑存储、轮换与吊销;共享引导凭据与每台设备正式通信凭据的职责也应分清。令牌可采用 JWT 等形式,但是否可撤销、何时过期,要看平台的实际验证和管理方案。
设备应校准系统时间,否则证书有效期检查可能失败;不能长期关闭证书校验来“修好连接”。TLS 保护传输后,Payload 的长度、范围和业务权限仍要检查,日志也不应打印密码、令牌或私钥。
7.2 连不上、收不到、反复掉线,分别从哪里查?
排查时沿着消息路径走,不要一出现故障就改 QoS 或重装 SDK。先确认失败发生在哪一层,再看对应证据。
| | |
|---|
| | 电脑端能连但板子不能连,优先检查回环监听与网络互通 |
| | |
| | 看 Broker 认证日志,确认客户端实际使用的参数 |
| | 用后台测试工具替代一端,确认消息发布位置与订阅匹配 |
| | 先断开其他同 ID 测试端,再观察断线原因与时间间隔 |
| | |
| requestId、离线队列、Retain、有效期 | |
| | 检查设备是否仍在上报、Broker 中保留的具体 Payload |
| | |
记录一条问题消息时,至少保留 Client ID、设备 ID、Topic、方向、QoS、Retain、业务 requestId、连接代次、时间以及错误码。协议消息编号可以辅助定位单段交互,但不能跨重连替代业务 ID。
抓包可以帮助理解 CONNECT、SUBSCRIBE、PUBLISH 和确认流程;TLS 抓包通常不能直接读到业务明文,此时应结合端点调试日志与 Broker 日志。记录必要信息时,对凭据和敏感业务内容做脱敏。
7.3 接手一个已有项目,怎样找到 MQTT 的关键代码?
不要从 Broker 的协议解析源码开始。拿一条设备命令作为线索,按执行顺序找入口:
- 找 Broker 地址、端口、MQTT 版本、Client ID 和凭据配置,确认连接到哪里。
- 找网络就绪事件和连接调用,确认何时创建客户端、如何处理连接结果。
- 找订阅调用与 SUBACK 处理,列出命令主题和恢复条件。
- 找消息回调,检查分片、长度、内存所有权和队列投递。
- 找业务解析和控制任务,确认请求 ID、幂等、驱动反馈及错误处理。
- 找发布接口,确认应答、状态、事件和在线消息各自从哪里产生。
- 找重连、缓存与退出流程,确认订阅、状态和凭据如何恢复或更新。
边读边填一张表,比记下许多孤立的函数名更有用:
| | | | |
|---|
farm/p1/d001/command | | | | |
farm/p1/d001/state | | | | |
如果现在能追踪一次“后台点开泵 → 命令抵达 → 驱动执行 → 页面显示真实结果”,并解释任意一步失败后如何发现和恢复,就已经具备阅读实际 MQTT 项目的基础。
第 8 章 设备变多以后:认识 MQTT 在完整系统中的位置
前面的目标是一台设备正确运行。规模扩大后,通信模型仍然相同,变化的是连接、消息与状态管理的成本。本章作为进阶选读,先建立判断问题的方向。
8.1 从一台设备到很多台,压力增加在哪里?
假设有 10 万台设备,每台每分钟上报一次,平均约为每秒 1667 条上报。这个数字还没有算订阅转发、确认报文、心跳、下控和重连。如果设备都在整分钟同时发,平均值又会掩盖瞬时尖峰。
所以评估容量至少要同时看:连接数、每秒发布量、Payload 大小、每条消息匹配多少订阅者、QoS 比例、离线积压量和 TLS 建连速率。空闲连接也会占用内存、文件描述符和内核资源,并非完全没有成本。
Broker 的连接承载量没有脱离条件的统一答案。具体硬件、实现、插件、持久化方式和消息模型都会影响结果,不能直接把某次“几十万空闲连接”的测试当作本项目容量保证。
设备侧可以错开采样与初始连接时间,并采用退避重连;不能靠延迟协议要求的心跳来缓解负载。平台侧则需要限流、队列上限、监控和针对实际业务的压测。
8.2 后台增加多个实例,怎样分担消息处理?
如果启动三个后台实例,三个都普通订阅 farm/p1/+/state,它们通常各收到一份。对于“各自完成不同业务”的服务,这正是需要的;对于“三个相同工作进程分担同一批任务”,就可能造成重复处理。
支持共享订阅的 Broker 可以让同一组内的订阅者分担匹配消息,例如:
$share/storage-workers/farm/p1/+/state
其中 storage-workers 是共享组名,后面才是真正的主题过滤器。同组内由 Broker 选择成员接收一条匹配消息,具体调度策略取决于实现。MQTT 5.0 定义了共享订阅;一些 Broker 也为 MQTT 3.1.1 提供扩展支持,应先核对服务端能力。
共享订阅解决分发方式,仍不能替代消费者的业务幂等,也不能简单认为它保证每台设备的消息永远交给同一进程。需要严格顺序或设备归属时,还要明确分区与调度策略。
Broker 集群关注连接接入、订阅路由、会话分布和节点故障恢复;四层负载均衡负责把 TCP 连接送到节点,TLS 卸载则取决于采用的代理能力和部署方式。并非所有集群都要求固定的一种哈希分片方案。
后台还可以把消息转到 Kafka,供存储、告警和分析服务继续消费。但桥接队列、确认时机、故障重试和背压仍需设计。不能因为增加了 Kafka,就推断任何下游故障都不会影响消息链路。
8.3 外网断了,现场设备还能继续联动吗?
如果传感器和水泵都直接连接云端,一旦外网断开,云上的“温度过高就启动设备”规则便可能无法继续执行。对现场控制来说,这有时比历史数据暂时不上云更严重。
一种办法是在网关部署边缘 Broker,让本地设备先连接它,再由网关把需要的消息桥接到云端。Mosquitto、NanoMQ 等可以用于相应的本地或边缘部署,具体选择取决于资源、功能与运维需求。
但仅仅把 Broker 放到本地还不够。联动规则也要能在本地执行,控制权限、配置和必要状态也要可用,现场才能在断外网时继续工作。涉及水泵保护等要求的逻辑,还应由设备或本地控制系统独立保证。
网络恢复后,网关按约定同步需要上传的数据。桥接需要明确方向、主题映射、消息有效期和循环防护,不能把两个方向的所有主题无条件互转,也不能默认本地 Broker 已经替你永久保存了全部历史。
回到最初那台设备,MQTT 负责把温度、命令和状态有组织地送到对应参与者。设备程序负责正确执行,后台负责业务与历史数据,网络与安全配置负责连接边界。把这些职责接起来,才是一套能从实验走向现场运行的物联网通信方案。