yuhaiming/water

Template

Fork 0

Files

yuhaiming 509e760f3d 设备bug修改新增设备解绑功能

2026-08-25 16:18:34 +08:00

21 KiB

Raw Permalink Blame History

Water MQTT 通信协议说明文档

1. 系统架构概览

┌─────────────┐       MQTT Broker        ┌─────────────────┐
│   物理设备    │  ◄══════════════════►   │   Water 后端服务   │
│  (ESP32等)   │    (SSL/TCP连接)         │  (Spring Boot)   │
└─────────────┘                          └─────────────────┘
       │                                         │
       │  上行 (publish)                          │  下行 (subscriber)
       │  /{identity}/publish/xxx                 │  /{deviceNo}/subscriber/xxx
       │                                         │
       ▼                                         ▼
  设备注册、数据上报、                        命令下发、排程同步、
  浇水完成、异常告警、ACK                     设备绑定/解绑/初始化

核心组件

组件	包路径	职责
`MqttClientManager`	`org.dromara.mqtt`	MQTT 客户端连接管理、消息收发
`MqttMessageDispatcher`	`org.dromara.app.mqtt`	上行消息路由分发（策略模式）
`MqttTopicHandler`	`org.dromara.app.mqtt`	上行消息处理器接口
`DeviceMqttCommandPublisher`	`org.dromara.mqtt`	下行命令发布器
`DeviceCommandServiceImpl`	`org.dromara.app.service.impl`	命令构造与业务编排
`MqttCommandAckService`	`org.dromara.mqtt`	命令 ACK 确认与重试
`DeviceIdentityResolver`	`org.dromara.app.handler`	设备标识解析（MAC/设备编号 → deviceNo）

2. Topic 约定

2.1 上行 Topic（设备 → 服务端）

服务端通过通配符订阅（如 +/publish/#），由 MqttMessageDispatcher 按正则分发到对应 Handler。

Topic 模式	正则	处理器	说明
`/{identity}/publish/register`	`^/([^/]+)/publish/register$`	`DeviceRegisterHandler`	设备注册
`/{identity}/publish/status`	`^/([^/]+)/publish/status$`	`DeviceStatusHandler`	设备 LWT 离线通知（仅处理 offline）
`/{identity}/publish/power`	`^/([^/]+)/publish/power$`	`DeviceDataHandler`	电量数据上报及在线心跳
`/{identity}/publish/finish/key`	`^/([^/]+)/publish/finish/key$`	`KeyFinishHandler`	按键/手动浇水完成
`/{identity}/publish/finish/schedule`	`^/([^/]+)/publish/finish/schedule$`	`ScheduleFinishHandler`	排程浇水完成
`/{identity}/publish/error`	`^/([^/]+)/publish/error$`	`ErromesHandler`	设备异常告警
`/{identity}/publish/ack`	`^/([^/]+)/publish/ack$`	`DeviceCommandAckHandler`	命令执行确认

首次注册可使用 MAC 地址；注册完成后的业务 Topic 使用 deviceNo。LWT 离线 Topic 兼容 MAC 地址，服务端通过 DeviceIdentityResolver 解析为 deviceNo。

2.2 下行 Topic（服务端 → 设备）

Topic 模式	用途	构造方式
`/{deviceNo}/subscriber/cmd`	通用命令下发	`DeviceMqttCommandPublisher.buildCommandTopic()`
`/{deviceNo}/subscriber/schedule`	排程专用下发	`DeviceCommandServiceImpl.buildScheduleTopic()`

可通过 mqtt.topics.publish-prefix 配置前缀，如配置为 /water，则实际 Topic 为 /water/{deviceNo}/subscriber/cmd。 deviceNo 统一为小写格式。

3. 消息格式

3.1 下行命令通用结构（DeviceCommand）

所有下行命令通过 DeviceCommand 对象构造，最终以 payload 字段的 JSON 发送到 MQTT。

// DeviceCommand 字段
String commandId;      // 命令唯一标识（UUID，自动生成）
String deviceNo;       // 设备编号
String deviceMac;      // 设备 MAC 地址
String commandType;    // 命令类型
String topic;          // 发送目标 Topic（可自定义，默认自动构建）
Map<String, Object> payload;  // 命令负载（最终序列化为 JSON 发送）
int retryCount;        // 当前重试次数
long createdAt;        // 创建时间戳
long lastSentAt;       // 最后发送时间戳
long nextRetryAt;      // 下次重试时间戳

