假设你正在做一套农业监测设备。板子能采集温湿度，也能控制一台水泵。现在，你希望在电脑上看到温度，并且点一下按钮，就能让远处的水泵启动。

先别急着背 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 可以运行在 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 会携带 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，新建以下两条连接：

两条连接均应显示成功，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 故意改错参数，看看消息为什么收不到

把实验结果先记下来，再按下面顺序制造几个小故障。每次只改一项，观察后恢复，避免多个问题混在一起。

实验结束后断开告警端，把设备和后台恢复为第 3.1 节的配置。

如果这几次变化都能解释清楚，你已经理解了 MQTT 的基本模型。接下来再问一个更实际的问题：网络出了问题，消息会不会丢、会不会重复？

第 4 章 消息发出以后：怎样判断它有没有可靠送达

温度每分钟上报一次，少一条可能很快就被下一条替代。但“给水泵增加 10 分钟运行时间”如果重复执行，后果就不同了。理解可靠性，要先区分消息交付和业务动作。

4.1 温度偶尔少一条，控制命令也能这样处理吗？

MQTT 提供三种 QoS，决定一段发送方与接收方之间怎样完成协议交付：

为什么 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 收到了确认，为什么还不能认定水泵已经启动？

设想后台收到一条协议确认，但设备的继电器驱动失败，或者水泵处于保护状态。消息送到了，动作却没有完成。

因此需要分清三层结果：

1. 发布请求提交成功：SDK 接受了发送请求，或将其放进本地队列。
2. 协议交付完成：例如 QoS 1 收到 PUBACK，确认的是对端 MQTT 协议接收。
3. 业务执行结果明确：设备报告命令执行成功或失败，并上报实际状态。

本例让设备在执行结束后，向 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，做一次可重复的实验：

1. 暂时取消后台的 state 订阅。
2. 设备向 farm/p1/d001/state 发布温度 26.3，Retain 开启。
3. 后台重新订阅，应立即收到 26.3，即使设备此时没有再次发布。
4. 设备以 Retain 开启的方式发布温度 27.0；后台取消后重订阅，应获得 27.0。
5. 设备向同一主题发送零字节 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：

设备正常建立业务通信后，主动向同一主题发布 {"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 连接，却忘了恢复订阅，会出现“显示已连接，但再也收不到命令”。

如果希望断线后保留订阅与相应待投递消息，可以使用持久会话。两种协议版本的配置方式不同：

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 网络恢复了，怎样避免旧命令和旧状态造成混乱？

假设设备离线了半小时，期间后台发过“开泵”，后来用户又取消了灌溉。此时补投旧命令，虽然符合消息交付规则，却可能不符合当前业务意图。

因此，重连要同时处理通信恢复和业务恢复。建议按下面的顺序组织：

1. 等待 Wi-Fi、蜂窝网络等底层网络可用，再建立 MQTT 连接。
2. 收到成功 CONNACK 后，检查会话是否恢复。未恢复时重新订阅；已有会话也要核对必要订阅与当前业务是否一致，并检查 SUBACK 结果。
3. 恢复期间到达的消息先进入受控队列，按业务规则检查有效期、请求 ID 和顺序，不能假设它们一定等到初始化结束才到达。
4. 读取设备真实状态，发布当前快照和在线状态，再进入正常运行；补投命令执行后继续发布新的应答与状态。

对于瞬时控制，本例默认不依赖离线命令积压：后台发现设备离线时提示无法执行，恢复后让用户重新确认。确实要排队时，应在命令中加入有效期、设备配置版本或其他前置条件。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 是布尔值还是字符串？一个未知字段应忽略还是拒绝？这些需要业务协议约定。

把前面用过的主题统一下来：

这张表是案例约定，不是所有项目必须照搬的 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 写得复杂，不等于建立了权限控制。

正式环境至少要解决三层问题：

例如，d001 只应订阅自己的 command，并发布自己的 state、event、command/ack 和 online；不能给 d002 发命令，也不能订阅整个农场的数据。认证成功但没有对应主题权限时，仍可能连接正常却收发失败。

TLS 还可以使用双向证书认证，让服务端验证设备证书。无论采用哪种凭据，都要考虑存储、轮换与吊销；共享引导凭据与每台设备正式通信凭据的职责也应分清。令牌可采用 JWT 等形式，但是否可撤销、何时过期，要看平台的实际验证和管理方案。

设备应校准系统时间，否则证书有效期检查可能失败；不能长期关闭证书校验来“修好连接”。TLS 保护传输后，Payload 的长度、范围和业务权限仍要检查，日志也不应打印密码、令牌或私钥。

7.2 连不上、收不到、反复掉线，分别从哪里查？

排查时沿着消息路径走，不要一出现故障就改 QoS 或重装 SDK。先确认失败发生在哪一层，再看对应证据。

记录一条问题消息时，至少保留 Client ID、设备 ID、Topic、方向、QoS、Retain、业务 requestId、连接代次、时间以及错误码。协议消息编号可以辅助定位单段交互，但不能跨重连替代业务 ID。

抓包可以帮助理解 CONNECT、SUBSCRIBE、PUBLISH 和确认流程；TLS 抓包通常不能直接读到业务明文，此时应结合端点调试日志与 Broker 日志。记录必要信息时，对凭据和敏感业务内容做脱敏。

7.3 接手一个已有项目，怎样找到 MQTT 的关键代码？

不要从 Broker 的协议解析源码开始。拿一条设备命令作为线索，按执行顺序找入口：

1. 找 Broker 地址、端口、MQTT 版本、Client ID 和凭据配置，确认连接到哪里。
2. 找网络就绪事件和连接调用，确认何时创建客户端、如何处理连接结果。
3. 找订阅调用与 SUBACK 处理，列出命令主题和恢复条件。
4. 找消息回调，检查分片、长度、内存所有权和队列投递。
5. 找业务解析和控制任务，确认请求 ID、幂等、驱动反馈及错误处理。
6. 找发布接口，确认应答、状态、事件和在线消息各自从哪里产生。
7. 找重连、缓存与退出流程，确认订阅、状态和凭据如何恢复或更新。

边读边填一张表，比记下许多孤立的函数名更有用：

如果现在能追踪一次“后台点开泵 → 命令抵达 → 驱动执行 → 页面显示真实结果”，并解释任意一步失败后如何发现和恢复，就已经具备阅读实际 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 负责把温度、命令和状态有组织地送到对应参与者。设备程序负责正确执行，后台负责业务与历史数据，网络与安全配置负责连接边界。把这些职责接起来，才是一套能从实验走向现场运行的物联网通信方案。