全栈物联网数据监控：从MQTT设备到Web看板#

1. 设备清单#

• 物联网设备：ML307R，运行度云DTU固件
• 服务器：阿里云轻量应用服务器
• 操作系统：Ubuntu 24.02 LTS
• 核心服务：EMQX 5.0+，MySQL 8.0，Python 3.10+

2. 网络配置#

1
服务器公网IP：自行查询
2
EMQX端口：1883（MQTT）、18083（管理界面）
3
Web应用端口：5001

1.1 系统更新与基础工具安装#

1
# 更新系统包
2
sudo apt update && sudo apt upgrade -y
3

4
# 安装必要工具
5
sudo apt install -y vim curl wget git net-tools

1.2 防火墙配置（阿里云控制台）#

在阿里云服务器安全组中开放以下端口：

• 1883（MQTT协议）
• 18083（EMQX Dashboard）
• 5001（Web应用）
• 3306（MySQL，建议仅内网访问）

1
# 本地防火墙检查（可选）
2
sudo ufw status
3
sudo ufw allow 1883/tcp
4
sudo ufw allow 18083/tcp
5
sudo ufw allow 5001/tcp

2.1 EMQX安装#

1
# 下载最新版EMQX
2
curl -s https://assets.emqx.com/scripts/install-emqx.sh | sudo bash
3

4
# 安装EMQX
5
sudo apt install emqx -y
6

7
# 启动服务
8
sudo systemctl start emqx
9
sudo systemctl enable emqx
10

11
# 检查状态
12
sudo systemctl status emqx

2.2 EMQX基础配置#

1. 访问管理界面：http://[服务器IP]:18083
2. 默认账号：admin / public
3. 创建应用用户（用于数据采集服务）：
   • 用户名：emqx_data
   • 密码：123456

2.3 WebSocket调试#

• 使用EMQX内置的

3.1 ML307R设备配置#

通过设备Web管理界面配置：

网络通道参数

• 通道类型：MQTT
• 服务器地址：服务器公网IP
• 端口：1883
• Client ID：device_001（按实际设备编号）

主题设置

• 订阅主题：test/topic/down（接收服务器指令）
• 发布主题：
  • test/topic/up（发送传感器数据）
  • test/topic/command（发送指令请求）

QoS设置

• QoS等级：1
• Retain：0
• 心跳间隔：600000ms（10分钟）
• 心跳消息：hello

3.2 数据格式示例#

设备发送的数据格式应为JSON：

1
{
2
  "id": "device_001",
3
  "params": {
4
    "TDS": {"value": 125.5},
5
    "COD": {"value": 15.3},
6
    "pH": {"value": 7.2},
7
    "Tem": {"value": 25.6},
8
    "Hum": {"value": 65.8}
9
  }
10
}

注意：度云在多个发布通道之间进行切换时，发送指令前需要添加指令指明发布通道 指令格式MQPUB,0,0,数据

4.1. 主要功能文件#

• 文件位置：通常放置在/home/admin/mqtt_to_mysql.py（根据实际部署路径调整）。
• 作用：作为数据处理服务，从MQTT代理（如EMQX）订阅指定主题的消息，将接收到的数据（如传感器数据）实时解析并存储到MySQL数据库。
• 实现功能：
  • 使用paho-mqtt库连接MQTT代理，订阅主题（例如sensor/data）。
  • 解析JSON格式的MQTT消息，提取关键字段（如时间戳、设备ID、数值）。
  • 使用mysql-connector-python库连接MySQL数据库，将数据插入到预定义的表中（例如sensor_logs）。
  • 包含错误处理机制，确保网络中断或数据库异常时能重连和日志记录。

2. 服务的自启动设置及维护#

• 创建systemd服务文件：

  • 在/etc/systemd/system/目录下创建服务文件mqtt_to_mysql.service：

    1
    [Unit]
    2
    Description=MQTT to MySQL Data Processing Service
    3
    After=network.target mysql.service
    4
    Requires=mysql.service
    5
    
    6
    [Service]
    7
    User=admin
    8
    WorkingDirectory=/home/admin
    9
    ExecStart=/home/admin/mqtt_venv/bin/python3 /home/admin/mqtt_to_mysql.py
    10
    Restart=always
    11
    RestartSec=10
    12
    Environment="PYTHONUNBUFFERED=1"
    13
    StandardOutput=syslog
    14
    StandardError=syslog
    15
    SyslogIdentifier=mqtt_to_mysql
    16
    
    17
    [Install]
    18
    WantedBy=multi-user.target
    

  • 说明：

    • After=mysql.service确保MySQL先启动。
    • 使用虚拟环境mqtt_venv中的Python解释器执行脚本。
    • Restart=always使服务崩溃后自动重启。

