MQTT、EMQX 与 Java：建立可解释的设备闭环

MQTT 解决异步消息分发

客户端向 broker 发布消息，broker 按订阅关系分发。主题组织路由，payload 的格式由应用约定，MQTT 本身不强制 JSON。MQTT 5.0 的规范由 OASIS 发布。MQTT 规范入口

flowchart LR
    C[CoreS3] -->|telemetry / ack| B[EMQX]
    B -->|订阅消息| J[Java 消费者]
    J --> D[(数据库)]
    J -->|command| B
    B -->|设备订阅| C
    J --> W[Web 或 App]

这是一份学习项目草案，没有据此修改现有后端或创建 EMQX 配置。Redis 仅在确有缓存、在线状态或协调需求时加入。

先分清 QoS、保留消息和遗嘱

机制	它负责什么	应用仍要负责什么
QoS 0	至多一次的协议交付方式	能否接受丢失
QoS 1	至少一次，可能重复	幂等与去重
QoS 2	协议接收方范围内恰好一次交付	数据库、外部动作与业务端到端一致性
Retain	为主题保留最近的保留消息	过期、清除、消息是否适合后到订阅者
Will	连接异常结束等协议条件下发布预设消息	心跳、业务健康与及时性判断

准确语义以 OASIS MQTT 5.0 为准。QoS 1 的 PUBACK 或客户端 published 事件不代表 Java 已落库，更不代表设备已经执行命令。

一份可以讨论的 topic 契约

以下全是教学命名。tenantId、deviceId 来自认证与授权绑定，不能相信 payload 自报身份。

Topic 后缀（前缀 `lab/{tenantId}/{deviceId}/`）	发布 → 订阅	建议 QoS / Retain	语义
`telemetry`	设备 → Java	1 / false	观测值；应用层可能重复
`status`	设备 → 平台	1 / true	最近连接状态，附时间与启动标识
`command`	Java → 设备	1 / false	指令；带 `commandId`、有效期及参数
`ack`	设备 → Java	1 / false	对某条指令的处理结果

状态 retained 消息表示最近已知值，不证明此刻在线。命令默认不保留，避免重连后自动执行很早以前的动作；配置同步可另设计“期望状态”模型。

命令成功必须有业务证据

第一条练习命令用“把采样周期设置为 1000 ms”，便于重复执行和观察。

平台创建 commandId
  → 发布 command
  → 设备校验身份、有效期、参数、重复性
  → 应用配置
  → 发布 ack（同一个 commandId）
  → 平台记录最终结果

本课程草案把状态分为 accepted、applied、rejected、failed。accepted 仅表示接收；若选择承诺配置掉电保存，applied 应在保存成功且运行参数生效后回传。平台等不到回执时显示“结果未知/等待查询”，不能直接推断执行失败。

相同 commandId 重发应返回已知结果，不重复执行有副作用动作。设备重启后若丢了内存中的去重记录，保证也会丢失：因此有副作用指令需要持久化或其他可恢复协议。第一轮选“设置绝对值”，先避开难以恢复的“再加一次”。

设备和 broker 的职责边界

设备使用唯一身份、TLS 和最小主题权限。EMQX 的认证解决“是谁”，授权解决“允许发布/订阅哪些 topic”。设备不能读取其他设备命令，后端按可信身份映射消息归属。EMQX Authorization

Java 消费者要对去重键设置唯一约束或使用等效的原子写入。遥测可以采用 (deviceId, bootId, sequence)；命令采用 commandId。数据库中的具体表和事务边界，待实现时再确认。

ESP-MQTT 收包的缓冲区边界

消息可能分片到达。接收端检查 data_len、total_data_len、current_data_offset，在大小上限内组装完整消息，再解析 JSON。不能假定一次回调就是完整 payload，或 payload 自带零终止符。ESP-MQTT Events

过关测试

正常上报；broker 重启后恢复；QoS 1 重复消息；设备重启后重复命令；过期命令；无权限 topic；超大/分片/非法 payload；命令已生效但回执丢失。逐项记录预期与实测，才算设备到后端链路验收。

返回学习入口 · 来源与版本