MQTT 服务设计

后端 MQTT 客户端的模块设计、消息路由机制和处理管道。MQTT 协议定义见 06.MQTT通信协议，EMQX 配置见 09.EMQX配置与安全。

---

1. MQTT Client 初始化

使用 mqtt.js 库，在 MqttModule 的 onModuleInit 生命周期中创建客户端实例。

连接参数

参数	值	说明
brokerUrl	mqtt://emqx:1883（开发）/ mqtts://emqx:8883（生产）	EMQX 地址
clientId	backend-{hostname}-{pid}	后端服务标识，用于 EMQX 区分不同实例
username	backend_admin	后端管理账号，EMQX 侧需配置 PostgreSQL 认证
password	环境变量注入	后端 MQTT 连接凭证
clean	true	不保留离线消息（后端不依赖持久化会话）
keepalive	60 秒	心跳间隔
reconnectPeriod	1s → 2s → 4s → 最大 30s	断线重连指数退避

TLS 配置（生产环境）

{
  ca: fs.readFileSync('/etc/emqx/certs/ca.pem'),
  rejectUnauthorized: true,
}

---

2. 装饰器式 Topic 路由

2.1 装饰器定义

三个核心装饰器：

装饰器	作用域	说明
@MqttController(prefix)	类	声明 Topic 前缀，如 sys
@SubscribeTopic(pattern)	方法	声明 Topic 模式，支持通配符 + 和 #
@MqttContext() / @Payload()	参数	注入 MQTT 上下文对象 / 解析后的消息体

2.2 路由注册流程

1. 应用启动 → MqttModule.onModuleInit() 初始化 MQTT Client
2. MqttExplorer 扫描所有带 @MqttController 的类
3. 提取每个 @SubscribeTopic 方法 → 拼接完整 Topic：{prefix}/{pattern}
4. 订阅时追加共享订阅前缀：$share/group1/{prefix}/{pattern}
5. MQTT Client 收到消息 → 匹配 Topic → 调用对应方法 → 传入解析后的 Payload 和 Context

共享订阅确保多 NestJS 实例负载均衡，详见 09.EMQX配置与安全 第 4 节。设备认证由 EMQX PostgreSQL 认证插件直连数据库完成，后端无需实现 /api/v1/devices/auth 或 /api/v1/devices/acl 端点。

2.3 路由表

订阅 Topic（共享订阅前缀省略）	处理方法	说明
sys/+/+/thing/event/property/post	handlePropertyPost	属性上报
sys/+/+/thing/service/+/set_reply	handleSetReply	控制指令回复
sys/+/+/thing/config/set_reply	handleConfigReply	配置下发回复
sys/+/+/thing/ota/progress	handleOtaProgress	OTA 进度上报
sys/+/+/thing/event/+/post	handleEventPost	设备事件（告警等）

2.4 Context 解析

MqttContext 提供以下信息：

方法	返回值	说明
getTopicParams()	{ productId, deviceId }	从 Topic 中提取 +/+ 位置的变量值
getTopic()	string	完整 Topic 字符串
getClientId()	string	MQTT ClientId（仅在需要物理 ID 时使用）

productId 和 deviceId 从 Topic 中按 +/+ 占位顺序提取：sys/{productId}/{deviceId}/thing/...。

---

3. 消息处理管道

从 MQTT 消息到业务逻辑的完整处理链：

MQTT 消息到达
  │
  ├─→ Topic 匹配 → 路由到 @SubscribeTopic 方法
  │     │
  │     ├─→ Payload 校验
  │     │     ├─ JSON.parse
  │     │     ├─ 验证必填字段（msgId、params）
  │     │     └─ 格式不符合 → 记录日志，丢弃（不回发 MQTT 错误）
  │     │
  │     ├─→ 设备身份校验
  │     │     ├─ 从 Topic 提取 deviceId
  │     │     └─ 查 PostgreSQL 确认设备存在且状态正常
  │     │
  │     └─→ 业务处理
  │           ├─ 属性上报 → DeviceService.updateShadow()
  │           ├─ set_reply → DeviceService.confirmCommand()
  │           ├─ config_reply → DeviceService.confirmConfig()
  │           └─ ota_progress → OtaService.updateProgress()
  │
  └─→ 异常 → 记录日志 + 错误计数器 + 告警（连续失败超过阈值）

校验失败不回发 MQTT 错误消息——设备端不会订阅错误 Topic，回发只会造成额外的消息风暴。

---

4. 指令下发

后端主动向设备发布消息，通过 MqttService.publish() 方法封装。

下发 Topic 对照

Topic	场景	触发时机
sys/{pid}/{did}/thing/service/property/set	用户控制设备	小程序/Web 下发控制指令
sys/{pid}/{did}/thing/config/set	注册后下发配置	设备首次注册激活时
sys/{pid}/{did}/thing/ota/upgrade	OTA 升级通知	运维触发固件升级

发布参数

参数	默认值	说明
topic	必填	完整 Topic 路径
payload	必填	JSON 对象，自动序列化
qos	1	至少一次送达
retain	false	指令不需要保留消息

---

5. 断线重连与容错

策略	配置
重连间隔	指数退避：1s → 2s → 4s → 8s → 最大 30s，无限重试
重连后操作	自动重新订阅所有 @SubscribeTopic 规则
EMQX 宕机	MQTT Client 持续重连，HTTP API 正常响应不受影响
消息积压	clean: true 不保留离线消息；设备端应缓存关键消息并在 Broker 恢复后补发

---

6. 与设备状态管理的集成

MQTT 消息到达后与设备状态的联动（引用 02.后台服务设计 3.2 节已修复的内容）：

属性上报路径：

handlePropertyPost → DeviceService.updateShadow
  ├── Redis: HSET device:shadow:{id} (热缓存，WebSocket 推送用)
  ├── PostgreSQL: UPDATE shadow JSONB (权威数据)
  └── InfluxDB: 异步写入时序数据

指令回复路径：

handleSetReply → DeviceService.confirmCommand
  ├── 根据 msgId 查找指令记录
  ├── 更新指令状态为完成
  └── WebSocket 推送执行结果至前端

完整的两级存储设计原因和同步策略见 12.缓存策略 第 3 节。

---

7. 目录结构

src/gateways/mqtt/
├── mqtt.module.ts          # 动态模块，onModuleInit 初始化客户端
├── mqtt.service.ts         # 发布消息、获取客户端实例
├── mqtt.decorators.ts      # @SubscribeTopic, @MqttController, @Payload, @MqttContext
├── mqtt.explorer.ts        # 路由扫描器，发现并注册订阅
├── mqtt.interface.ts       # MqttContext 接口定义
└── handlers/               # 按业务拆分的消息处理器
    ├── property.handler.ts  # 属性上报
    ├── command.handler.ts   # 控制回复
    ├── config.handler.ts    # 配置回复
    └── ota.handler.ts       # OTA 进度