发送时自动注入到 payload 的字段：

{
  "commandId": "自动生成的UUID",
  "commandType": "命令类型",
  // ... 其他业务字段
}

与用户绑定关系有关的命令（switchDevice、排程命令、queryPower、otaUpgrade 和自定义命令）还会携带当前关系的 bindingId。设备应在对应上报或 ACK 中原样返回 bindingId；服务端只接受与当前 ACTIVE 绑定一致的报文，旧绑定报文会被丢弃。bindDevice 使用命令自身携带的目标 bindingId，initDevice 不携带绑定标识。

3.2 上行 ACK 结构（DeviceCommandAck）

{
  "deviceNo": "设备编号",
  "commandId": "对应下发命令的commandId",
  "bindingId": "对应下发命令的绑定标识（有则原样返回）",
  "ack": "receive",
  "status": "兼容旧协议的确认标志",
  "message": "可选的文本消息",
  "powerLevel": 4,
  "charging": 0
}

设备确认标志优先使用 ack 字段（以 "receive" 开头即视为接受/成功，如 "receive"、"receive deviceNo"）；status 为兼容旧协议的字段。

powerLevel / charging 仅在 queryPower 命令的 ACK 中返回，为数字类型，服务端会自动转为字符串后分别更新到设备表的 power_level 和 power_status 字段。

也支持非 JSON 格式的纯文本 ACK（当设备只有一条待确认命令时可自动匹配）。

4. 下行命令类型详解

4.1 设备开关命令 — `switchDevice`

触发: 用户手动开启/关闭设备浇水
接口: PUT /app/v1/switchDevice
Topic: /{deviceNo}/subscriber/cmd

{
  "commandId": "xxx",
  "commandType": "switchDevice",
  "deviceNo": "01",
  "bindingId": "当前绑定标识",
  "cmd": "1",
  "startTime": "2026-06-25 14:35:00",
  "durationMin": 20
}

字段	类型	说明
`cmd`	String	`"1"` = 开启浇水，`"0"` = 停止浇水
`startTime`	String	开始时间（`yyyy-MM-dd HH:mm:ss` 或 `HH:mm`）
`durationMin`	Integer	持续时间（分钟），开启时必填且 > 0

附加行为:

开启时自动创建手动浇水日志记录
停止时自动结束进行中的手动浇水日志

4.2 设备绑定命令 — `bindDevice`

触发: 用户在 APP 绑定设备
接口: POST /app/v1/addDevice
Topic: /{deviceNo}/subscriber/cmd

{
  "commandId": "xxx",
  "commandType": "bindDevice",
  "bindStatus": true,
  "deviceNo": "01",
  "cmd": -1
}

字段	类型	说明
`bindStatus`	Boolean	`true` = 已绑定
`deviceNo`	String	分配给设备的编号
`cmd`	Integer	`-1` = 非浇水指令

4.3 设备初始化/解绑命令 — `initDevice`

触发: 用户删除/解绑设备
接口: DELETE /app/v1/deleteDevice/{deviceNos}
Topic: /{deviceNo}/subscriber/cmd

{
  "commandId": "xxx",
  "commandType": "initDevice",
  "deviceNo": "01",
  "initStatus": true,
  "cmd": "init"
}

字段	类型	说明
`initStatus`	Boolean	`true` = 需要恢复出厂设置
`cmd`	String	`"init"` = 初始化指令

附加行为: 服务端同时清除用户绑定、排程关联、浇水日志。

4.4 设备编号下发 — `registerDeviceNo`

触发: 设备通过 MQTT 注册上线后，服务端自动回复
接口: 无（DeviceRegisterHandler 自动触发）
Topic: /{deviceMac}/subscriber/cmd

这是唯一保留 MAC 作为 Topic 第一段的下行命令；其余下行命令统一使用 deviceNo。

{
  "commandId": "xxx",
  "commandType": "registerDeviceNo",
  "deviceNo": "01",
  "deviceMac": "aa:bb:cc:dd:ee:ff"
}

