跳至主要内容

联调与验证

大约 3 分钟

联调与验证

使用 MQTTX / mosquitto

MQTTX：Host、Port 按控制台返回；MQTT Version 选 3.1.1；Client ID / Username / Password 直接粘贴控制台「获取连接信息」返回的三个值。平台不校验 timestamp 是否过期，但 password 与该 timestamp 一一绑定，三者必须成对使用，手改 clientId 中的时间戳会导致鉴权失败。

mosquitto 命令行（Linux/macOS；一次算好凭证后复用，clientId 里的 timestamp 必须与 password 来自同一次计算）：

PK=pk12345; DN=dev001; DS=Secret123456
HOST=mqtt.xiaolizi.tech; PORT=2883

TS=$(python -c 'import time;print(int(time.time()*1000))')     # 13 位毫秒；Linux 也可用 date +%s%3N
MAIN="$PK.$DN"
CONTENT="clientId${MAIN}deviceName${DN}productKey${PK}timestamp${TS}"
CID="${MAIN}|timestamp=${TS},signmethod=hmacsha256|"
USER="${DN}&${PK}"
PASS=$(python -c 'import hmac,hashlib,sys;print(hmac.new(sys.argv[1].encode(),sys.argv[2].encode(),hashlib.sha256).hexdigest().upper())' "$DS" "$CONTENT")

# 订阅下行
mosquitto_sub -h "$HOST" -p "$PORT" -i "$CID" -u "$USER" -P "$PASS" -v -q 1 \
  -t "/sys/$PK/$DN/thing/service/#" \
  -t "/ext/rrpc/+/sys/$PK/$DN/thing/service/+" \
  -t "/sys/$PK/$DN/thing/event/+/post_reply" \
  -t "/shadow/get/$PK/$DN" \
  -t "/ota/device/upgrade/$PK/$DN"

# 另开一个终端，重复上面的变量赋值后再上报属性
mosquitto_pub -h "$HOST" -p "$PORT" -i "$CID" -u "$USER" -P "$PASS" -q 1 \
  -t "/sys/$PK/$DN/thing/event/property/post" \
  -m '{"id":"1","sys":{"ack":1},"params":{"temperature":23.6},"method":"thing.event.property.post"}'

注意：-i 中的 timestamp 变化后 password 必须同步重算，否则鉴权失败；因此不要在两个终端里各算各的。

TLS 场景把 -p 改为 9883、-h 改为 mqtts 域名，并增加 --cafile ca.pem。

Python 端到端示例

依赖：pip install paho-mqtt>=2.0

import hmac, hashlib, json, time, sys
import paho.mqtt.client as mqtt

PK, DN, SECRET = "pk12345", "dev001", "Secret123456"
HOST, PORT, KEEPALIVE = "mqtt.xiaolizi.tech", 2883, 120

ts = str(int(time.time() * 1000))
content = f"clientId{PK}.{DN}deviceName{DN}productKey{PK}timestamp{ts}"
client_id = f"{PK}.{DN}|timestamp={ts},signmethod=hmacsha256|"
username = f"{DN}&{PK}"
password = hmac.new(SECRET.encode(), content.encode(), hashlib.sha256).hexdigest().upper()

TOPIC_PROP = f"/sys/{PK}/{DN}/thing/event/property/post"
SUBSCRIBE = [
    (f"/sys/{PK}/{DN}/thing/service/property/set", 1),
    (f"/sys/{PK}/{DN}/thing/service/+", 1),
    (f"/ext/rrpc/+/sys/{PK}/{DN}/thing/service/+", 1),
    (f"/sys/{PK}/{DN}/thing/event/+/post_reply", 1),
    (f"/shadow/get/{PK}/{DN}", 1),
    (f"/ota/device/upgrade/{PK}/{DN}", 1),
]


def on_connect(client, userdata, flags, reason_code, properties=None):
    print("CONNACK:", reason_code)
    if reason_code == 0:
        client.subscribe(SUBSCRIBE)
        client.publish(TOPIC_PROP, json.dumps({
            "id": "1", "version": "1.0", "sys": {"ack": 1},
            "params": {"temperature": 23.6},
            "method": "thing.event.property.post"}), qos=1)


def on_message(client, userdata, msg):
    print("RECV", msg.topic, msg.payload.decode())
    data = json.loads(msg.payload.decode() or "{}")
    mid = data.get("id")
    if msg.topic.endswith("thing/service/property/set"):
        client.publish(f"/sys/{PK}/{DN}/thing/service/property/set_reply",
                       json.dumps({"id": mid, "messageId": mid, "success": True}), qos=1)
    elif "/ext/rrpc/" in msg.topic:
        # 同步服务调用：必须回到同一个 topic
        client.publish(msg.topic, json.dumps({"id": mid, "code": 200, "message": "success",
                                             "params": {"result": "ok"}}), qos=1)
    elif msg.topic.startswith(f"/shadow/get/"):
        if data.get("method") in ("reply", "control"):
            pass  # 对比 desired / reported 后执行


c = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=client_id,
                protocol=mqtt.MQTTv311, clean_session=True)
c.username_pw_set(username, password)
c.on_connect = on_connect
c.on_message = on_message
c.connect(HOST, PORT, KEEPALIVE)
c.loop_forever(retry_first_connection=False)

用后台接口手动注入下行消息

联调时可绕过业务入口，直接往设备 Topic 灌一条报文，验证设备的订阅与解析逻辑：

curl -X POST 'https://<平台地址>/admin-api/iot/device-info/publish' \
  -H 'Authorization: Bearer <后台登录token>' \
  -H 'tenant-id: 1' \
  -H 'Content-Type: application/json' \
  -d '{"deviceId": 1024, "topic": "/sys/pk12345/dev001/thing/service/property/set", "qos": 0,
       "msgContent": "{\"id\":\"debug-1\",\"timestamp\":1700000000000,\"params\":{\"Power\":\"on\"},\"method\":\"thing.service.property.set\"}"}'

deviceId 用于定位设备（调用方需具备 iot:device-info:publish 权限），topic 为完整下行 Topic，qos 由调用方指定，msgContent 是原始字符串内容。

同名的遗留接口 POST /admin-api/iot/device-info/device/publish 走 Mica Redis 流，在当前 EMQX 架构下不会真正下发，请勿使用。

验收清单

#检查项预期
1CONNACK0，后台显示设备在线
2属性上报（ack=1）收到 post_reply code:200，后台属性记录有值
3故意上报未定义 identifier属性记录无该键，云端日志出现 5092/6308
4事件上报事件记录可见，类型与物模型一致
5后台设置属性设备收到 property/set，params 与下发一致
6后台同步调用服务后台返回设备 params 输出；设备超时则报 2011000003
7影子上报与拉取method=reply，版本号符合预期
8断线重连重新订阅并拉影子，desired 差异被执行
9OTA：版本上报后台显示当前版本
10OTA：批次推送与进度收到 code:"1000" 指令，进度更新、最终 inform 新版本后批次置为成功
11本地日志thing/log/post 有回复且后台可见
12停用设备在线会话被踢掉，重连返回未授权
小李子科技