示例代码

本页提供可直接运行的接入示例，覆盖最常见的开发场景。所有示例只需替换令牌与地址即可运行。

• 公有云 SaaS 地址：cloudbridge.qiuhuasoft.top，端口 1883
• 令牌：设备详情页复制
• 依赖：Python 需 pip install paho-mqtt>=2.0；Node.js 需 npm install mqtt

一、Python：直连设备持续上报

import json, time
import paho.mqtt.client as mqtt          # pip install paho-mqtt>=2.0

TOKEN  = "设备 Access Token"
BROKER = "平台 MQTT 地址"

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=TOKEN)
client.username_pw_set(TOKEN)            # 用户名 = 令牌，密码留空
client.connect(BROKER, 1883, keepalive=60)
client.loop_start()

while True:
    temperature = 25 + (time.time() % 10)   # 换成真实采集值
    payload = {
        "ts": int(time.time() * 1000),
        "values": {"temperature": round(temperature, 1)},
    }
    client.publish("v1/devices/me/telemetry",
                   json.dumps(payload), qos=1)
    print("上报:", payload)
    time.sleep(10)

二、Node.js：直连设备上报 + 属性

import mqtt from "mqtt";                 // npm install mqtt

const TOKEN = "设备 Access Token";
const client = mqtt.connect(`mqtt://平台MQTT地址:1883`, {
  username: TOKEN,        // 用户名 = 令牌
  password: "",           // 密码留空
  clientId: TOKEN,        // 保证唯一
});

client.on("connect", () => {
  console.log("已连接平台");

  // 每 10 秒上报一次遥测
  setInterval(() => {
    const payload = {
      ts: Date.now(),
      values: { temperature: +(25 + Math.random() * 5).toFixed(1) },
    };
    client.publish("v1/devices/me/telemetry", JSON.stringify(payload), { qos: 1 });
  }, 10_000);

  // 上报一次属性（低频状态信息）
  client.publish(
    "v1/devices/me/attributes",
    JSON.stringify({ firmwareVersion: "1.0.0", location: "1号车间" }),
    { qos: 1 },
  );
});

三、Python：网关 + 子设备

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

GATEWAY_TOKEN = "网关设备的 Access Token"

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=GATEWAY_TOKEN)
client.username_pw_set(GATEWAY_TOKEN)
client.connect("平台 MQTT 地址", 1883, keepalive=60)
client.loop_start()

now = int(time.time() * 1000)

# 1) 通知子设备上线（每台子设备首次上报前发一次）
client.publish("v1/gateway/connect", json.dumps({"device": "Device A"}), qos=1)
client.publish("v1/gateway/connect", json.dumps({"device": "Device B"}), qos=1)

# 2) 一条报文带多台子设备的遥测
client.publish("v1/gateway/telemetry", json.dumps({
    "Device A": [{"ts": now, "values": {"temperature": 42, "humidity": 55}}],
    "Device B": [{"ts": now, "values": {"temperature": 38}}],
}), qos=1)

# 3) 子设备属性
client.publish("v1/gateway/attributes", json.dumps({
    "Device A": {"firmwareVersion": "2.1"},
}), qos=1)

前提：Device A、Device B 已在平台建档且挂在网关下，外部 ID 与报文键名完全一致。

四、Python：接收下行 RPC 指令

import json
import paho.mqtt.client as mqtt

TOKEN = "设备 Access Token"

def on_message(c, userdata, msg):
    instruction = json.loads(msg.payload.decode())
    print("收到指令:", instruction)      # {"method": ..., "params": {...}}
    if instruction.get("method") == "reboot":
        print("执行重启……")              # 换成真实的设备控制逻辑

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=TOKEN)
client.username_pw_set(TOKEN)
client.connect("平台 MQTT 地址", 1883, keepalive=60)
client.on_message = on_message

client.subscribe(f"v1/devices/{TOKEN}/rpc/request/+", qos=1)  # 订阅自己的 RPC 主题
client.loop_forever()

网关侧把订阅主题换成 v1/gateway/rpc/{网关令牌}，报文里按 device 字段路由到子设备，见《设备 RPC》。

五、用 MQTT 客户端工具调试

没有设备也能先跑通链路——用 MQTTX 等图形化客户端模拟一台设备：

1. 新建连接：Host 填平台地址，Port 1883，Username 填设备令牌，Password 留空；

2. 连接成功后，向 v1/devices/me/telemetry 发布一条：

   { "ts": 1735689600000, "values": { "temperature": 30 } }

3. 到平台的设备详情 → 实时数据查看是否出现 temperature 指标；

4. 再测试网关主题（用网关令牌连接）与属性主题。

调试建议：先用图形工具把"能连上、数据能进来"验证通过，再写设备端代码——绝大多数接入问题出在令牌、地址与主题拼写上，工具调试能把问题隔离在配置层。

六、嵌入式环境移植要点

环境	要点
ESP32 / ESP8266	PubSubClient / AsyncMqttClient 均可；令牌作为用户名，注意字符串长度
STM32 + 模组（如 EC800）	AT 指令的 MQTT 登录填令牌到用户名字段；JSON 组包注意转义
网关（Linux）	直接用 paho / mqtt.js / mosquitto_pub；注意断线重连与 Client ID 唯一

---

下一步：接入后的平台侧操作与协议边界，回到《接入总览》；REST 接口见《API 概览》。