---
type: study
status: active
created: 2026-09-10
updated: 2026-09-10
sensitivity: private
tags: [study, embedded, c, esp32, cores3]
sources:
  - "https://mqtt.org/mqtt-specification/"
  - "https://docs.oasis-open.org/mqtt/mqtt/v5.0/os/mqtt-v5.0-os.html"
  - "https://docs.emqx.com/en/emqx/latest/access-control/authz/authz.html"
  - "https://docs.espressif.com/projects/esp-idf/en/v5.5.1/esp32s3/api-reference/protocols/mqtt.html"
---
# MQTT、EMQX 与 Java：建立可解释的设备闭环

## MQTT 解决异步消息分发

客户端向 broker 发布消息，broker 按订阅关系分发。主题组织路由，payload 的格式由应用约定，MQTT 本身不强制 JSON。MQTT 5.0 的规范由 OASIS 发布。[MQTT 规范入口](https://mqtt.org/mqtt-specification/)

```mermaid
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](https://docs.oasis-open.org/mqtt/mqtt/v5.0/os/mqtt-v5.0-os.html) 为准。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”，便于重复执行和观察。

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

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

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

## 设备和 broker 的职责边界

设备使用唯一身份、TLS 和最小主题权限。EMQX 的认证解决“是谁”，授权解决“允许发布/订阅哪些 topic”。设备不能读取其他设备命令，后端按可信身份映射消息归属。[EMQX Authorization](https://docs.emqx.com/en/emqx/latest/access-control/authz/authz.html)

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

## ESP-MQTT 收包的缓冲区边界

消息可能分片到达。接收端检查 `data_len`、`total_data_len`、`current_data_offset`，在大小上限内组装完整消息，再解析 JSON。不能假定一次回调就是完整 payload，或 payload 自带零终止符。[ESP-MQTT Events](https://docs.espressif.com/projects/esp-idf/en/v5.5.1/esp32s3/api-reference/protocols/mqtt.html)

## 过关测试

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

---

[[Study/embedded-cores3/CoreS3学习入口|返回学习入口]] · [[Study/embedded-cores3/CoreS3资料来源与版本|来源与版本]]