尧图网站建设 
简介这份资源面向工业自动化与物联网方向的开发者、运维人员及技术学习者聚焦将Modbus协议设备接入MQTT消息队列的网关实现解决串口通信、数据转换、设备配置管理与远程监控等场景下的协议互通问题。压缩包共32个文件约273KB以JavaScript源码为主辅以Markdown说明文档、YAML配置、JSON数据、Dockerfile与Shell脚本覆盖核心逻辑、测试用例、容器化部署及使用说明等模块结构清晰便于按需查阅。目前已有69人学习下载。通过其中的modbus2mqtt实现代码、配置示例与测试脚本读者可理解Modbus设备接入、MQTT发布订阅、实时数据采集与跨平台部署的完整链路并借助文档与容器配置快速搭建实验环境适合作为物联网网关开发与协议转换的实践参考。1. 工业网关拆包实录Modbus 转 MQTT 到底解决了谁的痛点车间里一台 2012 年的温控表还在用 RS485 跑 Modbus RTU而中控室新上的看板只认 MQTT。这种「老设备说方言、新平台讲普通话」的场面几乎每个做设备接入的工程师都遇到过。这个开源项目就是干翻译的把 Modbus 协议RTU/TCP读上来的线圈、寄存器数据转成 MQTT 消息发布出去同时支持反向下发指令。它适合三类人——手里有一堆 PLC、传感器、数控机床要联网的集成商想用 STM32 或 Linux 盒子自建物联网网关的嵌入式开发者以及需要远程监控设备状态但不想动原有控制逻辑的运维。跨平台是它的卖点Windows 上跑调试、Linux 上跑生产都能用。下面我按「协议怎么转、代码怎么跑、坑在哪」的顺序拆一遍。2. Modbus 与 MQTT 的协议桥接从寄存器到 Topic 的映射逻辑2.1 为什么不能直接透传两种协议的语义鸿沟Modbus 是主从轮询模型数据存在线圈Coil、离散输入Discrete Input、保持寄存器Holding Register、输入寄存器Input Register四个区里每个区有地址和功能码。MQTT 是发布订阅模型数据以 Topic 为路径、Payload 为内容。直接透传的问题在于Modbus 的「地址功能码」在 MQTT 里没有对应概念订阅端拿到一串十六进制报文根本不知道是什么。常见做法是建立一张映射表把每个 Modbus 数据点映射成一个 MQTT TopicPayload 用 JSON 或纯数值。比如保持寄存器 40001 映射成device/plc01/holding/0值25.6。这样订阅端只关心 Topic 语义不用解析 Modbus 报文。项目里通常有一个配置文件或数据库表来维护这张映射改点位不用改代码。提示映射粒度别太细。一个寄存器一个 Topic 会导致 Topic 数量爆炸建议按设备或功能块聚合比如device/plc01/status一个 JSON 带多个字段。2.2 数据转换的核心参数字节序、数据类型、缩放因子Modbus 寄存器是 16 位但实际数据可能是 32 位浮点、32 位整型、甚至 64 位。这就涉及字节序问题。常见的有 ABCD大端、DCBA小端、BADC、CDAB 四种排列。我见过太多人读出来是乱码最后发现是字节序搞反了。除了字节序还有数据类型和缩放因子。比如温度寄存器原始值 256实际是 25.6℃缩放因子 0.1。这些参数必须在配置里写清楚。下面是一个典型的点位配置结构{ points: [ { name: temperature, slave_id: 1, function_code: 3, address: 0, quantity: 2, data_type: float32, byte_order: ABCD, scale: 0.1, unit: ℃, mqtt_topic: factory/line1/temp }, { name: motor_status, slave_id: 1, function_code: 1, address: 10, quantity: 1, data_type: bool, mqtt_topic: factory/line1/motor } ] }这段配置里function_code3 表示读保持寄存器1 表示读线圈。quantity是寄存器数量float32 占两个寄存器所以填 2。byte_order决定怎么拼字节。scale是缩放因子。mqtt_topic是发布目标。改点位只需要改这个文件不用动代码。2.3 轮询策略与 MQTT 发布节奏的配合Modbus 是轮询的MQTT 是事件驱动的。如果轮询周期 100ms但每个点都发 MQTTBroker 会被打爆。常见做法是分组轮询加变化上报把点位按实时性要求分组高频组 100ms 轮询但只在值变化超过阈值时才发布低频组 5s 轮询一次全量发布。另一个参数是 MQTT 的 QoS 等级。QoS 0 最多一次适合高频传感器数据QoS 1 至少一次适合控制指令QoS 2 恰好一次开销大一般不用。还有 retain 标志设为 true 时 Broker 会保留最后一条消息新订阅者立刻能拿到当前值适合状态类数据。# 轮询与发布的核心循环伪代码 import time from modbus_client import ModbusClient from mqtt_client import MqttClient modbus ModbusClient(port/dev/ttyUSB0, baudrate9600, parityN) mqtt MqttClient(broker127.0.0.1, port1883) last_values {} while True: for point in config[points]: raw modbus.read(point[slave_id], point[function_code], point[address], point[quantity]) value convert(raw, point[data_type], point[byte_order], point[scale]) # 变化上报只有值变了才发 if last_values.get(point[name]) ! value: mqtt.publish(point[mqtt_topic], value, qos1, retainTrue) last_values[point[name]] value time.sleep(0.1)这段循环里convert函数负责字节序转换和缩放。last_values字典做变化检测。retainTrue让新订阅者能立刻拿到当前值。实际项目中轮询周期和变化阈值都要根据现场调试没有万能值。3. 网关部署实操从串口配置到 MQTT Broker 联调3.1 串口参数配置与 Modbus RTU 链路打通串口是 Modbus RTU 的物理层参数不匹配什么都读不到。核心参数就五个波特率、数据位、停止位、校验位、流控。工业现场最常见的是 9600-8-N-1 或 19200-8-E-1。校验位尤其容易错有些设备标称无校验但实际用偶校验读出来全是超时。Linux 下串口设备名通常是/dev/ttyUSB0或/dev/ttyS0。权限问题也常见普通用户没权限打开串口需要把用户加到 dialout 组。Windows 下是COM1、COM3这种。跨平台项目一般用 pyserial 或 libmodbus 来屏蔽差异。# Linux 下查看串口设备 ls -l /dev/ttyUSB* /dev/ttyS* # 把当前用户加入 dialout 组需要重新登录生效 sudo usermod -aG dialout $USER # 用 modbus poll 或 mbpoll 命令行工具测试链路 mbpoll -m rtu -b 9600 -P none -d 8 -s 1 -a 1 -t 4 -r 1 -c 2 /dev/ttyUSB0mbpoll的参数含义-m rtu指定 Modbus RTU 模式-b 9600波特率-P none无校验-d 8数据位 8-s 1停止位 1-a 1从站地址 1-t 4读输入寄存器-r 1起始地址 1-c 2读两个寄存器。如果这条命令能返回数据说明物理链路和协议参数都对了。3.2 MQTT Broker 选型与本地搭建MQTT Broker 是消息中枢选型看规模和功能。小规模用 Mosquitto 就够了单文件、配置简单、资源占用低。大规模或需要集群用 EMQX 或 HiveMQ。本地调试我一般用 MosquittoDocker 一行命令跑起来。# Docker 启动 Mosquitto docker run -d --name mosquitto \ -p 1883:1883 \ -p 9001:9001 \ -v /opt/mosquitto/config:/mosquitto/config \ -v /opt/mosquitto/data:/mosquitto/data \ eclipse-mosquitto:2 # 配置文件 mosquitto.conf 关键项 # listener 1883 # allow_anonymous true # persistence true # persistence_location /mosquitto/data/listener 1883是 MQTT 默认端口。allow_anonymous true允许匿名连接生产环境要关掉并配用户名密码。persistence true让消息持久化重启不丢。WebSocket 端口 9001 是给浏览器端用的不需要可以不开。3.3 网关配置文件逐项拆解与启动验证网关的配置文件一般分三块Modbus 连接参数、点位映射表、MQTT 连接参数。下面是一个完整示例modbus: mode: rtu port: /dev/ttyUSB0 baudrate: 9600 parity: N stopbits: 1 bytesize: 8 timeout: 1 slaves: - id: 1 points: - name: temp fc: 3 addr: 0 qty: 2 type: float32 order: ABCD scale: 0.1 topic: factory/line1/temp mqtt: broker: 127.0.0.1 port: 1883 client_id: gateway_01 username: password: keepalive: 60 qos: 1 retain: truetimeout: 1是串口读超时秒数现场干扰大可以调到 2 或 3。client_id必须唯一两个网关用同一个 ID 会互相踢下线。keepalive: 60是心跳间隔Broker 超过 1.5 倍时间没收到心跳会断开。启动后验证分三步先看串口有没有数据上来再看 MQTT 有没有消息发布最后用订阅端确认 Payload 正确。# 订阅所有 Topic 查看网关发布的数据 mosquitto_sub -h 127.0.0.1 -p 1883 -t factory/# -v # 手动发布一条指令测试反向控制 mosquitto_pub -h 127.0.0.1 -p 1883 -t factory/line1/cmd -m {motor:1}-t factory/#里的#是通配符匹配所有子 Topic。-v显示 Topic 名。如果订阅端能看到 JSON 数据说明正向链路通了。反向控制需要网关订阅指令 Topic 并写 Modbus 寄存器这部分逻辑一般在代码里单独处理。4. 避坑与排查串口读不到、MQTT 断连、数据错位的血泪经验4.1 串口打开失败或读超时现象程序报Permission denied或SerialException或者能打开但读数据一直超时。原因Linux 下权限不足是最常见的其次是串口被其他程序占用比如调试助手没关还有波特率或校验位不匹配。解决先ls -l /dev/ttyUSB0看权限没有rw就给用户加 dialout 组。用lsof /dev/ttyUSB0查占用进程。参数不匹配就用 mbpoll 逐项试先确认能读到数据再跑网关。4.2 MQTT 频繁断连重连现象日志里反复出现Connection lost和Reconnecting。原因client_id 冲突、keepalive 设置过短、网络抖动、Broker 达到最大连接数。解决确保每个网关 client_id 唯一。keepalive 别低于 30 秒。网络不稳就开 MQTT 的自动重连并设clean_sessionfalse这样重连后能收到离线消息。Broker 端检查max_connections配置。4.3 读出来的数值明显不对现象温度显示 6553.5 或者负数明显超出量程。原因字节序搞反了或者数据类型选错了。float32 的 ABCD 和 CDAB 读出来完全是两个数。解决先用 Modbus Poll 这类工具读原始寄存器值手动算一遍。比如两个寄存器值是 0x41C8 和 0x0000ABCD 排列是 0x41C80000转 float 是 25.0。CDAB 排列是 0x000041C8转出来就很小。确认字节序后再改配置。4.4 变化上报导致数据「跳变」丢失现象订阅端看到的数据偶尔跳一下或者中间某个值没收到。原因变化阈值设得太小传感器噪声导致频繁触发或者 QoS 0 在弱网下丢包。解决加死区过滤比如温度变化小于 0.5℃ 不上报。关键数据用 QoS 1。如果还是丢检查 Broker 的max_queued_messages是不是太小。4.5 多从站轮询时响应慢现象从站多了以后每个点的刷新周期变长实时性下降。原因串口是半双工主站逐个轮询从站多了总时间线性增长。解决提高波特率19200 或 38400合并连续寄存器一次读取把不重要的点降频。实在不行分多个串口并行。5. 进阶技巧用规则引擎做数据清洗与告警联动网关把数据发到 MQTT 只是第一步真正产生价值的是后面的处理。我一般会在 Broker 后面挂一个规则引擎做三件事数据清洗、阈值告警、指令联动。数据清洗是把原始值转成业务语义。比如寄存器值 0/1 转成「运行/停止」数值超量程标记为无效。规则引擎里用 SQL 风格的语法就能写SELECT payload.temp AS temperature, CASE WHEN payload.temp 80 THEN overheat WHEN payload.temp 10 THEN low ELSE normal END AS temp_status FROM factory/line1/temp这段规则订阅温度 Topic输出带状态标签的新消息。EMQX 的规则引擎、Node-RED 都能做类似的事。清洗后的数据再入库或推给看板前端就不用写判断逻辑了。阈值告警是另一个高频需求。规则里判断超过阈值就发到告警 Topic再由通知服务发邮件或短信。注意加防抖不然温度在阈值附近波动会疯狂告警。常见做法是连续 N 次超限才触发或者加一个回差带。指令联动是反向控制。比如检测到温度超限自动发指令停电机。这里要特别小心自动控制指令必须加确认机制发出去后检查设备状态是否真的变了没变就重试或告警。我见过自动停机和手动开机打架设备来回启停的翻车现场。验证规则是否生效用 mosquitto_sub 订阅输出 Topic 就行。调试阶段建议把规则输出单独开一个 Topic别直接混在生产数据里。# 订阅清洗后的数据 mosquitto_sub -h 127.0.0.1 -t factory/line1/cleaned/# -v # 订阅告警 mosquitto_sub -h 127.0.0.1 -t factory/alarm/# -v从那以后我每次上规则引擎都强制先在测试环境跑 24 小时确认没有误报和指令冲突再切生产。这套网关加规则引擎的组合能把老设备的接入周期从两周压到两天。希望帮到你。本文还有配套的精品资源点击获取 工业网关实战：Modbus转MQTT协议桥接与部署避坑指南
简介这份资源面向工业自动化与物联网方向的开发者、运维人员及技术学习者聚焦将Modbus协议设备接入MQTT消息队列的网关实现解决串口通信、数据转换、设备配置管理与远程监控等场景下的协议互通问题。压缩包共32个文件约273KB以JavaScript源码为主辅以Markdown说明文档、YAML配置、JSON数据、Dockerfile与Shell脚本覆盖核心逻辑、测试用例、容器化部署及使用说明等模块结构清晰便于按需查阅。目前已有69人学习下载。通过其中的modbus2mqtt实现代码、配置示例与测试脚本读者可理解Modbus设备接入、MQTT发布订阅、实时数据采集与跨平台部署的完整链路并借助文档与容器配置快速搭建实验环境适合作为物联网网关开发与协议转换的实践参考。1. 工业网关拆包实录Modbus 转 MQTT 到底解决了谁的痛点车间里一台 2012 年的温控表还在用 RS485 跑 Modbus RTU而中控室新上的看板只认 MQTT。这种「老设备说方言、新平台讲普通话」的场面几乎每个做设备接入的工程师都遇到过。这个开源项目就是干翻译的把 Modbus 协议RTU/TCP读上来的线圈、寄存器数据转成 MQTT 消息发布出去同时支持反向下发指令。它适合三类人——手里有一堆 PLC、传感器、数控机床要联网的集成商想用 STM32 或 Linux 盒子自建物联网网关的嵌入式开发者以及需要远程监控设备状态但不想动原有控制逻辑的运维。跨平台是它的卖点Windows 上跑调试、Linux 上跑生产都能用。下面我按「协议怎么转、代码怎么跑、坑在哪」的顺序拆一遍。2. Modbus 与 MQTT 的协议桥接从寄存器到 Topic 的映射逻辑2.1 为什么不能直接透传两种协议的语义鸿沟Modbus 是主从轮询模型数据存在线圈Coil、离散输入Discrete Input、保持寄存器Holding Register、输入寄存器Input Register四个区里每个区有地址和功能码。MQTT 是发布订阅模型数据以 Topic 为路径、Payload 为内容。直接透传的问题在于Modbus 的「地址功能码」在 MQTT 里没有对应概念订阅端拿到一串十六进制报文根本不知道是什么。常见做法是建立一张映射表把每个 Modbus 数据点映射成一个 MQTT TopicPayload 用 JSON 或纯数值。比如保持寄存器 40001 映射成device/plc01/holding/0值25.6。这样订阅端只关心 Topic 语义不用解析 Modbus 报文。项目里通常有一个配置文件或数据库表来维护这张映射改点位不用改代码。提示映射粒度别太细。一个寄存器一个 Topic 会导致 Topic 数量爆炸建议按设备或功能块聚合比如device/plc01/status一个 JSON 带多个字段。2.2 数据转换的核心参数字节序、数据类型、缩放因子Modbus 寄存器是 16 位但实际数据可能是 32 位浮点、32 位整型、甚至 64 位。这就涉及字节序问题。常见的有 ABCD大端、DCBA小端、BADC、CDAB 四种排列。我见过太多人读出来是乱码最后发现是字节序搞反了。除了字节序还有数据类型和缩放因子。比如温度寄存器原始值 256实际是 25.6℃缩放因子 0.1。这些参数必须在配置里写清楚。下面是一个典型的点位配置结构{ points: [ { name: temperature, slave_id: 1, function_code: 3, address: 0, quantity: 2, data_type: float32, byte_order: ABCD, scale: 0.1, unit: ℃, mqtt_topic: factory/line1/temp }, { name: motor_status, slave_id: 1, function_code: 1, address: 10, quantity: 1, data_type: bool, mqtt_topic: factory/line1/motor } ] }这段配置里function_code3 表示读保持寄存器1 表示读线圈。quantity是寄存器数量float32 占两个寄存器所以填 2。byte_order决定怎么拼字节。scale是缩放因子。mqtt_topic是发布目标。改点位只需要改这个文件不用动代码。2.3 轮询策略与 MQTT 发布节奏的配合Modbus 是轮询的MQTT 是事件驱动的。如果轮询周期 100ms但每个点都发 MQTTBroker 会被打爆。常见做法是分组轮询加变化上报把点位按实时性要求分组高频组 100ms 轮询但只在值变化超过阈值时才发布低频组 5s 轮询一次全量发布。另一个参数是 MQTT 的 QoS 等级。QoS 0 最多一次适合高频传感器数据QoS 1 至少一次适合控制指令QoS 2 恰好一次开销大一般不用。还有 retain 标志设为 true 时 Broker 会保留最后一条消息新订阅者立刻能拿到当前值适合状态类数据。# 轮询与发布的核心循环伪代码 import time from modbus_client import ModbusClient from mqtt_client import MqttClient modbus ModbusClient(port/dev/ttyUSB0, baudrate9600, parityN) mqtt MqttClient(broker127.0.0.1, port1883) last_values {} while True: for point in config[points]: raw modbus.read(point[slave_id], point[function_code], point[address], point[quantity]) value convert(raw, point[data_type], point[byte_order], point[scale]) # 变化上报只有值变了才发 if last_values.get(point[name]) ! value: mqtt.publish(point[mqtt_topic], value, qos1, retainTrue) last_values[point[name]] value time.sleep(0.1)这段循环里convert函数负责字节序转换和缩放。last_values字典做变化检测。retainTrue让新订阅者能立刻拿到当前值。实际项目中轮询周期和变化阈值都要根据现场调试没有万能值。3. 网关部署实操从串口配置到 MQTT Broker 联调3.1 串口参数配置与 Modbus RTU 链路打通串口是 Modbus RTU 的物理层参数不匹配什么都读不到。核心参数就五个波特率、数据位、停止位、校验位、流控。工业现场最常见的是 9600-8-N-1 或 19200-8-E-1。校验位尤其容易错有些设备标称无校验但实际用偶校验读出来全是超时。Linux 下串口设备名通常是/dev/ttyUSB0或/dev/ttyS0。权限问题也常见普通用户没权限打开串口需要把用户加到 dialout 组。Windows 下是COM1、COM3这种。跨平台项目一般用 pyserial 或 libmodbus 来屏蔽差异。# Linux 下查看串口设备 ls -l /dev/ttyUSB* /dev/ttyS* # 把当前用户加入 dialout 组需要重新登录生效 sudo usermod -aG dialout $USER # 用 modbus poll 或 mbpoll 命令行工具测试链路 mbpoll -m rtu -b 9600 -P none -d 8 -s 1 -a 1 -t 4 -r 1 -c 2 /dev/ttyUSB0mbpoll的参数含义-m rtu指定 Modbus RTU 模式-b 9600波特率-P none无校验-d 8数据位 8-s 1停止位 1-a 1从站地址 1-t 4读输入寄存器-r 1起始地址 1-c 2读两个寄存器。如果这条命令能返回数据说明物理链路和协议参数都对了。3.2 MQTT Broker 选型与本地搭建MQTT Broker 是消息中枢选型看规模和功能。小规模用 Mosquitto 就够了单文件、配置简单、资源占用低。大规模或需要集群用 EMQX 或 HiveMQ。本地调试我一般用 MosquittoDocker 一行命令跑起来。# Docker 启动 Mosquitto docker run -d --name mosquitto \ -p 1883:1883 \ -p 9001:9001 \ -v /opt/mosquitto/config:/mosquitto/config \ -v /opt/mosquitto/data:/mosquitto/data \ eclipse-mosquitto:2 # 配置文件 mosquitto.conf 关键项 # listener 1883 # allow_anonymous true # persistence true # persistence_location /mosquitto/data/listener 1883是 MQTT 默认端口。allow_anonymous true允许匿名连接生产环境要关掉并配用户名密码。persistence true让消息持久化重启不丢。WebSocket 端口 9001 是给浏览器端用的不需要可以不开。3.3 网关配置文件逐项拆解与启动验证网关的配置文件一般分三块Modbus 连接参数、点位映射表、MQTT 连接参数。下面是一个完整示例modbus: mode: rtu port: /dev/ttyUSB0 baudrate: 9600 parity: N stopbits: 1 bytesize: 8 timeout: 1 slaves: - id: 1 points: - name: temp fc: 3 addr: 0 qty: 2 type: float32 order: ABCD scale: 0.1 topic: factory/line1/temp mqtt: broker: 127.0.0.1 port: 1883 client_id: gateway_01 username: password: keepalive: 60 qos: 1 retain: truetimeout: 1是串口读超时秒数现场干扰大可以调到 2 或 3。client_id必须唯一两个网关用同一个 ID 会互相踢下线。keepalive: 60是心跳间隔Broker 超过 1.5 倍时间没收到心跳会断开。启动后验证分三步先看串口有没有数据上来再看 MQTT 有没有消息发布最后用订阅端确认 Payload 正确。# 订阅所有 Topic 查看网关发布的数据 mosquitto_sub -h 127.0.0.1 -p 1883 -t factory/# -v # 手动发布一条指令测试反向控制 mosquitto_pub -h 127.0.0.1 -p 1883 -t factory/line1/cmd -m {motor:1}-t factory/#里的#是通配符匹配所有子 Topic。-v显示 Topic 名。如果订阅端能看到 JSON 数据说明正向链路通了。反向控制需要网关订阅指令 Topic 并写 Modbus 寄存器这部分逻辑一般在代码里单独处理。4. 避坑与排查串口读不到、MQTT 断连、数据错位的血泪经验4.1 串口打开失败或读超时现象程序报Permission denied或SerialException或者能打开但读数据一直超时。原因Linux 下权限不足是最常见的其次是串口被其他程序占用比如调试助手没关还有波特率或校验位不匹配。解决先ls -l /dev/ttyUSB0看权限没有rw就给用户加 dialout 组。用lsof /dev/ttyUSB0查占用进程。参数不匹配就用 mbpoll 逐项试先确认能读到数据再跑网关。4.2 MQTT 频繁断连重连现象日志里反复出现Connection lost和Reconnecting。原因client_id 冲突、keepalive 设置过短、网络抖动、Broker 达到最大连接数。解决确保每个网关 client_id 唯一。keepalive 别低于 30 秒。网络不稳就开 MQTT 的自动重连并设clean_sessionfalse这样重连后能收到离线消息。Broker 端检查max_connections配置。4.3 读出来的数值明显不对现象温度显示 6553.5 或者负数明显超出量程。原因字节序搞反了或者数据类型选错了。float32 的 ABCD 和 CDAB 读出来完全是两个数。解决先用 Modbus Poll 这类工具读原始寄存器值手动算一遍。比如两个寄存器值是 0x41C8 和 0x0000ABCD 排列是 0x41C80000转 float 是 25.0。CDAB 排列是 0x000041C8转出来就很小。确认字节序后再改配置。4.4 变化上报导致数据「跳变」丢失现象订阅端看到的数据偶尔跳一下或者中间某个值没收到。原因变化阈值设得太小传感器噪声导致频繁触发或者 QoS 0 在弱网下丢包。解决加死区过滤比如温度变化小于 0.5℃ 不上报。关键数据用 QoS 1。如果还是丢检查 Broker 的max_queued_messages是不是太小。4.5 多从站轮询时响应慢现象从站多了以后每个点的刷新周期变长实时性下降。原因串口是半双工主站逐个轮询从站多了总时间线性增长。解决提高波特率19200 或 38400合并连续寄存器一次读取把不重要的点降频。实在不行分多个串口并行。5. 进阶技巧用规则引擎做数据清洗与告警联动网关把数据发到 MQTT 只是第一步真正产生价值的是后面的处理。我一般会在 Broker 后面挂一个规则引擎做三件事数据清洗、阈值告警、指令联动。数据清洗是把原始值转成业务语义。比如寄存器值 0/1 转成「运行/停止」数值超量程标记为无效。规则引擎里用 SQL 风格的语法就能写SELECT payload.temp AS temperature, CASE WHEN payload.temp 80 THEN overheat WHEN payload.temp 10 THEN low ELSE normal END AS temp_status FROM factory/line1/temp这段规则订阅温度 Topic输出带状态标签的新消息。EMQX 的规则引擎、Node-RED 都能做类似的事。清洗后的数据再入库或推给看板前端就不用写判断逻辑了。阈值告警是另一个高频需求。规则里判断超过阈值就发到告警 Topic再由通知服务发邮件或短信。注意加防抖不然温度在阈值附近波动会疯狂告警。常见做法是连续 N 次超限才触发或者加一个回差带。指令联动是反向控制。比如检测到温度超限自动发指令停电机。这里要特别小心自动控制指令必须加确认机制发出去后检查设备状态是否真的变了没变就重试或告警。我见过自动停机和手动开机打架设备来回启停的翻车现场。验证规则是否生效用 mosquitto_sub 订阅输出 Topic 就行。调试阶段建议把规则输出单独开一个 Topic别直接混在生产数据里。# 订阅清洗后的数据 mosquitto_sub -h 127.0.0.1 -t factory/line1/cleaned/# -v # 订阅告警 mosquitto_sub -h 127.0.0.1 -t factory/alarm/# -v从那以后我每次上规则引擎都强制先在测试环境跑 24 小时确认没有误报和指令冲突再切生产。这套网关加规则引擎的组合能把老设备的接入周期从两周压到两天。希望帮到你。本文还有配套的精品资源点击获取