WebSocket 实时通信

前端（小程序/Web 后台）通过 WebSocket 接收设备实时状态推送的架构设计。

1. 整体架构

数据流向：

设备 MQTT 上报 → 后端 MQTT Client 收到 → 数据处理 → 查询路由表 → WebSocket 推送给前端

三个关键角色：

角色	职责
设备（生产者）	通过 MQTT 上报属性数据，触发状态变更事件
后端（桥接器）	维护 MQTT 订阅 + WebSocket 连接池，将设备消息路由到正确的用户
前端（消费者）	小程序或 Web 后台，通过 WebSocket 接收推送，无需轮询

设备上报和上下线感知的时序图见 02.核心业务流程。

---

2. 连接建立与认证

连接端点

环境	端点
开发	ws://host:3000/ws
生产	wss://host/ws

认证方式

连接时携带 JWT Token，支持两种传递方式：

方式	示例	适用场景
URL 参数	ws://host/ws?token=eyJhbG...	简单直接，小程序推荐
首条消息	{"type":"auth","token":"eyJhbG..."}	Token 较长时使用

后端验证流程

1. 解析 JWT 获取 userId，验证 Token 有效性
2. 验证通过：将 WebSocket 连接与用户绑定，存入连接池 Map<userId, WebSocket[]>
3. 验证失败：服务端主动关闭连接，返回状态码 4001

一个用户可同时持有多个 WebSocket 连接（如手机小程序 + PC 浏览器），所有连接均收到相同的设备推送。

---

3. 消息格式

统一格式（服务端 → 客户端）

{
  "type": "device.status | device.shadow",
  "deviceId": "d1a2b3c4-e5f6-7890-abcd-ef1234567890",
  "payload": { },
  "timestamp": "2026-05-27T10:00:00Z"
}

字段	类型	说明
type	string	消息类型
deviceId	string	设备逻辑 ID
payload	object	具体数据，不同 type 结构不同
timestamp	string	服务端 ISO 8601 时间戳

消息类型

device.status — 设备上下线通知：

{
  "type": "device.status",
  "deviceId": "d1a2b3c4-...",
  "payload": { "online": true },
  "timestamp": "2026-05-27T10:00:00Z"
}

device.shadow — 设备影子变更（属性值变化）：

{
  "type": "device.shadow",
  "deviceId": "d1a2b3c4-...",
  "payload": { "relay_1": true, "relay_2": false, "temperature": 45.2 },
  "timestamp": "2026-05-27T10:00:05Z"
}

客户端 → 服务端

type	方向	说明	payload
auth	客户端→服务端	认证	{"token":"eyJhbG..."}
subscribe	客户端→服务端	订阅设备	{"deviceIds":["d1...","d2..."]}
unsubscribe	客户端→服务端	取消订阅	{"deviceIds":["d1..."]}

---

4. 设备订阅与路由

这是最核心的部分，解决"一个用户怎样只收到自己设备的消息"。

路由表结构

后端维护两张映射表：

结构	说明
Map<userId, Set<webSocket>>	用户连接池（已由认证流程维护）
Map<deviceId, Set<userId>>	设备→用户路由表，记录哪些用户正在关注哪些设备

订阅流程

1. 前端发起订阅：用户打开设备详情页或首页时，前端发送：

   {"type":"subscribe","deviceIds":["d1a2b3c4-...","d2b3c4d5-..."]}

2. 后端更新路由表：将 userId 加入每个 deviceId 对应的关注集合

3. MQTT 消息到达时：

   • 解析消息获取 deviceId
   • 查路由表，获取关注该设备的 userId 列表
   • 遍历所有 userId，找到其所有 WebSocket 连接，逐个发送

4. 取消订阅：用户离开页面时：

   {"type":"unsubscribe","deviceIds":["d1a2b3c4-..."]}

   后端将 userId 从对应 deviceId 的关注集合中移除

5. 断开清理：用户 WebSocket 断开时，自动清理该用户的所有路由表条目

路由示意

设备 d1 上报温度 → 路由表查询 → userId=[A, B]
                                 → A 的 ws 连接 × 2（小程序+PC）
                                 → B 的 ws 连接 × 1（小程序）
                                 → 共发送 3 条推送消息

---

5. 断线重连与心跳

心跳（服务端）

参数	值	说明
Ping 间隔	30 秒	服务端发送 Ping 帧
超时判定	90 秒	未收到任何响应则断开连接
断开处理	立即	关闭 WebSocket，清理该用户路由表

浏览器和微信小程序会自动响应 Ping 帧（pong），无需手动处理。

重连（客户端）

前端需要实现自动重连策略：

重连次数	等待时间
第 1 次	1 秒
第 2 次	2 秒
第 3 次	4 秒
第 4 次	8 秒
第 5 次及以后	30 秒（最大间隔）

重连成功后需要：

1. 重新发送认证消息（若 Token 已过期，先调用刷新接口）
2. 重新发送当前页面的设备订阅消息

---

6. Nginx 反向代理配置

若前端通过 Nginx 访问后端，需配置 WebSocket 代理：

location /ws {
    proxy_pass http://backend:3000;
    proxy_http_version 1.1;
    proxy_set_header Upgrade $http_upgrade;
    proxy_set_header Connection "upgrade";
    proxy_set_header Host $host;
    proxy_set_header X-Real-IP $remote_addr;
    proxy_read_timeout 3600s;
    proxy_send_timeout 3600s;
}

关键点：proxy_read_timeout 需设置为较长值（如 3600 秒），否则 Nginx 会在默认 60 秒后切断 WebSocket 长连接。

---

7. 连接数估算与扩展

单实例容量

单 NestJS 实例（Node.js 单线程）可承载 5000 个 WebSocket 并发连接，对应 非功能需求 中的性能指标。

水平扩展

组件	策略
Nginx / 负载均衡	使用 ip_hash 保证同一用户连到同一实例，简化路由表维护
NestJS 多实例	共享 Redis 存储用户/设备路由表，任一实例收到 MQTT 消息均可查询完整路由
MQTT 共享订阅	多实例间通过 $share/group1/ 分摊设备消息

技术选型

方案	优点	缺点	建议
Socket.IO（@nestjs/websockets）	内置房间（room）机制、自动重连、降级兼容	额外协议开销，非标准 WebSocket	推荐，房间机制直接映射"设备→房间→用户"
原生 ws 库	轻量、标准协议	需自行实现路由、重连、心跳	适合极简场景

Socket.IO 的 room 机制天然适合本场景：

• socket.join(deviceId) → 用户加入设备房间
• io.to(deviceId).emit(...) → 向关注该设备的所有用户推送
• 无需自行维护 Map<deviceId, Set<userId>> 路由表