MQTT 设备接入文档

1. 1
   创建真实设备

   管理员在“设备管理”填写唯一 deviceId、名称和位置。

2. 2
   保存一次性凭据

   创建成功后立即保存 Client ID、用户名、密码及 Topic；关闭窗口后不会再次显示原密码。

3. 3
   安装公共根证书

   从凭据窗口下载 ca.crt，将其配置为 Broker 信任锚。

4. 4
   校准设备时间

   TLS 证书校验依赖正确时间，设备启动后应先完成可靠时间同步。

首次下发 CA 时请通过另一个可信渠道核对指纹；证书轮换后本文档也需要同步更新。

• Topic 区分大小写，Device-001 与 device-001 不相同。
• 生产 ACL 应按发布和订阅方向收紧，不允许设备互相访问。
• 命令 Topic 禁止 Retain，避免设备重连后执行旧命令。
• 当前 PoC 测试账号只允许访问自己的测试前缀；正式设备不得复用。

{
  "ts": 1787649000000,
  "seq": 1024,
  "metrics": {
    "temperature": 24.6,
    "humidity": 61.2,
    "pressure": 101.3
  },
  "status": "online"
}

现场数据来自 Modbus 时，建议由边缘网关完成寄存器读取、基址处理、量纲换算和质量判断，再将标准 JSON 发布至 MQTTS。

推荐遗嘱消息

MQTTX · 模拟设备在线

1. 先断开连接，在连接编辑页展开 Will Message；Topic 填写 devices/aigelun-device-001/status。
2. 遗嘱 Payload 填写 {"status":"offline"}，QoS 选择 1，开启 Retain，然后重新连接。
3. 在发布区使用同一状态 Topic，QoS 选择 1，开启 Retain。
4. 发布 {"status":"online","ts":1787737442000}；实际调试时把 ts 换成当前 Unix 毫秒时间戳。
5. 每 30–60 秒重发在线状态或遥测。正常断开前主动发布带时间戳的 offline；异常断网则由 Broker 发送遗嘱。

Mosquitto CLI · 先订阅

mosquitto_sub \
  -h 47.94.172.189 \
  -p 8883 \
  -V mqttv311 \
  --cafile ./ca.crt \
  -u '<设备账号>' \
  -P '<设备密码>' \
  -i '<deviceId>-debug-sub' \
  -t 'devices/<deviceId>/#' \
  -q 1 -v

Mosquitto CLI · 再发布

mosquitto_pub \
  -h 47.94.172.189 \
  -p 8883 \
  -V mqttv311 \
  --cafile ./ca.crt \
  -u '<设备账号>' \
  -P '<设备密码>' \
  -i '<deviceId>-debug-pub' \
  -t 'devices/<deviceId>/telemetry' \
  -m '{"ts":1787649000000,"metrics":{"temperature":24.6,"humidity":61.2}}' \
  -q 1

Python · paho-mqtt 2.x

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

device_id = os.environ["DEVICE_ID"]
status_topic = f"devices/{device_id}/status"
telemetry_topic = f"devices/{device_id}/telemetry"

client = mqtt.Client(
    callback_api_version=mqtt.CallbackAPIVersion.VERSION2,
    client_id=device_id,
    protocol=mqtt.MQTTv311,
)
client.username_pw_set(
    os.environ["MQTT_USERNAME"],
    os.environ["MQTT_PASSWORD"],
)
client.tls_set(
    ca_certs="ca.crt",
    cert_reqs=ssl.CERT_REQUIRED,
    tls_version=ssl.PROTOCOL_TLS_CLIENT,
)
client.will_set(
    status_topic,
    json.dumps({"status": "offline"}),
    qos=1,
    retain=True,
)

client.connect("47.94.172.189", 8883, keepalive=60)
client.loop_start()
client.publish(
    status_topic,
    json.dumps({"status": "online", "ts": int(time.time() * 1000)}),
    qos=1,
    retain=True,
).wait_for_publish()
client.publish(
    telemetry_topic,
    json.dumps({
        "ts": int(time.time() * 1000),
        "seq": 1,
        "metrics": {"temperature": 24.6, "humidity": 61.2},
    }),
    qos=1,
).wait_for_publish()

Node.js · mqtt

const fs = require('node:fs');
const mqtt = require('mqtt');

const deviceId = process.env.DEVICE_ID;
const statusTopic = `devices/${deviceId}/status`;

const client = mqtt.connect('mqtts://47.94.172.189:8883', {
  protocolVersion: 4,
  clientId: deviceId,
  username: process.env.MQTT_USERNAME,
  password: process.env.MQTT_PASSWORD,
  ca: fs.readFileSync('./ca.crt'),
  rejectUnauthorized: true,
  keepalive: 60,
  reconnectPeriod: 3000,
  will: {
    topic: statusTopic,
    payload: Buffer.from(JSON.stringify({ status: 'offline' })),
    qos: 1,
    retain: true
  }
});

client.on('connect', () => {
  client.publish(
    `devices/${deviceId}/telemetry`,
    JSON.stringify({
      ts: Date.now(),
      seq: 1,
      metrics: { temperature: 24.6, humidity: 61.2 }
    }),
    { qos: 1, retain: false }
  );
});

Node.js 的 protocolVersion: 4 表示 MQTT 3.1.1；使用 MQTT 5.0 时改为 5。

Windows 端口检查

Test-NetConnection 47.94.172.189 -Port 8883

该命令只能检查 TCP 端口，不能验证 MQTT 登录、Topic ACL 或证书。