通过Ubuntu服务器部署Mqtt服务的记录#

1
##查看指令
2
mysql -u emqx_user -p123456 -D emqx_data -e "
3
SELECT '--- 当前总记录数 ---' AS '';
4
SELECT count(*) AS total_rows FROM mqtt_msg;
5
SELECT '--- 最近20条数据 (精简模式) ---' AS '';
6
SELECT id, device_id, tds, cod, ph, temp, hum, created_at FROM mqtt_msg ORDER BY id DESC LIMIT 20;"
7

8
##监控器指令
9
# 每 1 秒刷新一次
10
watch -n 1 'mysql -u emqx_user -p123456 -D emqx_data -e "SELECT count(*) AS Total FROM mqtt_msg; SELECT id, device_id, tds, cod, ph, temp, hum, created_at FROM mqtt_msg ORDER BY id DESC LIMIT 40;"'
11

12
##重启指令
13
# 1. 重新加载系统服务配置
14
sudo systemctl daemon-reload
15
# 2. 重启你的脚本服务
16
sudo systemctl restart mqtt-saver.service
17
# 3. 再次查看日志
18
sudo journalctl -u mqtt-saver -f
19

20
指令库查询
21
watch -t -n 1 'mysql -u emqx_user -p123456 -D emqx_data --table -e "
22
SELECT
23
    COUNT(*) AS 指令总数,
24
    SUM(status = \"pending\") AS 待处理,
25
    SUM(status = \"executed\") AS 已执行,
26
    SUM(status = \"failed\") AS 失败数
27
FROM mqtt_command;
28

29
SELECT
30
    id,
31
    command_type AS 类型,
32
    device_id AS 设备ID,
33
    -- 提取 JSON 中的 command 字段值
34
    raw_command->>\"$.command\" AS 指令内容,
35
    CASE
36
        WHEN status = \"pending\"  THEN \"待处理\"
37
        WHEN status = \"executed\" THEN \"已执行\"
38
        WHEN status = \"failed\"   THEN \"失败\"
39
        ELSE status
40
    END AS 状态,
41
    DATE_FORMAT(created_at, \"%H:%i:%s\") AS 接收时间,
42
    IFNULL(DATE_FORMAT(executed_at, \"%H:%i:%s\"), \"-\") AS 执行时间
43
FROM mqtt_command
44
ORDER BY id DESC
45
LIMIT 10;"'
46

47
数据库查询
48
watch -t -n 1 'mysql -u emqx_user -p123456 -D emqx_data --table -e "
49
SELECT
50
    COUNT(*) AS 采集总数,
51
    COUNT(DISTINCT device_id) AS 设备数量
52
FROM mqtt_msg;
53

54
SELECT
55
    id,
56
    device_id AS 设备ID,
57
    ROUND(tds, 1) AS TDS,
58
    ROUND(cod, 1) AS COD,
59
    ROUND(ph, 1) AS pH值,
60
    ROUND(temp, 1) AS 温度,
61
    ROUND(hum, 1) AS 湿度,
62
    DATE_FORMAT(created_at, \"%m-%d %H:%i:%s\") AS 采集时间
63
FROM mqtt_msg
64
ORDER BY id DESC
65
LIMIT 30;"'

1
CREATE TABLE IF NOT EXISTS mqtt_msg (
2
    id INT AUTO_INCREMENT PRIMARY KEY,
3
    device_id VARCHAR(50),      -- 对应 JSON 中的 id
4
    tds FLOAT,
5
    cod FLOAT,
6
    toc FLOAT,
7
    uv254 FLOAT,
8
    ph FLOAT,
9
    temp FLOAT,
10
    hum FLOAT,
11
    zhexian1 FLOAT,
12
    zhexian2 FLOAT,
13
    raw_data TEXT,              -- 保留一份原始数据备份
14
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
15
);

1
CREATE TABLE IF NOT EXISTS mqtt_command (
2
id INT AUTO_INCREMENT PRIMARY KEY,
3
command_type VARCHAR(50) COMMENT '指令类型，如：restart, config, update, stop等',
4
device_id VARCHAR(50) COMMENT '目标设备ID（可为空）',
5
sender VARCHAR(50) COMMENT '指令发送者',
6
raw_command TEXT COMMENT '原始指令内容',
7
status VARCHAR(20) DEFAULT 'pending' COMMENT '指令状态：pending, executed, failed',
8
result TEXT COMMENT '执行结果（可选）',
9
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
10
executed_at TIMESTAMP NULL COMMENT '指令执行时间',
11
INDEX idx_device_id (device_id),
12
INDEX idx_status (status),
13
INDEX idx_created_at (created_at)
14
);