• 自启动设置：

  • 重新加载systemd配置：sudo systemctl daemon-reload
  • 启用开机自启动：sudo systemctl enable mqtt_to_mysql.service
  • 启动服务：sudo systemctl start mqtt_to_mysql.service

• 维护操作：

  • 检查服务状态：sudo systemctl status mqtt_to_mysql.service
  • 停止服务：sudo systemctl stop mqtt_to_mysql.service
  • 重启服务：sudo systemctl restart mqtt_to_mysql.service
  • 查看日志：sudo journalctl -u mqtt_to_mysql.service -f

3. MySQL的设置，维护，查询#

• 设置：

  • 登录MySQL：sudo mysql -u root -p

  • 创建数据库（例如emqx_data）：CREATE DATABASE emqx_data;

  • 创建用户并授权（例如用户emqx_user，密码123456）：

    1
    CREATE USER 'emqx_user'@'localhost' IDENTIFIED BY '123456';
    2
    GRANT ALL PRIVILEGES ON emqx_data.* TO 'emqx_user'@'localhost';
    3
    FLUSH PRIVILEGES;
    

  • 创建数据表（示例）：

    1
    USE emqx_data;
    2
    CREATE TABLE sensor_logs (
    3
        id INT AUTO_INCREMENT PRIMARY KEY,
    4
        timestamp DATETIME DEFAULT CURRENT_TIMESTAMP,
    5
        device_id VARCHAR(50),
    6
        value FLOAT,
    7
        topic VARCHAR(255)
    8
    );
    

• 维护：

  • 定期备份数据库：使用mysqldump -u emqx_user -p emqx_data > backup.sql
  • 监控性能：检查MySQL错误日志（/var/log/mysql/error.log）或使用SHOW PROCESSLIST;
  • 优化表：定期运行OPTIMIZE TABLE sensor_logs;

• 查询：

  • 连接数据库：mysql -u emqx_user -p emqx_data
  • 示例查询：
    • 查看最新数据：SELECT * FROM sensor_logs ORDER BY timestamp DESC LIMIT 10;
    • 统计设备数据量：SELECT device_id, COUNT(*) FROM sensor_logs GROUP BY device_id;

5.1 MySQL安装#

1
# 安装MySQL
2
sudo apt install mysql-server -y
3

4
# 启动服务
5
sudo systemctl start mysql
6
sudo systemctl enable mysql
7

8
# 运行安全脚本
9
sudo mysql_secure_installation

5.2 创建数据库和用户#

1
-- 登录MySQL
2
sudo mysql
3

4
-- 创建数据库
5
CREATE DATABASE emqx_data CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci;
6

7
-- 创建用户
8
CREATE USER 'emqx_user'@'localhost' IDENTIFIED BY '123456';
9

10
-- 授予权限
11
GRANT ALL PRIVILEGES ON emqx_data.* TO 'emqx_user'@'localhost';
12

13
-- 刷新权限
14
FLUSH PRIVILEGES;
15

16
-- 退出
17
EXIT;

5.3 创建数据表#

1
-- 切换到emqx_data数据库
2
USE emqx_data;
3

4
-- 创建数据表
5
CREATE TABLE IF NOT EXISTS mqtt_msg (
6
    id INT AUTO_INCREMENT PRIMARY KEY,
7
    device_id VARCHAR(50),
8
    tds FLOAT,
9
    cod FLOAT,
10
    toc FLOAT,
11
    uv254 FLOAT,
12
    ph FLOAT,
13
    temp FLOAT,
14
    hum FLOAT,
15
    zhexian1 FLOAT,
16
    zhexian2 FLOAT,
17
    raw_data TEXT,
18
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
19
    INDEX idx_created_at (created_at),
20
    INDEX idx_device_id (device_id)
21
);
22