4.5 排程绑定命令 — `bindSchedule`

触发: 用户绑定排程到设备 / 修改排程内容 / 修改排程状态
接口: POST /app/v1/addScheduleDevice · PUT /app/v1/updataschedule · PUT /app/v1/editScheduleStatus
Topic: /{deviceNo}/subscriber/schedule（排程专用 Topic）

{
  "commandId": "xxx",
  "commandType": "bindSchedule",
  "cmd": -1,
  "deviceNo": "01",
  "bindingId": "当前绑定标识",
  "schedule": {
    "id": 1,
    "name": "每日浇水",
    "status": "1"
  },
  "details": [
    {
      "id": 10,
      "weekday": "1",
      "timeData": [
        { "startTime": "08:00", "durationMin": 15 },
        { "startTime": "18:00", "durationMin": 10 }
      ],
      "triggerType": "0",
      "status": "1"
    }
  ]
}

字段	类型	说明
`schedule.id`	Long	排程 ID
`schedule.name`	String	排程名称
`schedule.status`	String	`"1"` = 启用，`"0"` = 停用
`details[].weekday`	String	星期几（`1`-`7`）
`details[].timeData`	Array	时间段列表
`details[].triggerType`	String	触发类型
`details[].status`	String	明细状态

4.6 排程解绑命令 — `unbindSchedule`

触发: 用户删除排程与设备的绑定关系 / 删除排程
接口: DELETE /app/v1/deleteScheduleDevice · DELETE /app/v1/deleteschedule/{ids}
Topic: /{deviceNo}/subscriber/schedule

{
  "commandId": "xxx",
  "commandType": "unbindSchedule",
  "cmd": -1,
  "deviceNo": "01",
  "bindingId": "当前绑定标识",
  "scheduleId": 1,
  "unbind": true
}

字段	类型	说明
`scheduleId`	Long	要解绑的排程 ID
`unbind`	Boolean	`true` = 解绑

4.7 自定义命令 — 任意 `commandType`

触发: 通过 sendCustomCommand 接口下发
Topic: /{deviceNo}/subscriber/cmd

{
  "commandId": "xxx",
  "commandType": "自定义类型",
  // ... 自定义扩展字段
}

4.8 电量查询命令 — `queryPower`

触发: 用户在 APP 主动查询设备电量接口: POST /app/v1/queryPower/{deviceNo} Topic: /{deviceNo}/subscriber/cmd

{
  "commandId": "xxx",
  "commandType": "queryPower",
  "deviceNo": "01",
  "bindingId": "当前绑定标识"
}

设备 ACK 回复（/{deviceNo}/publish/ack）：

{
  "deviceNo": "2074069050702561282",
  "commandId": "xxx",
  "bindingId": "当前绑定标识",
  "powerLevel": 4,
  "charging": 0,
  "ack": "receive"
}

附加行为: 服务端收到 ACK 后，将 powerLevel 更新到设备表的 power_level 字段、charging 更新到 power_status 字段，并刷新 power_level_updatatime。powerLevel/charging 为数字类型（服务端自动转字符串入库），电量查询 ACK 不依赖 status 字段。

4.9 设备 OTA 升级命令 — `otaUpgrade`

触发: 管理员在设备列表选择设备（或全部在线设备）下发固件升级接口: POST /app/firmware/upgrade · POST /app/firmware/upgradeAll Topic: /{deviceNo}/subscriber/cmd

{
  "commandId": "xxx",
  "commandType": "otaUpgrade",
  "deviceNo": "01",
  "firmwareUrl": "https://api.example.com/app/firmware/download/0123456789abcdef0123456789abcdef.bin",
  "firmwareVersion": "1.0.0",
  "md5": "可选，固件 MD5 校验值"
}

字段	类型	说明
`firmwareUrl`	String	服务器固件下载地址
`firmwareVersion`	String	目标固件版本号
`md5`	String	固件 MD5 校验值（可选）

升级结果回报: 设备收到命令后回 ACK 确认，下载固件并升级，重启后重新走 /publish/register 上报新 version，服务端据此更新设备表的 fw_ver 字段。

4.10 设备恢复出厂命令 — `factoryReset`

触发: 管理员在设备列表控制设备恢复出厂接口: POST /app/device/factory-reset/{deviceNo} Topic: /{deviceNo}/subscriber/cmd

