3. 设备动态注册与预注册批量导入对比实践

任务说明

本案例要求使用MQTT协议完成设备与云平台之间的消息发布与订阅。你将编写一个Python客户端，连接到公共MQTT Broker（broker.emqx.io），向主题iot101/dev/status发布设备状态消息，同时订阅主题iot101/dev/command，接收云端下发的控制指令并打印输出。通过本案例，掌握MQTT协议中发布/订阅模型的基本工作方式。

实现方案

使用paho-mqtt客户端库实现。程序启动后同时执行两个动作：订阅命令主题，并周期性发布状态消息。为保证发布与订阅在同一连接上工作，使用loop_start()启动后台网络循环，主线程负责定时发布消息。

代码实现

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

BROKER = "broker.emqx.io"
PORT = 1883
TOPIC_STATUS = "iot101/dev/status"
TOPIC_COMMAND = "iot101/dev/command"
CLIENT_ID = "iot101-device-001"

def on_connect(client, userdata, flags, rc):
    if rc == 0:
        print("连接成功")
        client.subscribe(TOPIC_COMMAND, qos=1)
    else:
        print(f"连接失败，返回码：{rc}")

def on_message(client, userdata, msg):
    payload = msg.payload.decode("utf-8")
    print(f"收到命令：{msg.topic} -> {payload}")

def on_disconnect(client, userdata, rc):
    print("连接断开")

client = mqtt.Client(client_id=CLIENT_ID, protocol=mqtt.MQTTv311)
client.on_connect = on_connect
client.on_message = on_message
client.on_disconnect = on_disconnect

client.connect(BROKER, PORT, keepalive=60)
client.loop_start()

# 主线程循环发布状态消息
try:
    seq = 0
    while True:
        seq += 1
        payload = json.dumps({
            "device_id": CLIENT_ID,
            "seq": seq,
            "status": "online",
            "ts": int(time.time())
        })
        client.publish(TOPIC_STATUS, payload, qos=1)
        print(f"发布状态：{payload}")
        time.sleep(5)
except KeyboardInterrupt:
    print("用户中断，正在退出...")
finally:
    client.loop_stop()
    client.disconnect()

运行与验证

依赖安装：

pip install paho-mqtt

运行方式：

python mqtt_pub_sub.py

验证方法：程序启动后，观察控制台输出。若连接成功，会打印“连接成功”，随后每5秒打印一条“发布状态：{...}”消息。同时，程序已订阅iot101/dev/command主题，当有消息发布到该主题时，会打印“收到命令：...”。

判定标准：

1. 控制台出现“连接成功”，说明TCP连接与MQTT握手正常；
2. 每5秒出现一条状态发布日志，说明发布功能正常；
3. 使用另一个MQTT客户端（如MQTT X或mosquitto_pub）向iot101/dev/command主题发送消息，本程序控制台能打印出对应内容，说明订阅功能正常。

关键说明

本案例使用QoS 1级别确保消息至少送达一次。实际部署时，可将Broker地址替换为私有云平台地址，并在on_connect回调中增加用户名密码认证逻辑。若需验证双向通信，可在另一终端运行mosquitto_sub -h broker.emqx.io -t iot101/dev/status观察设备上报的状态消息。