23
-- 创建指令表
24
CREATE TABLE IF NOT EXISTS mqtt_command (
25
    id INT AUTO_INCREMENT PRIMARY KEY,
26
    command_type VARCHAR(50) COMMENT '指令类型',
27
    device_id VARCHAR(50) COMMENT '目标设备ID',
28
    sender VARCHAR(50) COMMENT '指令发送者',
29
    raw_command TEXT COMMENT '原始指令内容',
30
    status VARCHAR(20) DEFAULT 'pending' COMMENT '指令状态',
31
    result TEXT COMMENT '执行结果',
32
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
33
    executed_at TIMESTAMP NULL COMMENT '指令执行时间',
34
    INDEX idx_status (status),
35
    INDEX idx_device_id (device_id),
36
    INDEX idx_created_at (created_at)
37
);
38

39
-- 创建下发记录表
40
CREATE TABLE IF NOT EXISTS mqtt_downlink (
41
    id INT AUTO_INCREMENT PRIMARY KEY,
42
    command_id INT,
43
    device_id VARCHAR(50),
44
    topic VARCHAR(255),
45
    payload TEXT,
46
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
47
    INDEX idx_command_id (command_id),
48
    INDEX idx_device_id (device_id)
49
);

5.4 数据库维护命令#

1
# 查看数据统计
2
mysql -u emqx_user -p123456 -D emqx_data -e "
3
SELECT '数据统计' AS '';
4
SELECT COUNT(*) AS 总记录数,
5
       COUNT(DISTINCT device_id) AS 设备数量,
6
       MIN(created_at) AS 最早记录,
7
       MAX(created_at) AS 最新记录
8
FROM mqtt_msg;
9
"
10

11
# 查看最近数据
12
mysql -u emqx_user -p123456 -D emqx_data --table -e "
13
SELECT id, device_id,
14
       ROUND(tds,1) AS TDS,
15
       ROUND(cod,1) AS COD,
16
       ROUND(ph,1) AS pH,
17
       ROUND(temp,1) AS 温度,
18
       ROUND(hum,1) AS 湿度,
19
       DATE_FORMAT(created_at, '%m-%d %H:%i') AS 时间
20
FROM mqtt_msg
21
ORDER BY id DESC
22
LIMIT 10;"
23

24
# 清空数据表（谨慎使用）
25
mysql -u emqx_user -p123456 -D emqx_data -e "
26
TRUNCATE TABLE mqtt_msg;
27
TRUNCATE TABLE mqtt_command;"

5.5 服务的介绍、运行与维护#

1
[Unit]
2
Description=MQTT to MySQL Super Subscriber
3
After=network.target emqx.service
4

5
[Service]
6
Type=simple
7
User=admin
8
WorkingDirectory=/home/admin/
9
ExecStart=/home/admin/mqtt_venv/bin/python3 /home/admin/mqtt_to_mysql.py
10
Restart=always
11
RestartSec=5
12

13
[Install]
14
WantedBy=multi-user.target

1
# 创建服务文件
2
sudo nano /etc/systemd/system/mqtt_to_mysql.service
3

4
# 将上述服务配置粘贴到文件中
5
# 保存并退出编辑器
6

7
# 重新加载systemd配置
8
sudo systemctl daemon-reload
9

10
# 启用开机自启动
11
sudo systemctl enable mqtt_to_mysql.service
12

13
# 启动服务
14
sudo systemctl start mqtt_to_mysql.service

1
# 查看服务状态（运行状态、日志信息）
2
sudo systemctl status mqtt_to_mysql.service
3

4
# 启动服务
5
sudo systemctl start mqtt_to_mysql.service
6

7
# 停止服务
8
sudo systemctl stop mqtt_to_mysql.service
9

10
# 重启服务（适用于更新代码后）
11
sudo systemctl restart mqtt_to_mysql.service
12

13
# 查看实时日志
14
sudo journalctl -u mqtt_to_mysql.service -f
15

16
# 查看最近100行日志
17
sudo journalctl -u mqtt_to_mysql.service -n 100
18

19
# 禁用开机自启动
20
sudo systemctl disable mqtt_to_mysql.service

1
# 检查服务是否正在运行
2
systemctl is-active mqtt_to_mysql.service
3

4
# 检查服务是否启用开机启动
5
systemctl is-enabled mqtt_to_mysql.service
6

7
# 查看服务的依赖关系
8
systemctl list-dependencies mqtt_to_mysql.service