{
  "commandId": "xxx",
  "commandType": "factoryReset",
  "deviceNo": "01"
}

服务端只下发 MQTT 命令并等待设备 ACK，不清理服务端的用户绑定、排程或浇水日志。

5. 上行消息类型详解

5.1 设备注册 — `/{identity}/publish/register`

设备上电或重连后发送注册消息。

{
  "deviceName": "花园浇水器",
  "deviceMac": "aa:bb:cc:dd:ee:ff",
  "powerLevel": "85",
  "deviceEm": "型号",
  "deviceSn": "序列号",
  "version": "1.0.0"
}

服务端处理:

通过 MAC 或设备编号解析入库设备
更新设备注册信息（名称、电量、固件版本等）
自动回复设备编号（registerDeviceNo 命令）

注册成功即将设备标记为在线；设备取得 deviceNo 后仍必须定时上报电量，以持续刷新在线心跳。

5.2 电量上报 — `/{identity}/publish/power`

{
  "powerLevel": "78",
  "bindingId": "当前绑定标识"
}

服务端处理: 更新设备电量，将数据库设备状态改为在线，并将 Redis 在线心跳刷新为 600 秒。

设备必须周期上报，即使电量没有变化也要发送。建议每 5 分钟上报一次，连续 10 分钟未收到有效电量消息且期间无 ACK 刷新时，服务端将设备判定为离线。

5.3 设备离线遗嘱 — `/{identity}/publish/status`

{
  "status": "offline"
}

服务端处理: 收到非 retained 的 offline 或 0 后立即将设备改为离线。online 或 1 不再用于上线，服务端会忽略，上线状态只由电量心跳维护。

设备应将该消息配置为 MQTT LWT，建议 retain=false。Topic 第一段优先使用 deviceNo；设备注册前无法取得 deviceNo 时可使用 MAC 地址。

兼容设备也可以发送以下格式，服务端会使用 deviceMac 解析设备，并将 offline=true（字符串或布尔值）按离线处理：

{
  "deviceMac": "AA:BB:CC:DD:EE:FF",
  "offline": "true"
}

5.4 按键浇水完成 — `/{identity}/publish/finish/key`

{
  "deviceNo": "01",
  "commandId": "对应的命令ID",
  "bindingId": "当前绑定标识",
  "startTime": "2026-06-25 08:00:00",
  "endTime": "2026-06-25 08:15:00",
  "durationMin": 15,
  "triggerON": "schedule 或其他"
}

服务端处理:

triggerON = "schedule" → 排程触发，调用 confirmScheduleLog
其他 → 手动触发，调用 insertByBo 创建新记录

5.5 排程浇水完成 — `/{identity}/publish/finish/schedule`

{
  "deviceNo": "01",
  "commandId": "对应的命令ID",
  "bindingId": "当前绑定标识",
  "startTime": "2026-06-25 08:00:00",
  "endTime": "2026-06-25 08:15:00",
  "durationMin": 15,
  "triggerON": "mqtt on 或其他"
}

服务端处理: 调用 confirmScheduleLog 记录排程浇水日志。
triggerON = "mqtt on" 时标记为手动触发类型。

5.6 设备异常告警 — `/{identity}/publish/error`

{
  "errorCode": "E001",
  "message": "水泵故障",
  "bindingId": "当前绑定标识"
}

服务端处理: 当前仅日志记录，未做告警推送。

5.7 命令 ACK — `/{identity}/publish/ack`

{
  "commandId": "对应的命令ID",
  "status": "1",
  "message": "执行成功"
}

服务端处理:

从 Redis 移除 pending 命令
缓存 ACK 结果
刷新设备在线状态

6. 命令可靠性机制

6.1 命令生命周期

构造命令 → 存入 Redis(pending) → 发布到 MQTT → 等待 ACK
                ↓                                  ↓
           定时扫描                          收到 ACK
                ↓                                  ↓
         超时未确认?                        从 pending 移除
                ↓                          存入 Redis(ack)
      重试(最多 3 次)
                ↓
         达到上限? → 放弃并记录日志

6.2 重试策略

