📚 课程目录第4章 MQTT发布订阅实现传感器数据上云

4. MQTT发布订阅实现传感器数据上云

任务目标

本案例实现一个基于MQTT协议的传感器数据上云闭环：周期读取模拟温度/湿度传感器数据，通过MQTT发布到云平台主题；同时订阅云端控制指令主题，收到指令后切换传感器采集状态（启动/停止）。代码采用Python + paho-mqtt实现，可在本地或边缘网关直接运行验证。

环境依赖

Python 3.8+
安装依赖：pip install paho-mqtt
可用的MQTT Broker（如本地Mosquitto，或公共Broker如broker.emqx.io:1883）

完整实现代码

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

# ========== 配置 ==========
BROKER = "broker.emqx.io"   # 替换为实际Broker地址
PORT = 1883
TOPIC_PUB = "iiot/gateway/sensor_data"   # 传感器数据上行主题
TOPIC_SUB = "iiot/gateway/control"       # 控制指令下行主题
CLIENT_ID = "gw_001"
PUB_INTERVAL = 3   # 发布间隔（秒）

# ========== 全局状态 ==========
publish_enabled = True   # 是否允许发布传感器数据（由控制指令切换）

# ========== 传感器模拟 ==========
def read_sensor():
    """模拟读取温度(℃)和湿度(%RH)"""
    temp = round(random.uniform(20.0, 35.0), 1)
    humi = round(random.uniform(40.0, 70.0), 1)
    return {"temperature": temp, "humidity": humi}

# ========== MQTT回调 ==========
def on_connect(client, userdata, flags, rc):
    if rc == 0:
        print(f"[MQTT] 连接成功: {BROKER}:{PORT}")
        client.subscribe(TOPIC_SUB)
        print(f"[MQTT] 已订阅主题: {TOPIC_SUB}")
    else:
        print(f"[MQTT] 连接失败, 返回码={rc}")

def on_message(client, userdata, msg):
    """处理云端下发的控制指令"""
    global publish_enabled
    try:
        payload = json.loads(msg.payload.decode())
        action = payload.get("action")
        if action == "start":
            publish_enabled = True
            print("[控制] 收到启动指令 → 开始发布传感器数据")
        elif action == "stop":
            publish_enabled = False
            print("[控制] 收到停止指令 → 暂停发布传感器数据")
        else:
            print(f"[控制] 未知指令: {action}")
    except Exception as e:
        print(f"[控制] 指令解析失败: {e}")

# ========== 发布线程 ==========
def publish_loop(client):
    """周期发布传感器数据"""
    while True:
        if publish_enabled:
            data = read_sensor()
            data["device_id"] = CLIENT_ID
            data["timestamp"] = int(time.time())
            payload = json.dumps(data)
            client.publish(TOPIC_PUB, payload, qos=1)
            print(f"[发布] {payload}")
        time.sleep(PUB_INTERVAL)

# ========== 主程序 ==========
def main():
    client = mqtt.Client(client_id=CLIENT_ID, protocol=mqtt.MQTTv311)
    client.on_connect = on_connect
    client.on_message = on_message

    try:
        client.connect(BROKER, PORT, keepalive=60)
    except Exception as e:
        print(f"[错误] 无法连接Broker: {e}")
        return

    client.loop_start()   # 启动网络循环（非阻塞）

    # 启动发布线程
    pub_thread = threading.Thread(target=publish_loop, args=(client,), daemon=True)
    pub_thread.start()

    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        print("\n[退出] 正在断开连接...")
        client.loop_stop()
        client.disconnect()

if __name__ == "__main__":
    main()

运行与验证

启动程序：python mqtt_sensor.py
观察上行数据：控制台每3秒打印一条[发布]日志，包含温度、湿度、设备ID和时间戳，说明传感器数据已成功发布到iiot/gateway/sensor_data主题。
验证下行控制：另开一个终端，使用MQTT客户端（如mosquitto_pub）发送控制指令：

mosquitto_pub -h broker.emqx.io -t iiot/gateway/control -m '{"action":"stop"}'

此时原程序控制台应打印[控制] 收到停止指令 → 暂停发布传感器数据，且不再输出[发布]日志。再发送{"action":"start"}，发布恢复。

通过条件：上述发布/停止/恢复的完整流程在控制台日志中清晰可见，即证明MQTT发布订阅双向通信闭环成功。

关键说明

本案例使用模拟传感器数据，实际接入时只需替换read_sensor()函数为真实传感器读取逻辑（如Modbus、I2C等），其余代码无需改动。
Broker地址默认为公共测试Broker（broker.emqx.io），生产环境请替换为实际云平台MQTT接入地址，并配置用户名/密码认证（在client.username_pw_set()中添加）。
QoS设为1确保消息至少送达一次，可根据业务需求调整。