6.1 py文件的位置和运行思路#

• 文件位置：/home/admin/mqtt_dashboard/app.py
• 运行思路：
  • 基于Flask框架构建Web服务，提供数据查询API和页面渲染
  • 连接MySQL数据库，查询水质监控数据（TDS、COD、pH、温度、湿度等）
  • 通过RESTful API提供JSON格式数据给前端界面
  • 集成简单的数据管理功能（如清空数据库记录）
  • 支持多数据源查询，包括实时采样数据和历史趋势数据

6.2 html文件的运行思路#

• 文件结构：位于/home/admin/mqtt_dashboard/templates/目录（或作为静态文件）
• 运行思路：
  • 使用Bootstrap 5框架构建响应式界面，适配不同设备
  • 集成ECharts图表库展示水质参数实时趋势曲线
  • 前端JavaScript通过定时轮询（默认5秒）从后端API获取最新数据
  • 实现模块化设计，包含三个主要面板：
    1. 趋势分析面板：实时曲线展示各水质参数变化
    2. 指令记录面板：显示最近的控制指令执行状态
    3. 采样数据面板：表格展示最新30条采样数据
  • 支持手动刷新、暂停自动更新、清空数据库等交互功能

6.3 服务的运行与维护#

• 服务配置：

  1
  # 创建服务文件
  2
  sudo nano /etc/systemd/system/mqtt_dashboard.service
  3
  
  4
  # 内容参考（基于提供的配置）：
  5
  [Unit]
  6
  Description=MQTT Data Dashboard Web Service
  7
  After=network.target mysql.service
  8
  
  9
  [Service]
  10
  User=admin
  11
  WorkingDirectory=/home/admin/mqtt_dashboard
  12
  ExecStart=/home/admin/mqtt_venv/bin/python3 app.py
  13
  Restart=always
  14
  RestartSec=10
  15
  Environment="PYTHONUNBUFFERED=1"
  16
  StandardOutput=syslog
  17
  StandardError=syslog
  18
  SyslogIdentifier=mqtt_dashboard
  19
  
  20
  [Install]
  21
  WantedBy=multi-user.target
  

• 服务管理命令：

  1
  # 重载配置
  2
  sudo systemctl daemon-reload
  3
  
  4
  # 启用开机启动
  5
  sudo systemctl enable mqtt_dashboard.service
  6
  
  7
  # 启动服务
  8
  sudo systemctl start mqtt_dashboard.service
  9
  
  10
  # 查看状态
  11
  sudo systemctl status mqtt_dashboard.service
  12
  
  13
  # 查看日志
  14
  sudo journalctl -u mqtt_dashboard.service -f
  15
  
  16
  # 停止服务
  17
  sudo systemctl stop mqtt_dashboard.service
  18
  
  19
  # 重启服务
  20
  sudo systemctl restart mqtt_dashboard.service
  

6.4 网页的使用方式#

1. 访问界面：

   • 在浏览器输入服务器IP地址及端口（默认5001），如：http://192.168.1.100:5001

2. 主要功能模块：

   • 趋势图表：实时展示各水质参数变化趋势，鼠标悬停可查看具体数值
   • 指令记录：显示最近发送的控制指令及执行状态
   • 采样数据：表格形式展示最近30条水质采样记录

3. 操作按钮功能：

   • 🔄 手动刷新：立即获取最新数据，无需等待自动更新
   • ⏸️/▶️ 暂停/继续自动更新：切换自动更新功能（默认每5秒更新）
   • 🗑️ 清空数据库：弹出确认对话框，可选择清空采样数据或指令记录

4. 性能监控：

   • 右下角显示系统状态，包括：
     • 最后更新时间
     • 请求响应时间
     • 自动更新状态

5. 快捷键支持：

   • F5键：手动刷新数据
   • 空格键：切换自动更新状态

6. 错误处理：

   • 网络异常时显示错误提示，3秒后自动重试
   • 数据加载时显示加载指示器，避免用户误操作

7. 页面优化特性：

   • 页面不可见时自动暂停更新，恢复可见后继续
   • 防抖机制防止频繁请求
   • 数据缓存避免重复加载

7.1 服务状态检查脚本#

创建 check_services.sh：

1
#!/bin/bash
2
echo "=== 服务状态检查 ==="
3
echo "检查时间: $(date)"
4

