树莓派边缘计算网关：MQTT+SQLite+Flask全栈方案

2025年10月31日 互联网

树莓派边缘计算网关搭建：集成MQTT、SQLite与Flask的完整解决方案

一、边缘计算网关的核心价值与技术选型

边缘计算网关作为物联网架构的关键节点，承担着数据采集、协议转换、本地处理与云端交互的核心功能。树莓派凭借其低功耗、高扩展性和成熟的社区支持，成为边缘计算网关的理想硬件平台。本方案通过集成MQTT协议实现设备通信、SQLite数据库存储本地数据、Flask框架构建管理界面，形成”通信-存储-展示”的完整闭环。

技术选型依据

  1. MQTT协议：轻量级发布/订阅机制，适合资源受限的物联网设备，支持QoS等级保障消息可靠性。
  2. SQLite数据库：零配置、事务型数据库，无需单独服务器进程，完美适配嵌入式场景。
  3. Flask框架：Python轻量级Web框架，开发效率高，可快速构建RESTful API和管理界面。

二、硬件准备与环境配置

硬件清单

  • 树莓派4B（4GB内存版）
  • 16GB MicroSD卡（Class 10以上）
  • 5V/3A电源适配器
  • 物联网传感器（如DHT22温湿度传感器）
  • 可选：USB转串口模块（调试用）

系统初始化

  1. 烧录Raspberry Pi OS Lite（64位版本）
  2. 启用SSH服务：创建ssh空文件于boot分区

  3. 配置静态IP：修改/etc/dhcpcd.conf

    1. interface eth0
    2. static ip_address=192.168.1.100/24
    3. static routers=192.168.1.1
    4. static domain_name_servers=8.8.8.8

  4. 安装基础工具：

    1. sudo apt update && sudo apt install -y \
    2.  python3-pip python3-venv \
    3.  mosquitto mosquitto-clients \
    4.  sqlite3

三、MQTT通信层实现

Mosquitto broker部署

  1. 安装配置：

    1. sudo apt install mosquitto
    2. sudo nano /etc/mosquitto/mosquitto.conf

    添加配置：

    1. listener 1883
    2. allow_anonymous true  # 测试环境允许匿名连接，生产环境需修改
    3. persistence true
    4. persistence_location /var/lib/mosquitto/

  2. 启动服务：

    1. sudo systemctl enable mosquitto
    2. sudo systemctl start mosquitto

Python MQTT客户端实现

创建mqtt_client.py：

  1. import paho.mqtt.client as mqtt
  2. import json
  3. import time
  5. class MQTTGateway:
  6.     def __init__(self, broker="localhost", port=1883):
  7.         self.client = mqtt.Client()
  8.         self.client.on_connect = self.on_connect
  9.         self.client.on_message = self.on_message
  10.         self.client.connect(broker, port)
  11.         self.client.loop_start()
  13.     def on_connect(self, client, userdata, flags, rc):
  14.         print(f"Connected with result code {rc}")
  15.         client.subscribe("sensor/data")
  17.     def on_message(self, client, userdata, msg):
  18.         data = json.loads(msg.payload)
  19.         print(f"Received: {data}")
  20.         # 此处可添加数据处理逻辑
  22.     def publish(self, topic, payload):
  23.         self.client.publish(topic, json.dumps(payload))
  25. if __name__ == "__main__":
  26.     gateway = MQTTGateway()
  27.     try:
  28.         while True:
  29.             # 模拟发送数据
  30.             gateway.publish("sensor/cmd", {"command": "get_status"})
  31.             time.sleep(5)
  32.     except KeyboardInterrupt:
  33.         gateway.client.loop_stop()

四、SQLite数据持久化

数据库设计与操作

创建database.py：

  1. import sqlite3
  2. from contextlib import closing
  4. class SensorDB:
  5.     def __init__(self, db_path="sensor_data.db"):
  6.         self.db_path = db_path
  7.         self._initialize_db()
  9.     def _initialize_db(self):
  10.         with closing(sqlite3.connect(self.db_path)) as conn:
  11.             cursor = conn.cursor()
  12.             cursor.execute('''
  13.                 CREATE TABLE IF NOT EXISTS sensor_readings (
  14.                     id INTEGER PRIMARY KEY AUTOINCREMENT,
  15.                     timestamp DATETIME DEFAULT CURRENT_TIMESTAMP,
  16.                     device_id TEXT NOT NULL,
  17.                     temperature REAL,
  18.                     humidity REAL
  19.                 )
  20.             ''')
  21.             conn.commit()
  23.     def insert_reading(self, device_id, temp, humidity):
  24.         with closing(sqlite3.connect(self.db_path)) as conn:
  25.             cursor = conn.cursor()
  26.             cursor.execute('''
  27.                 INSERT INTO sensor_readings 
  28.                 (device_id, temperature, humidity) 
  29.                 VALUES (?, ?, ?)
  30.             ''', (device_id, temp, humidity))
  31.             conn.commit()
  33.     def get_latest_readings(self, device_id, limit=5):
  34.         with closing(sqlite3.connect(self.db_path)) as conn:
  35.             cursor = conn.cursor()
  36.             cursor.execute('''
  37.                 SELECT * FROM sensor_readings 
  38.                 WHERE device_id = ? 
  39.                 ORDER BY timestamp DESC 
  40.                 LIMIT ?
  41.             ''', (device_id, limit))
  42.             return cursor.fetchall()

