2. 基于MQTT的边缘消息通信

任务目标

搭建一个MQTT Broker作为消息，实现边缘设备（模拟客户端）与云端（模拟客户端）之间的双向消息通信。边缘设备发布传感器数据到指定主题，云端订阅该主题接收数据；云端下发控制指令到另一主题，边缘设备订阅并响应。

环境与依赖

• 操作系统：Linux / macOS / Windows（均支持）
• Python 3.8+
• 安装依赖：pip install paho-mqtt
• MQTT Broker：使用公共测试Broker broker.emqx.io（端口1883），无需本地部署；如需本地隔离环境，可安装 mosquitto 并启动 mosquitto -p 1883

实现方案

使用 paho-mqtt 客户端库，在单个Python脚本中同时模拟边缘设备端与云端。边缘设备每2秒向 edge/sensor 主题发布一条模拟温度数据；云端订阅该主题并打印接收到的数据。同时云端向 cloud/command 主题发布控制指令，边缘设备订阅该主题并打印收到的指令。通过两个客户端实例在同一进程中运行，完整演示双向通信链路。

关键完整代码

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

BROKER = "broker.emqx.io"
PORT = 1883
SENSOR_TOPIC = "edge/sensor"
COMMAND_TOPIC = "cloud/command"

# ---------- 边缘设备端 ----------
def edge_on_connect(client, userdata, flags, rc):
    print(f"[边缘设备] 已连接，结果码: {rc}")
    client.subscribe(COMMAND_TOPIC)

def edge_on_message(client, userdata, msg):
    print(f"[边缘设备] 收到云端指令: {msg.payload.decode()}")

def edge_publish_loop():
    edge_client = mqtt.Client(client_id="edge-device-001")
    edge_client.on_connect = edge_on_connect
    edge_client.on_message = edge_on_message
    edge_client.connect(BROKER, PORT, 60)
    edge_client.loop_start()
    while True:
        payload = json.dumps({
            "device_id": "edge-001",
            "temperature": round(random.uniform(20.0, 35.0), 2),
            "timestamp": time.time()
        })
        edge_client.publish(SENSOR_TOPIC, payload, qos=1)
        time.sleep(2)

# ---------- 云端 ----------
def cloud_on_connect(client, userdata, flags, rc):
    print(f"[云端] 已连接，结果码: {rc}")
    client.subscribe(SENSOR_TOPIC)

def cloud_on_message(client, userdata, msg):
    data = json.loads(msg.payload.decode())
    print(f"[云端] 收到边缘数据: 设备={data['device_id']}, "
          f"温度={data['temperature']}°C")

def cloud_command_loop():
    cloud_client = mqtt.Client(client_id="cloud-server-001")
    cloud_client.on_connect = cloud_on_connect
    cloud_client.on_message = cloud_on_message
    cloud_client.connect(BROKER, PORT, 60)
    cloud_client.loop_start()
    cmd_id = 1
    while True:
        command = json.dumps({
            "cmd": "read_sensor",
            "cmd_id": cmd_id,
            "params": {"interval": 5}
        })
        cloud_client.publish(COMMAND_TOPIC, command, qos=1)
        print(f"[云端] 已下发指令: {command}")
        cmd_id += 1
        time.sleep(5)

if __name__ == "__main__":
    t1 = threading.Thread(target=edge_publish_loop, daemon=True)
    t2 = threading.Thread(target=cloud_command_loop, daemon=True)
    t1.start()
    t2.start()
    # 主线程保持运行
    while True:
        time.sleep(1)

运行方式

python mqtt_bidirectional.py

验证方法与通过标准

运行脚本后，观察控制台输出。通过标准如下：

• 边缘设备和云端均打印“已连接，结果码: 0”，表示成功接入Broker；
• 每2秒出现一条 [云端] 收到边缘数据 日志，包含设备ID和温度值，证明边缘→云端方向通信正常；
• 每5秒出现一条 [边缘设备] 收到云端指令 日志，包含指令内容，证明云端→边缘方向通信正常；
• 两条方向的消息持续交替出现，无异常断开或报错。

若使用本地mosquitto，将 BROKER 改为 "127.0.0.1" 即可，验证标准不变。

常见问题

• 连接失败：检查网络是否能访问 broker.emqx.io 的1883端口；若公司网络限制外网，改用本地mosquitto。
• 收不到消息：确认订阅主题与发布主题完全一致（包括大小写和层级）；确认QoS设置一致。
• 消息乱序：本案例使用QoS 1保证至少送达一次，若需严格有序可改用QoS 2或增加序列号校验。