5
echo -e "\n1. EMQX服务:"
6
sudo systemctl is-active emqx
7

8
echo -e "\n2. MySQL服务:"
9
sudo systemctl is-active mysql
10

11
echo -e "\n3. 数据采集服务:"
12
sudo systemctl is-active mqtt-collector
13

14
echo -e "\n4. Web服务:"
15
sudo systemctl is-active mqtt-web
16

17
echo -e "\n5. 端口监听状态:"
18
netstat -tulpn | grep -E '(1883|18083|5001|3306)'
19

20
echo -e "\n6. 数据采集服务日志（最近5条）:"
21
sudo journalctl -u mqtt-collector -n 5 --no-pager
22

23
echo -e "\n7. 数据库连接测试:"
24
mysql -u emqx_user -p123456 -e "SELECT '数据库连接正常' AS Status;" 2>/dev/null || echo "数据库连接失败"
25

26
echo -e "\n=== 检查完成 ==="

7.2 数据备份脚本#

创建 backup_database.sh：

1
#!/bin/bash
2
BACKUP_DIR="/root/backups"
3
DATE=$(date +%Y%m%d_%H%M%S)
4
BACKUP_FILE="$BACKUP_DIR/emqx_data_$DATE.sql"
5

6
mkdir -p $BACKUP_DIR
7

8
echo "开始备份数据库..."
9
mysqldump -u emqx_user -p123456 emqx_data > $BACKUP_FILE
10

11
if [ $? -eq 0 ]; then
12
    echo "备份成功: $BACKUP_FILE"
13

14
    # 压缩备份文件
15
    gzip $BACKUP_FILE
16

17
    # 删除7天前的备份
18
    find $BACKUP_DIR -name "*.sql.gz" -mtime +7 -delete
19

20
    echo "备份完成，文件已压缩。"
21
else
22
    echo "备份失败!"
23
    exit 1
24
fi

7.3 自动清理旧数据#

创建 clean_old_data.sh：

1
#!/bin/bash
2
# 自动清理30天前的数据
3

4
echo "开始清理旧数据..."
5
echo "清理时间: $(date)"
6

7
# 清理30天前的数据记录
8
mysql -u emqx_user -p123456 -D emqx_data -e "
9
DELETE FROM mqtt_msg WHERE created_at < DATE_SUB(NOW(), INTERVAL 30 DAY);
10
DELETE FROM mqtt_command WHERE created_at < DATE_SUB(NOW(), INTERVAL 30 DAY);
11
DELETE FROM mqtt_downlink WHERE created_at < DATE_SUB(NOW(), INTERVAL 30 DAY);
12
"
13

14
echo "数据清理完成。"

8.1 常见问题及解决方案#

问题1：MQTT连接失败

1
检查步骤：
2
1. 确认EMQX服务运行状态：sudo systemctl status emqx
3
2. 检查端口监听：netstat -tulpn | grep 1883
4
3. 检查防火墙设置：sudo ufw status
5
4. 检查阿里云安全组规则

问题2：数据库连接失败

1
检查步骤：
2
1. 确认MySQL服务状态：sudo systemctl status mysql
3
2. 检查用户权限：mysql -u emqx_user -p123456
4
3. 检查数据库是否存在：SHOW DATABASES;

问题3：Web界面无法访问

1
检查步骤：
2
1. 确认Flask服务状态：sudo systemctl status mqtt-web
3
2. 检查端口监听：netstat -tulpn | grep 5001
4
3. 查看服务日志：sudo journalctl -u mqtt-web -f

问题4：数据未保存到数据库

1
检查步骤：
2
1. 检查数据采集服务日志：sudo journalctl -u mqtt-collector -f
3
2. 检查MQTT主题订阅状态
4
3. 检查数据库表结构是否匹配

8.2 日志查看命令#

1
# 查看实时日志
2
sudo journalctl -f
3

4
# 查看指定服务日志
5
sudo journalctl -u mqtt-collector -f
6
sudo journalctl -u mqtt-web -f
7
sudo journalctl -u emqx -f
8

9
# 查看特定时间段的日志
10
sudo journalctl -u mqtt-collector --since "2024-01-01 00:00:00" --until "2024-01-01 23:59:59"
11

12
# 查看错误日志
13
sudo journalctl -u mqtt-collector -p err