配置项	默认值	说明
`maxRetryCount`	3	最大重试次数
`retryIntervalMs`	5000	重试间隔（ms）
`scanIntervalMs`	5000	定时扫描间隔（ms）
`ackLockWaitMs`	3000	ACK 等待同一命令重试锁的最长时间（ms）
`pendingTtlSeconds`	86400	pending 命令 TTL（秒）
`ackTtlSeconds`	86400	ACK 结果缓存 TTL（秒）

6.3 离线处理

重试前检查设备在线状态（Redis 缓存）
设备离线时不执行 MQTT 发布，仅递增重试计数
达到最大重试次数后自动放弃

6.4 Redis 键规则

Key 模式	说明
`mqtt:command:pending:{commandId}`	待确认命令
`mqtt:command:ack:{commandId}`	ACK 确认结果
`mqtt:command:pending:ids`	待确认命令 ID 集合（Set）
`lock:mqtt:command:retry:{commandId}`	命令操作分布式锁
`mqtt:device:status:{deviceNo}`	设备在线状态缓存
`lock:mqtt:device:status:{deviceNo}`	设备状态写入锁

7. 接口 → MQTT 命令映射表

接口	路由	MQTT 命令类型	备注
绑定设备	`POST /addDevice`	`bindDevice`	Service 层下发
解绑设备	`DELETE /deleteDevice/{deviceNos}`	`initDevice`	Service 层下发（每台设备）
开关设备	`PUT /switchDevice`	`switchDevice`	Service 层下发 + 浇水日志
查询电量	`POST /queryPower/{deviceNo}`	`queryPower`	Service 层下发，ACK 回复电量
OTA 升级	`POST /firmware/upgrade` · `POST /firmware/upgradeAll`	`otaUpgrade`	Service 层下发，重启注册后更新 fw_ver
恢复出厂	`POST /device/factory-reset/{deviceNo}`	`factoryReset`	仅下发 MQTT 命令，不修改服务端关系数据
绑定排程设备	`POST /addScheduleDevice`	`bindSchedule`	Controller 层下发
解绑排程设备	`DELETE /deleteScheduleDevice`	`unbindSchedule`	Controller 层下发
修改排程状态	`PUT /editScheduleStatus`	`bindSchedule`	向所有绑定设备重新下发
修改排程内容	`PUT /updataschedule`	`bindSchedule`	向所有绑定设备重新下发
删除排程	`DELETE /deleteschedule/{ids}`	`unbindSchedule`	删除前通知所有绑定设备
设备注册（自动）	—	`registerDeviceNo`	设备上线后自动回复

8. 配置参考

mqtt:
  enabled: true
  broker-url: ssl://your-broker:8883
  client-id: water-server-01
  username: server
  password: xxxxxx
  qos: 1
  keep-alive: 60
  connection-timeout: 30
  max-inflight: 1000
  clean-session: false
  automatic-reconnect: true

  tls:
    enabled: true
    skip-verify: false

  topics:
    subscribe:
      - "+/publish/#"
   publish-prefix: ""           # 为空时 Topic = /{deviceNo}/subscriber/cmd

  async:
    core-pool-size: 8
    max-pool-size: 32
    consumer-count: 16
    queue-capacity: 5000
    batch-size: 100
    offer-timeout-ms: 50
    poll-timeout-ms: 100

  command-ack:
    enabled: true
    max-retry-count: 3
    retry-interval-ms: 5000
    scan-interval-ms: 5000
    pending-ttl-seconds: 86400
    ack-ttl-seconds: 86400
    pending-key-prefix: "mqtt:command:pending:"
    ack-key-prefix: "mqtt:command:ack:"
    pending-set-key: "mqtt:command:pending:ids"
    retry-lock-key-prefix: "lock:mqtt:command:retry:"
    retry-lock-ttl-ms: 30000
    ack-lock-wait-ms: 3000
    device-status-cache-prefix: "mqtt:device:status:"
    device-status-cache-ttl-seconds: 600

  offline-check:
    enabled: true
    ttl-compat-enabled: true
    interval-ms: 30000

9. 扩展指南

新增上行消息处理器

实现 MqttTopicHandler 接口
添加 @Component 注解
定义 topicPattern() 返回匹配正则（必须包含一个捕获组提取设备标识）
实现 handle(deviceIdentity, payload) 方法

