Files
water/docs/mqtt-spec.md
2026-07-13 14:22:11 +08:00

15 KiB
Raw 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/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 命令执行确认

{identity} 可以是设备编号deviceNo或 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/cmddeviceNo 统一为小写格式。


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": "命令类型",
  // ... 其他业务字段
}

3.2 上行 ACK 结构DeviceCommandAck

{
  "commandId": "对应下发命令的commandId",
  "status": "1",
  "message": "可选的文本消息"
}

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


4. 下行命令类型详解

4.1 设备开关命令 — switchDevice

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

{
  "commandId": "xxx",
  "commandType": "switchDevice",
  "deviceNo": "01",
  "cmd": "1",
  "startTime": "2026-06-25 14:35:00",
  "durationMin": 20
}
字段 类型 说明
cmd String "1" = 开启浇水,"0" = 停止浇水
startTime String 开始时间(yyyy-MM-dd HH:mm:ssHH: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",
  "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",
  "scheduleId": 1,
  "unbind": true
}
字段 类型 说明
scheduleId Long 要解绑的排程 ID
unbind Boolean true = 解绑

4.7 自定义命令 — 任意 commandType

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

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

5. 上行消息类型详解

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

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

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

服务端处理:

  1. 通过 MAC 或设备编号解析入库设备
  2. 更新设备注册信息(名称、电量、固件版本等)
  3. 刷新设备在线状态到 Redis
  4. 自动回复设备编号(registerDeviceNo 命令)

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

{
  "powerLevel": "78"
}

服务端处理: 更新设备电量 + 刷新在线状态。


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

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

服务端处理:

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

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

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

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


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

{
  "errorCode": "E001",
  "message": "水泵故障"
}

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


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

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

服务端处理:

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

6. 命令可靠性机制

6.1 命令生命周期

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

6.2 重试策略

配置项 默认值 说明
maxRetryCount 3 最大重试次数
retryIntervalMs 5000 重试间隔ms
scanIntervalMs 5000 定时扫描间隔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 /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
    device-status-cache-prefix: "mqtt:device:status:"
    device-status-cache-ttl-seconds: 300

9. 扩展指南

新增上行消息处理器

  1. 实现 MqttTopicHandler 接口
  2. 添加 @Component 注解
  3. 定义 topicPattern() 返回匹配正则(必须包含一个捕获组提取设备标识)
  4. 实现 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 后自动发现。

新增下行命令类型

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

10. 设备在线状态检测

设备在线状态通过 Redis 缓存管理:

  • 写入时机: 设备注册、电量上报、ACK 确认时刷新
  • 缓存格式: { "deviceNo": "01", "status": "1", "lastReportTime": "2026-06-25T14:30:00Z" }
  • TTL: 默认 300 秒5 分钟)
  • 离线判定: 缓存过期 = 设备离线
  • 并发保护: 使用分布式锁(lock:mqtt:device:status:{deviceNo})防止并发写入