📚 课程目录第3章基于MQTT的物联网数据监控

3. 基于MQTT的物联网数据监控

任务说明

本案例要求实现一个基于MQTT协议的物联网数据监控系统。系统需完成：设备端模拟温度传感器数据发布、MQTT Broker消息中转、订阅端实时接收并解析数据、异常温度报警判定与输出。最终通过命令行可观测的日志验证完整数据链路。

系统架构

实现代码

项目包含三个文件：publisher.py（设备模拟）、subscriber.py（监控报警）、requirements.txt（依赖）。使用paho-mqtt客户端库实现MQTT通信。

# requirements.txt
paho-mqtt==1.6.1

# publisher.py - 模拟温度传感器设备
import json
import random
import time
import paho.mqtt.client as mqtt

BROKER = "localhost"
PORT = 1883
TOPIC = "sensor/temp"
CLIENT_ID = "device_sim_001"

def on_connect(client, userdata, flags, rc):
    if rc == 0:
        print(f"[设备] 已连接Broker: {BROKER}:{PORT}")
    else:
        print(f"[设备] 连接失败, code={rc}")

client = mqtt.Client(client_id=CLIENT_ID)
client.on_connect = on_connect
client.connect(BROKER, PORT, keepalive=60)
client.loop_start()

try:
    seq = 0
    while True:
        # 模拟温度: 25~50°C 随机波动
        temperature = round(random.uniform(25.0, 50.0), 1)
        payload = json.dumps({
            "device_id": CLIENT_ID,
            "seq": seq,
            "temperature": temperature,
            "timestamp": time.time()
        })
        result = client.publish(TOPIC, payload, qos=1)
        result.wait_for_publish()
        print(f"[设备] 发布 seq={seq}, temp={temperature}°C")
        seq += 1
        time.sleep(2)  # 每2秒发布一次
except KeyboardInterrupt:
    print("[设备] 手动停止")
finally:
    client.loop_stop()
    client.disconnect()

# subscriber.py - 监控订阅端，带温度报警
import json
import paho.mqtt.client as mqtt

BROKER = "localhost"
PORT = 1883
TOPIC = "sensor/temp"
CLIENT_ID = "monitor_001"
ALARM_THRESHOLD = 45.0  # 报警阈值

def on_connect(client, userdata, flags, rc):
    if rc == 0:
        print(f"[监控] 已连接Broker，订阅主题: {TOPIC}")
        client.subscribe(TOPIC, qos=1)
    else:
        print(f"[监控] 连接失败, code={rc}")

def on_message(client, userdata, msg):
    try:
        data = json.loads(msg.payload.decode())
        temp = data["temperature"]
        device = data["device_id"]
        seq = data["seq"]
        
        # 报警判定
        if temp >= ALARM_THRESHOLD:
            print(f"[报警] 设备{device} seq={seq} 温度 {temp}°C ≥ {ALARM_THRESHOLD}°C!")
        else:
            print(f"[正常] 设备{device} seq={seq} 温度 {temp}°C")
    except (json.JSONDecodeError, KeyError) as e:
        print(f"[监控] 消息解析失败: {e}, 原始: {msg.payload}")

client = mqtt.Client(client_id=CLIENT_ID)
client.on_connect = on_connect
client.on_message = on_message
client.connect(BROKER, PORT, keepalive=60)

try:
    client.loop_forever()
except KeyboardInterrupt:
    print("[监控] 手动停止")
    client.disconnect()

运行与验证

前置条件：安装Python 3.6+，并启动MQTT Broker（如mosquitto，默认监听1883端口）。

# 1. 安装依赖
pip install -r requirements.txt

# 2. 启动Broker（若未运行）
mosquitto -v

# 3. 终端A：启动监控订阅端
python subscriber.py

# 4. 终端B：启动设备模拟器
python publisher.py

验证方法：观察两个终端的输出。设备端每2秒发布一条温度数据；监控端实时收到并打印。当温度值≥45°C时，监控端输出[报警]前缀日志，否则输出[正常]前缀。

判定标准：

设备端显示已连接Broker且持续输出[设备] 发布...日志；
监控端显示已连接Broker，订阅主题，且每条设备发布消息都能在监控端看到对应[正常]或[报警]日志；
由于模拟温度范围25~50°C，运行足够长时间（通常10秒内）必然出现至少一条[报警]日志，证明报警功能生效；
Ctrl+C可正常终止两端程序，无异常堆栈。

预期输出示例（与代码逻辑严格对应，非实测数据）：

# 设备端
[设备] 已连接Broker: localhost:1883
[设备] 发布 seq=0, temp=32.5°C
[设备] 发布 seq=1, temp=47.2°C

# 监控端
[监控] 已连接Broker，订阅主题: sensor/temp
[正常] 设备device_sim_001 seq=0 温度 32.5°C
[报警] 设备device_sim_001 seq=1 温度 47.2°C ≥ 45.0°C!