@Component
public class MyNewHandler implements MqttTopicHandler {

    private static final Pattern PATTERN = Pattern.compile("^/([^/]+)/publish/mytype$");

    @Override
    public Pattern topicPattern() { return PATTERN; }

    @Override
    public void handle(String deviceIdentity, String payload) {
        // 处理逻辑
    }
}

无需修改 MqttMessageDispatcher，新 Handler 注册为 Bean 后自动发现。

新增下行命令类型

在 IDeviceCommandService 接口添加方法签名
在 DeviceCommandServiceImpl 实现命令构造与下发
在 Controller 调用新方法

10. 设备在线状态检测

设备在线状态由电量上报和离线遗嘱共同管理：

上线时机: 注册成功或收到包含 powerLevel 的有效电量上报时刷新
ACK 兼容逻辑: 保留现有 ACK 在线刷新逻辑，收到有效 ACK 也会延长在线缓存
缓存格式: { "deviceNo": "01", "status": "1", "lastReportTime": "2026-06-25T14:30:00Z" }
TTL: 默认 600 秒（10 分钟）
超时离线: 每 30 秒扫描一次非离线设备，缓存过期后更新数据库状态为离线
遗嘱离线: 收到非 retained 的 LWT status=offline/0 时立即离线
忽略在线状态: status=online/1 不会刷新在线状态
并发保护: 使用分布式锁（lock:mqtt:device:status:{deviceNo}）防止并发写入

21 KiB Raw Permalink Blame History Unescape Escape

Water MQTT 通信协议说明文档

1. 系统架构概览

核心组件

2. Topic 约定

2.1 上行 Topic（设备 → 服务端）

2.2 下行 Topic（服务端 → 设备）

3. 消息格式

3.1 下行命令通用结构（DeviceCommand）

3.2 上行 ACK 结构（DeviceCommandAck）

4. 下行命令类型详解

4.1 设备开关命令 — switchDevice

4.2 设备绑定命令 — bindDevice

4.3 设备初始化/解绑命令 — initDevice

4.4 设备编号下发 — registerDeviceNo

4.5 排程绑定命令 — bindSchedule

4.6 排程解绑命令 — unbindSchedule

4.7 自定义命令 — 任意 commandType

4.8 电量查询命令 — queryPower

4.9 设备 OTA 升级命令 — otaUpgrade

4.10 设备恢复出厂命令 — factoryReset

5. 上行消息类型详解

5.1 设备注册 — /{identity}/publish/register

5.2 电量上报 — /{identity}/publish/power

5.3 设备离线遗嘱 — /{identity}/publish/status

5.4 按键浇水完成 — /{identity}/publish/finish/key

5.5 排程浇水完成 — /{identity}/publish/finish/schedule

5.6 设备异常告警 — /{identity}/publish/error

5.7 命令 ACK — /{identity}/publish/ack

6. 命令可靠性机制

6.1 命令生命周期

6.2 重试策略

6.3 离线处理

6.4 Redis 键规则

7. 接口 → MQTT 命令映射表

8. 配置参考

9. 扩展指南

新增上行消息处理器

新增下行命令类型

10. 设备在线状态检测

21 KiB

Raw Permalink Blame History

4.1 设备开关命令 — `switchDevice`

4.2 设备绑定命令 — `bindDevice`

4.3 设备初始化/解绑命令 — `initDevice`

4.4 设备编号下发 — `registerDeviceNo`

4.5 排程绑定命令 — `bindSchedule`

4.6 排程解绑命令 — `unbindSchedule`

4.7 自定义命令 — 任意 `commandType`

4.8 电量查询命令 — `queryPower`

4.9 设备 OTA 升级命令 — `otaUpgrade`

4.10 设备恢复出厂命令 — `factoryReset`

5.1 设备注册 — `/{identity}/publish/register`

5.2 电量上报 — `/{identity}/publish/power`

5.3 设备离线遗嘱 — `/{identity}/publish/status`

5.4 按键浇水完成 — `/{identity}/publish/finish/key`

5.5 排程浇水完成 — `/{identity}/publish/finish/schedule`

5.6 设备异常告警 — `/{identity}/publish/error`

5.7 命令 ACK — `/{identity}/publish/ack`