1. 边缘设备MQTT连接云平台

任务目标

在边缘设备（以Linux主机模拟）上通过MQTT协议连接IoT云平台（以公共Broker broker.emqx.io:1883模拟），完成一次主题订阅与消息发布，验证双向消息收发链路。

环境依赖

• Python 3.7+，安装 paho-mqtt：pip install paho-mqtt
• 网络可访问 broker.emqx.io（1883端口，无需加密）

完整实现代码

import paho.mqtt.client as mqtt
import time
import threading

BROKER = "broker.emqx.io"
PORT = 1883
TOPIC = "edgeai/device001/data"
CLIENT_ID = "edge_device_001"

received_messages = []

def on_connect(client, userdata, flags, rc):
    if rc == 0:
        print(f"[OK] 已连接Broker: {BROKER}:{PORT}")
        client.subscribe(TOPIC, qos=1)
        print(f"[OK] 已订阅主题: {TOPIC}")
    else:
        print(f"[ERR] 连接失败, 返回码={rc}")

def on_message(client, userdata, msg):
    payload = msg.payload.decode()
    received_messages.append(payload)
    print(f"[RECV] 主题: {msg.topic}, 消息: {payload}")

def on_publish(client, userdata, mid):
    print(f"[SENT] 消息已发布, mid={mid}")

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

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

# 等待连接与订阅完成
time.sleep(2)

# 发布一条测试消息到自身订阅的主题
test_payload = f"hello-from-{CLIENT_ID}-{int(time.time())}"
client.publish(TOPIC, test_payload, qos=1)
print(f"[INFO] 已发送: {test_payload}")

# 等待消息回环
time.sleep(3)

client.loop_stop()
client.disconnect()

if test_payload in received_messages:
    print("[PASS] 消息收发闭环验证通过")
else:
    print("[FAIL] 未收到自身发布的消息，请检查网络或Broker状态")

运行方式

python mqtt_edge_client.py

验证方法与判定标准

• 连接成功：控制台输出 [OK] 已连接Broker，说明TCP与MQTT握手正常。
• 订阅成功：输出 [OK] 已订阅主题，说明已注册该主题的监听。
• 消息发布：输出 [SENT] 消息已发布，说明publish调用已由Broker确认（QoS 1）。
• 消息回环：输出 [RECV] 且内容与发送的 test_payload 一致，说明Broker将消息正确路由回订阅端。
• 最终判定：输出 [PASS] 消息收发闭环验证通过，即完成边缘设备经MQTT与云平台双向通信的完整链路。

常见问题

• 连接超时：检查防火墙是否放行1883端口，或更换网络环境重试。
• 收不到自身消息：确认订阅与发布使用完全相同的主题字符串（含大小写）；公共Broker可能对短时重复连接有频率限制，可等待数秒后重跑。
• QoS说明：本案例使用QoS 1，保证消息至少送达一次；若需严格去重可改用QoS 2，但公共Broker可能增加延迟。