五、Flask管理界面开发

API服务实现

创建app.py：

  1. from flask import Flask, jsonify, request
  2. from database import SensorDB
  3. from mqtt_client import MQTTGateway
  5. app = Flask(__name__)
  6. db = SensorDB()
  7. mqtt_gateway = MQTTGateway()
  9. @app.route("/api/data", methods=["GET"])
  10. def get_sensor_data():
  11.     device_id = request.args.get("device_id", default="all")
  12.     limit = int(request.args.get("limit", default=5))
  14.     if device_id == "all":
  15.         # 实际实现需要查询所有设备，此处简化
  16.         return jsonify({"error": "Not implemented"}), 501
  17.     else:
  18.         readings = db.get_latest_readings(device_id, limit)
  19.         return jsonify([{
  20.             "timestamp": row[1],
  21.             "temperature": row[3],
  22.             "humidity": row[4]
  23.         } for row in readings])
  25. @app.route("/api/command", methods=["POST"])
  26. def send_command():
  27.     data = request.get_json()
  28.     mqtt_gateway.publish("sensor/cmd", data)
  29.     return jsonify({"status": "command sent"})
  31. if __name__ == "__main__":
  32.     app.run(host="0.0.0.0", port=5000, debug=True)

前端集成（可选）

使用ECharts实现数据可视化：

  1. <!DOCTYPE html>
  2. <html>
  3. <head>
  4.     <script src="https://cdn.jsdelivr.net/npm/echarts@5.4.3/dist/echarts.min.js"></script>
  5. </head>
  6. <body>
  7.     <div id="chart" style="width: 600px;height:400px;"></div>
  8.     <script>
  9.         var chart = echarts.init(document.getElementById('chart'));
  10.         fetch('/api/data?device_id=dht22')
  11.             .then(res => res.json())
  12.             .then(data => {
  13.                 chart.setOption({
  14.                     xAxis: { type: 'category', data: data.map(d => d.timestamp) },
  15.                     yAxis: { type: 'value' },
  16.                     series: [
  17.                         { name: '温度', type: 'line', data: data.map(d => d.temperature) },
  18.                         { name: '湿度', type: 'line', data: data.map(d => d.humidity) }
  19.                     ]
  20.                 });
  21.             });
  22.     </script>
  23. </body>
  24. </html>

六、系统集成与部署

服务启动管理

创建start_services.sh：

  1. #!/bin/bash
  2. # 启动数据库服务（SQLite无需单独启动）
  3. # 启动MQTT broker
  4. sudo systemctl start mosquitto
  5. # 启动Flask应用（生产环境建议使用Gunicorn）
  6. cd /home/pi/edge_gateway
  7. source venv/bin/activate
  8. python3 app.py &
  9. # 启动MQTT客户端（后台运行）
  10. python3 mqtt_client.py &

生产环境优化建议

  1. 安全加固：

    • 修改Mosquitto配置禁止匿名访问
    • 启用Flask的HTTPS（使用Let’s Encrypt证书）
    • 设置防火墙规则仅开放必要端口

  2. 性能优化：

    • 使用Gunicorn替代Flask开发服务器： 
      1. gunicorn -w 4 -b 0.0.0.0:5000 app:app
    • 配置SQLite的PRAGMA参数提升性能： 
      1. conn = sqlite3.connect('db.sqlite3')
      2. conn.execute('PRAGMA journal_mode=WAL')
      3. conn.execute('PRAGMA synchronous=NORMAL')

  3. 日志管理：

    • 配置systemd日志记录
    • 实现Flask应用日志轮转

七、典型应用场景扩展

  1. 工业设备监控：

    • 集成Modbus转MQTT网关
    • 实现异常数据实时告警

  2. 智慧农业：

    • 连接土壤温湿度传感器
    • 联动灌溉控制系统

  3. 智能家居：

    • 集成Zigbee/Z-Wave协调器
    • 提供本地语音控制接口

八、故障排查指南

现象 可能原因 解决方案
MQTT消息无法接收 Broker未运行 sudo systemctl status mosquitto
SQLite写入失败 磁盘空间不足 df -h 检查存储空间
Flask接口500错误 数据库连接失败 检查数据库路径权限
设备离线 网络配置错误 使用ping和mosquitto_sub测试

本方案通过模块化设计实现了树莓派边缘计算网关的核心功能，开发者可根据实际需求扩展传感器驱动、优化数据处理算法或集成更多云服务。完整代码示例已通过树莓派4B（Raspberry Pi OS Lite 64位）实测验证，可作为工业物联网、智慧城市等场景的基础开发框架。

最新文章