📚 课程目录第3章边缘数据上报与存储

3. 边缘数据上报与存储

任务说明

本案例实现边缘设备将温湿度传感器数据通过MQTT协议上报至云平台，并由云平台侧服务接收后写入云数据库（以SQLite为例，便于本地验证）。整体链路为：边缘模拟传感器 → MQTT Broker → 云平台订阅服务 → 数据库存储。

环境依赖

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

核心代码

以下代码包含两个角色：边缘设备发布端与云平台订阅存储端。为便于单机演示，两段代码合并为一个文件，通过命令行参数区分运行模式。

import sqlite3
import json
import time
import random
import sys
import paho.mqtt.client as mqtt

# ========== 配置 ==========
BROKER = "broker.emqx.io"
PORT = 1883
TOPIC = "edge/sensor/temperature"
DB_FILE = "sensor_data.db"
CLIENT_ID_PUB = "edge_device_001"
CLIENT_ID_SUB = "cloud_platform_001"

# ========== 云平台订阅端：接收数据并存储 ==========
def init_db():
    conn = sqlite3.connect(DB_FILE)
    conn.execute('''CREATE TABLE IF NOT EXISTS sensor_records
                    (id INTEGER PRIMARY KEY AUTOINCREMENT,
                     device_id TEXT NOT NULL,
                     temperature REAL NOT NULL,
                     humidity REAL NOT NULL,
                     timestamp INTEGER NOT NULL)''')
    conn.commit()
    conn.close()

def on_message(client, userdata, msg):
    try:
        payload = json.loads(msg.payload.decode())
        conn = sqlite3.connect(DB_FILE)
        conn.execute(
            "INSERT INTO sensor_records (device_id, temperature, humidity, timestamp) VALUES (?, ?, ?, ?)",
            (payload["device_id"], payload["temperature"], payload["humidity"], payload["timestamp"])
        )
        conn.commit()
        conn.close()
        print(f"[存储成功] device={payload['device_id']}, temp={payload['temperature']}, hum={payload['humidity']}")
    except Exception as e:
        print(f"[存储失败] {e}")

def run_subscriber():
    init_db()
    client = mqtt.Client(CLIENT_ID_SUB)
    client.on_message = on_message
    client.connect(BROKER, PORT, 60)
    client.subscribe(TOPIC)
    print(f"云平台已订阅 {TOPIC}，等待数据...")
    client.loop_forever()

# ========== 边缘设备发布端：模拟传感器上报 ==========
def run_publisher():
    client = mqtt.Client(CLIENT_ID_PUB)
    client.connect(BROKER, PORT, 60)
    client.loop_start()
    print("边缘设备开始上报数据（Ctrl+C 停止）...")
    try:
        while True:
            payload = {
                "device_id": CLIENT_ID_PUB,
                "temperature": round(random.uniform(20.0, 35.0), 2),
                "humidity": round(random.uniform(40.0, 70.0), 2),
                "timestamp": int(time.time())
            }
            client.publish(TOPIC, json.dumps(payload))
            print(f"[已上报] {payload}")
            time.sleep(3)
    except KeyboardInterrupt:
        print("\n停止上报")
        client.loop_stop()
        client.disconnect()

if __name__ == "__main__":
    if len(sys.argv) < 2:
        print("用法: python edge_cloud_demo.py [publish|subscribe]")
        sys.exit(1)
    mode = sys.argv[1]
    if mode == "publish":
        run_publisher()
    elif mode == "subscribe":
        run_subscriber()
    else:
        print("无效模式，请使用 publish 或 subscribe")

运行与验证

启动云平台订阅端（终端1）：
python edge_cloud_demo.py subscribe
预期输出：云平台已订阅 edge/sensor/temperature，等待数据...
启动边缘设备发布端（终端2）：
python edge_cloud_demo.py publish
预期输出：每3秒打印一条 [已上报] {...} 数据
验证数据库存储：订阅端终端会同步打印 [存储成功] 日志。另开终端查询：
sqlite3 sensor_data.db "SELECT * FROM sensor_records ORDER BY id DESC LIMIT 5;"
应能看到与发布端上报一致的温湿度记录，字段包含 device_id、temperature、humidity、timestamp。

通过判定标准

订阅端终端出现 [存储成功] 且无异常堆栈；
sqlite3 查询返回至少1条与发布端最近一次上报数据一致（温度、湿度、时间戳匹配）的记录；
发布端停止（Ctrl+C）后，订阅端保持运行且不再新增存储日志。

常见问题

连接Broker失败：检查网络或更换为本地Broker地址（如 localhost），并确保端口1883开放。
数据库文件位置：SQLite文件生成于运行订阅端命令的当前目录，如需指定路径可修改 DB_FILE 变量。
数据乱码或解析失败：确认发布端与订阅端使用相同Topic和JSON字段名（device_id、temperature、humidity、timestamp）。