MQTT 设备接入

MQTT 是名尚智联的主协议，对接外置 EMQX 5.x Broker。设备通过 Serial + AuthCode 认证，Broker 配置 Webhook 回调后端 /api/emqx/auth 完成身份校验。

---

一、认证机制

1.1 连接参数

参数	值	说明
Broker 地址	tcp://<host>:1884	EMQX MQTT TCP
WebSocket	ws://<host>:18084/mqtt	EMQX WS
ClientID	{serial}_{随机ID}	如 ESP32-001_a1b2c3
Username	{serial}	设备序列号
Password	{authCode}	注册设备时返回的认证码

1.2 认证流程

设备 → EMQX CONNECT (serial, authCode)
  → EMQX Webhook POST /api/emqx/auth
    → 后端查 ms_device 表
      → serial 匹配 & authCode 一致 → 允许 (returncode 0)
      → 不匹配 → 拒绝 (returncode 5)

后端白名单已配置，/api/emqx/** 走 Webhook 自验，无需 Sa-Token。

---

二、Topic 规范

2.1 设备上报

Topic	方向	Payload
device/{serial}/properties	设备 → 平台	{"props":{"temperature":25.6}}
device/{serial}/events	设备 → 平台	{"event":"alarm","data":{"level":"high"}}
device/{serial}/telemetry	设备 → 平台（心跳）	{"online":true,"uptime":86400}

2.2 平台下发

Topic	方向	Payload
device/{serial}/command	平台 → 设备	{"cmd":"restart","id":"xxx"}
device/{serial}/command/reply	设备 → 平台	{"id":"xxx","status":"ok"}
device/{serial}/ota	平台 → 设备	{"url":"https://...firmware.bin","version":"1.2.3"}

2.3 通配订阅

device/+/properties     # 所有设备属性上报
device/+/events          # 所有设备事件
device/{serial}/#        # 单设备全部主题

---

三、ESP32 示例（Arduino）

#include <WiFi.h>
#include <PubSubClient.h>

const char* ssid = "YourWiFi";
const char* password = "wifi_password";

// MQTT 凭据 — 从管理后台设备详情页获取
const char* mqtt_server = "your-server.com";
const int   mqtt_port   = 1884;
const char* device_serial = "ESP32-001";
const char* device_auth   = "a1b2c3d4e5f6...";   // 注册设备时返回的 authCode

WiFiClient espClient;
PubSubClient client(espClient);

void setup() {
  Serial.begin(115200);
  WiFi.begin(ssid, password);
  while (WiFi.status() != WL_CONNECTED) delay(500);

  client.setServer(mqtt_server, mqtt_port);
  client.setCallback(callback);
  reconnect();
}

void reconnect() {
  String clientId = String(device_serial) + "_" + String(random(0xFFFF), HEX);
  while (!client.connected()) {
    if (client.connect(clientId.c_str(), device_serial, device_auth)) {
      client.subscribe(("device/" + String(device_serial) + "/command").c_str());
    } else {
      delay(2000);
    }
  }
}

void callback(char* topic, byte* payload, unsigned int len) {
  Serial.printf("[recv] %s → %s\n", topic, String((char*)payload).c_str());
  // 解析命令、执行操作、回复 command/reply
}

void loop() {
  if (!client.connected()) reconnect();
  client.loop();

  // 上报属性（每 10 秒）
  static unsigned long last = 0;
  if (millis() - last > 10000) {
    float temp = 25.0 + analogRead(34) * 0.01;  // 模拟
    float humid = 60.0;
    String payload = "{\"props\":{\"temperature\":" + String(temp, 1)
                   + ",\"humidity\":" + String(humid, 1) + "}}";
    client.publish(("device/" + String(device_serial) + "/properties").c_str(), payload.c_str());
    last = millis();
  }
}

3.1 依赖库

platformio.ini:
  lib_deps = knolleary/PubSubClient@^2.8
             bblanchon/ArduinoJson@^7.0

---

四、Python 示例

import json
import time
import paho.mqtt.client as mqtt

SERIAL = "ESP32-001"
AUTH   = "a1b2c3d4e5f6..."
BROKER = "your-server.com"
PORT   = 1884

def on_connect(client, userdata, flags, rc):
    client.subscribe(f"device/{SERIAL}/command")
    print(f"connected rc={rc}")

def on_message(client, userdata, msg):
    print(f"[{msg.topic}] {msg.payload.decode()}")

client = mqtt.Client(client_id=f"{SERIAL}_py", protocol=mqtt.MQTTv311)
client.username_pw_set(SERIAL, AUTH)
client.on_connect = on_connect
client.on_message = on_message
client.connect(BROKER, PORT, 60)
client.loop_start()

while True:
    payload = json.dumps({"props": {"temperature": 25.6, "humidity": 62.3}})
    client.publish(f"device/{SERIAL}/properties", payload)
    time.sleep(10)

---

五、EMQX 配置（服务端）

5.1 Webhook

在 EMQX Dashboard → 认证 → Webhook 中添加：

字段	值
URL	http://backend:8080/api/emqx/auth
Method	POST
Content-Type	application/json

5.2 ACL 规则

# 只允许设备发布/订阅自己的 Topic
{allow, {user, _}, publish, ["device/${username}/+"]}.
{allow, {user, _}, subscribe, ["device/${username}/#"]}.
{deny, all}.

5.3 心跳建议

• MQTT Keep-Alive：300 秒
• 平台在线状态超时：360 秒（略大于 Keep-Alive）

---

六、常见问题

问题	排查
CONNACK rc=5（认证拒绝）	serial 或 authCode 不匹配，检查设备是否存在、authCode 是否正确
CONNACK rc=3（Broker 不可用）	检查 EMQX 是否运行、端口是否可达
上报后 Redis 无数据	检查 Topic 是否为 device/{serial}/properties、payload 是否含 props 字段
连接频繁断开	检查 Keep-Alive、网络稳定性、ClientID 是否冲突