尧图网站开发 
简介这份资源面向工业自动化与物联网方向的开发者、运维人员及技术学习者聚焦Modbus设备接入与MQTT消息发布订阅的完整实现解决传统工业设备数据上云、协议转换与远程监控的落地问题。压缩包共32个文件约273KB以JavaScript源码为主体辅以Markdown说明文档、YAML配置、JSON数据、Dockerfile与Shell脚本覆盖核心逻辑、配置管理与容器化部署等环节。内容围绕Modbus协议解析、串口通信、数据转换、设备配置管理及实时数据采集展开并借助MQTT消息队列实现设备间低带宽可靠通信支持跨平台运行。项目采用开源方式组织目录结构清晰便于按模块阅读与二次开发。目前已有69人学习下载适合希望快速理解Modbus转MQTT网关实现思路、搭建物联网数据采集原型或进行设备远程监控实践的技术人员参考。1. 工业网关这条链路Modbus 采上来MQTT 发出去中间到底发生了什么车间里一台老 PLC 跑着 Modbus RTU波特率 9600挂在 RS485 总线上老板要在大屏上看实时产量还要手机能远程监控。你手上没有原厂网关只有一块带串口的 Linux 开发板或者一台工控机。这个标题讲的就是把这件事跑通用 Modbus 协议从设备侧采集数据做数据转换再通过 MQTT 消息队列发布出去同时管好设备配置和远程监控。它解决的是「异构设备怎么统一接入」的问题适合做工业物联网落地、设备接入层开发、以及需要跨平台部署网关的工程师。核心链路只有三段串口通信拿字节、Modbus 解析成结构化数据、MQTT 发布订阅送出去。听起来简单但每一段都有翻车点下面按能复现的路径拆开讲。2. 先跑通最小链路串口通信 Modbus RTU 采集2.1 为什么先啃串口通信而不是直接上 Modbus TCP很多人一上来就想用 Modbus TCP觉得网口方便。但现场大量存量设备是 RS485/RS232 串口Modbus RTU 才是绕不开的基本功。串口通信的本质是字节流你发一串十六进制设备回一串十六进制中间没有 HTTP 那种自描述结构全靠约定。Modbus RTU 报文结构是「从站地址 功能码 数据 CRC 校验」帧与帧之间靠 3.5 个字符时间的静默间隔区分。这个静默间隔是玄学重灾区——波特率 9600 时约 4ms波特率 115200 时不到 0.4ms很多 USB 转串口芯片根本卡不准导致粘包。所以第一步不是写 Modbus 逻辑而是确认串口能稳定收发原始字节。Linux 下串口设备通常是/dev/ttyUSB0或/dev/ttyS0。先用命令行确认参数别急着写代码# 查看串口参数波特率、数据位、校验位、停止位 stty -F /dev/ttyUSB0 -a # 设置为 Modbus RTU 常见参数9600 8N1无流控 stty -F /dev/ttyUSB0 9600 cs8 -cstopb -parenb -crtscts # 用十六进制方式发一帧读保持寄存器的请求从站1功能码03起始0数量2 # 01 03 00 00 00 02 CRC实际 CRC 需计算这里演示发送动作 printf \x01\x03\x00\x00\x00\x02\xC4\x0B /dev/ttyUSB0 # 另开终端读返回的原始字节 xxd -g 1 /dev/ttyUSB0stty里cs8是 8 数据位-cstopb是 1 位停止位-parenb是无校验-crtscts关硬件流控。Modbus RTU 默认就是 8N1但现场偶尔遇到 7E1 的老设备参数对不上就是收不到任何回应。printf那行发的是原始字节\x转义在 bash 里需要printf支持注意 CRC 两个字节是低字节在前。xxd -g 1按单字节分组显示方便你肉眼核对报文。这一步能收到正确回应说明物理层和串口参数没问题再往上写 Modbus 解析才有意义。2.2 用 Python 把 Modbus RTU 读寄存器跑成可复用函数命令行验证完落到代码。Python 生态里pymodbus和minimalmodbus都常用但做网关我倾向自己封装串口读写因为要控制超时和重试库的默认行为不一定贴合现场。下面是一个最小可用的 Modbus RTU 读保持寄存器实现import serial import struct import time def crc16_modbus(data: bytes) - bytes: 计算 Modbus CRC16返回低字节在前的两字节 crc 0xFFFF for b in data: crc ^ b for _ in range(8): if crc 0x0001: crc (crc 1) ^ 0xA001 else: crc 1 return struct.pack(H, crc) def read_holding_registers(port, slave_id, start_addr, count, baudrate9600, timeout1.0): 读保持寄存器功能码 0x03 # 构造请求从站地址 功能码 起始地址(2字节) 数量(2字节) req struct.pack(BBHH, slave_id, 0x03, start_addr, count) req crc16_modbus(req) with serial.Serial(port, baudrate, bytesize8, parityN, stopbits1, timeouttimeout) as ser: ser.reset_input_buffer() ser.write(req) # 预期返回长度地址1 功能码1 字节数1 数据2*count CRC2 expected 5 count * 2 resp ser.read(expected) if len(resp) expected: raise TimeoutError(f响应不完整收到 {len(resp)} 字节) # 校验 CRC if crc16_modbus(resp[:-2]) ! resp[-2:]: raise ValueError(CRC 校验失败) # 解析数据区 byte_count resp[2] values struct.unpack( H * (byte_count // 2), resp[3:3 byte_count]) return values if __name__ __main__: # 读从站1起始地址0读2个寄存器 print(read_holding_registers(/dev/ttyUSB0, 1, 0, 2))crc16_modbus里多项式0xA001是 Modbus 规定的反向多项式初值0xFFFF算完用H小端打包因为 Modbus 报文里 CRC 低字节在前。struct.pack(BBHH, ...)用大端因为 Modbus 协议字段本身是大端。ser.reset_input_buffer()很关键清掉上一次残留否则粘包时你会读到上一帧的尾巴。expected长度算错是新手常见问题功能码 03 的响应是「地址 功能码 字节数 N 字节数据 2 字节 CRC」所以是 5 count*2。超时设 1 秒对 9600 波特率足够太短会误判太长会拖慢轮询。这个函数跑通你就有了采集侧的地基。2.3 轮询多个从站时怎么避免总线冲突一条 RS485 总线上挂多个从站是常态。Modbus 是主从协议同一时刻只能有一个主站发问从站不会主动说话。所以轮询必须串行不能并发。常见做法是维护一个从站列表逐个读每个之间留一点间隔import time slaves [ {id: 1, start: 0, count: 4}, {id: 2, start: 0, count: 2}, {id: 3, start: 10, count: 6}, ] def poll_all(port): result {} for s in slaves: try: vals read_holding_registers(port, s[id], s[start], s[count]) result[s[id]] vals except Exception as e: result[s[id]] None print(f从站 {s[id]} 读取失败: {e}) time.sleep(0.05) # 帧间间隔给总线留恢复时间 return resulttime.sleep(0.05)是保守值实际按波特率和帧长算50ms 对 9600 波特率足够。失败不要直接抛异常中断整个轮询记 None 继续下一个否则一个从站掉线会拖垮整条链路。这里有个血泪经验RS485 接线 A/B 反了不会报错只会一直超时排查时先拿万用表量差分电压别死磕代码。3. 数据转换把 Modbus 寄存器变成有业务含义的 MQTT 消息3.1 寄存器到物理量的映射表怎么设计Modbus 读回来的是 16 位无符号整数但业务要的是温度、转速、产量。中间要做数据转换缩放、字节序、位拆分。这一步没有标准答案全靠设备手册。我一般用一张配置表驱动而不是把转换逻辑写死在代码里字段名从站寄存器地址类型缩放系数单位说明temperature10int160.1℃有符号需处理负温speed11uint161rpm直接读status_bits12bitfield--bit0 运行bit1 报警total_count20uint321件两个寄存器拼int16和uint16的区别在于最高位是否当符号位温度这类可能为负的必须按有符号解析否则 -5℃ 会读成 65531。uint32跨两个寄存器字节序有 ABCD 和 CDAB 两种三菱、西门子、汇川各不相同手册上一般写「高字在前」或「低字在前」。bitfield是把一个寄存器的 16 个 bit 当 16 个开关用状态字最常见。这张表建议存成 JSON 或 YAML网关启动时加载改点位不用改代码。3.2 用配置驱动转换避免每接一个设备就改代码把上面的表落成 JSON转换函数读配置执行import json CONFIG { device_id: plc_line1, slave_id: 1, points: [ {name: temperature, addr: 0, type: int16, scale: 0.1, unit: C}, {name: speed, addr: 1, type: uint16, scale: 1, unit: rpm}, {name: status_bits, addr: 2, type: bitfield, bits: { running: 0, alarm: 1, ready: 2}} ] } def convert(raw_values, points): raw_values 是按地址顺序读回的寄存器列表 out {} for p in points: idx p[addr] if p[type] int16: v raw_values[idx] if v 32767: # 处理有符号 v - 65536 out[p[name]] round(v * p[scale], 2) elif p[type] uint16: out[p[name]] raw_values[idx] * p[scale] elif p[type] bitfield: bits raw_values[idx] for bname, bpos in p[bits].items(): out[bname] bool(bits (1 bpos)) return outint16那段if v 32767: v - 65536是补码还原等价于struct.unpack(h, ...)但直接算更直观。round(v * p[scale], 2)保留两位小数避免浮点噪声。bitfield用bits (1 bpos)取某一位返回布尔值。这样一份配置对应一台设备新增设备只加配置不改代码这是网关能规模化的前提。注意addr这里假设读回的列表下标和地址一一对应如果跳着读要改成按地址查字典。3.3 数据转换里最容易错的字节序和缩放字节序翻车率极高。同一个 32 位值ABCD 和 CDAB 读出来能差 65536 倍。判断方法拿一个已知值比如设备显示 1000读两个寄存器看哪个组合能还原。缩放系数也别想当然手册写「0.1℃/bit」就是乘 0.1写「实际值 寄存器值 / 10」也是乘 0.1但写「放大 10 倍存储」就要除 10。我一般会在转换后加一个合理性校验比如温度超出 -50~200℃ 就标记异常防止缩放写错导致大屏显示离谱数字。4. MQTT 消息发布订阅把转换后的数据送出去4.1 MQTT 主题设计别用一个大 topic 塞所有数据MQTT 是发布订阅模型核心是 topic。topic 设计不好后期订阅端会痛不欲生。常见做法是按层级组织factory/{workshop}/{line}/{device_id}/data # 实时数据 factory/{workshop}/{line}/{device_id}/status # 在线状态 factory/{workshop}/{line}/{device_id}/cmd # 下行指令用/分层订阅端可以用通配符匹配单层、#匹配多层。比如订阅factory/workshop1//plc_line1/data就能拿到该设备所有产线数据。别把所有设备数据塞进一个 topic那样订阅端要自己过滤流量也浪费。QoS 选择上实时数据用 QoS 0 够了丢了下一轮补状态和指令用 QoS 1保证至少到达。retain 标志对状态类消息有用新订阅者能立刻拿到最后状态但实时数据不要 retain否则新订阅者会被旧数据刷屏。4.2 用 paho-mqtt 把采集数据发布出去Python 侧用paho-mqtt最省事import paho.mqtt.client as mqtt import json import time BROKER 127.0.0.1 PORT 1883 TOPIC_TPL factory/workshop1/line1/{device}/data client mqtt.Client(client_idgateway_01) client.connect(BROKER, PORT, keepalive60) client.loop_start() # 后台线程处理网络收发 def publish_data(device_id, payload): topic TOPIC_TPL.format(devicedevice_id) # QoS 0不 retain实时数据 client.publish(topic, json.dumps(payload), qos0, retainFalse) # 模拟一轮采集发布 while True: raw read_holding_registers(/dev/ttyUSB0, 1, 0, 3) data convert(raw, CONFIG[points]) data[ts] int(time.time() * 1000) publish_data(CONFIG[device_id], data) time.sleep(1)client.loop_start()启动后台网络线程主线程只管采集和发布不用手动调loop()。json.dumps把字典序列化成字符串订阅端解析方便。ts加毫秒时间戳方便订阅端判断数据新鲜度。qos0对应「最多一次」适合高频实时数据。如果 broker 断线paho会自动重连但重连期间 publish 的消息会丢对可靠性要求高的场景要自己加本地缓存队列。4.3 订阅端怎么验证数据真的通了发布出去不算完得验证。命令行用mosquitto_sub最直接# 订阅所有设备数据-v 显示 topic mosquitto_sub -h 127.0.0.1 -p 1883 -t factory/workshop1/line1//data -v # 订阅单设备只看 plc_line1 mosquitto_sub -h 127.0.0.1 -t factory/workshop1/line1/plc_line1/data-v会同时打印 topic 和 payload方便确认路由对不对。如果收不到先查 broker 是否在跑、端口是否监听、topic 拼写是否一致。MQTT 的 topic 大小写敏感Data和data是两个 topic。还有个坑client_id 重复会导致互相踢下线网关的 client_id 要唯一别用默认随机值又手动写死成一样的。5. 避坑与排查网关跑起来后最容易翻车的 5 个点5.1 串口能发不能收或收到乱码现象write成功但read超时或收到一堆0xFF。原因通常是串口参数不匹配波特率、校验位、A/B 线接反、或者 USB 转串口驱动问题。解决先用stty -a核对参数再用示波器或万用表量 RS485 差分电压A-B 之间应有 2V 左右压差。驱动问题换ftdi或ch340官方驱动别用系统自带。5.2 Modbus 响应 CRC 校验失败现象收到字节数对但 CRC 对不上。原因多是粘包——上一帧的尾巴混进来了或者帧间静默时间不够。解决读之前reset_input_buffer()读的时候按预期长度精确读不要用read_all()。如果还不行在发送前加 10ms 延时给总线恢复时间。5.3 MQTT 发布成功但订阅端收不到现象publish返回成功订阅端没消息。原因可能是 topic 不匹配、QoS 设置导致消息被丢弃、或者 broker 权限限制。解决用mosquitto_sub -t #订阅所有 topic 看有没有数据确认 topic 拼写。如果 broker 配了 ACL检查网关的 username 有没有发布权限。5.4 数据转换后数值离谱现象温度显示 6553.1℃。原因有符号数没处理或者缩放系数用反。解决对照手册确认类型和系数加合理性范围校验超范围就丢弃并告警别让脏数据进大屏。5.5 网关跑几天后内存涨或卡死现象进程内存持续增长或串口无响应。原因串口句柄没释放、MQTT 消息积压、或异常没捕获导致线程挂掉。解决串口用with上下文管理MQTT 发布加异常捕获长时间运行加看门狗定时重启。我一般会加一个心跳 topic网关每分钟发一次订阅端收不到就告警。6. 进阶让网关支持远程配置和跨平台部署6.1 用 MQTT 下行指令动态改采集配置网关部署到现场后改点位不该跑现场。做法是订阅一个配置 topic收到新配置就热加载import json def on_message(client, userdata, msg): if msg.topic.endswith(/config): new_cfg json.loads(msg.payload) # 校验后替换全局配置 global CONFIG CONFIG new_cfg print(配置已热更新) client.subscribe(factory/workshop1/line1/plc_line1/config, qos1) client.on_message on_messageon_message是 paho 的回调收到消息自动触发。配置里可以带点位表、轮询周期、缩放系数。注意热更新时要加锁避免采集线程读到半截配置。校验不能省错误的配置会让网关读错地址。6.2 跨平台部署Linux 工控机和 Windows 现场机怎么统一同一套 Python 代码在 Linux 和 Windows 上跑差异主要在串口设备名和路径。Linux 是/dev/ttyUSB0Windows 是COM3。用配置区分import platform def get_serial_port(cfg): if platform.system() Windows: return cfg[serial_port_win] # 如 COM3 return cfg[serial_port_linux] # 如 /dev/ttyUSB0打包用pyinstaller可以出单文件可执行程序Windows 和 Linux 各打一份。依赖里pyserial和paho-mqtt都跨平台。如果现场是 ARM 开发板注意pyserial的 wheel 要选对架构或者直接源码安装。部署时把配置文件和可执行文件放一起改配置不用重新打包。6.3 一个验证网关稳定性的土办法别等上线才发现问题。我习惯在实验室做 72 小时连续跑接一个 Modbus 从站模拟器比如 Modbus Slave 软件让网关按 1 秒周期采集发布同时用脚本订阅并记录消息数。72 小时后看两个指标消息总数是否符合预期1 秒 1 条72 小时约 25.9 万条以及内存占用是否平稳。如果消息数少了说明有丢包或异常中断内存涨了说明有泄漏。这个土办法帮我提前发现过串口句柄没释放的问题比上线后半夜被叫起来强。做网关这行后悔药就是提前压测希望帮到你。本文还有配套的精品资源点击获取 
简介这份资源面向工业自动化与物联网方向的开发者、运维人员及技术学习者聚焦Modbus设备接入与MQTT消息发布订阅的完整实现解决传统工业设备数据上云、协议转换与远程监控的落地问题。压缩包共32个文件约273KB以JavaScript源码为主体辅以Markdown说明文档、YAML配置、JSON数据、Dockerfile与Shell脚本覆盖核心逻辑、配置管理与容器化部署等环节。内容围绕Modbus协议解析、串口通信、数据转换、设备配置管理及实时数据采集展开并借助MQTT消息队列实现设备间低带宽可靠通信支持跨平台运行。项目采用开源方式组织目录结构清晰便于按模块阅读与二次开发。目前已有69人学习下载适合希望快速理解Modbus转MQTT网关实现思路、搭建物联网数据采集原型或进行设备远程监控实践的技术人员参考。1. 工业网关这条链路Modbus 采上来MQTT 发出去中间到底发生了什么车间里一台老 PLC 跑着 Modbus RTU波特率 9600挂在 RS485 总线上老板要在大屏上看实时产量还要手机能远程监控。你手上没有原厂网关只有一块带串口的 Linux 开发板或者一台工控机。这个标题讲的就是把这件事跑通用 Modbus 协议从设备侧采集数据做数据转换再通过 MQTT 消息队列发布出去同时管好设备配置和远程监控。它解决的是「异构设备怎么统一接入」的问题适合做工业物联网落地、设备接入层开发、以及需要跨平台部署网关的工程师。核心链路只有三段串口通信拿字节、Modbus 解析成结构化数据、MQTT 发布订阅送出去。听起来简单但每一段都有翻车点下面按能复现的路径拆开讲。2. 先跑通最小链路串口通信 Modbus RTU 采集2.1 为什么先啃串口通信而不是直接上 Modbus TCP很多人一上来就想用 Modbus TCP觉得网口方便。但现场大量存量设备是 RS485/RS232 串口Modbus RTU 才是绕不开的基本功。串口通信的本质是字节流你发一串十六进制设备回一串十六进制中间没有 HTTP 那种自描述结构全靠约定。Modbus RTU 报文结构是「从站地址 功能码 数据 CRC 校验」帧与帧之间靠 3.5 个字符时间的静默间隔区分。这个静默间隔是玄学重灾区——波特率 9600 时约 4ms波特率 115200 时不到 0.4ms很多 USB 转串口芯片根本卡不准导致粘包。所以第一步不是写 Modbus 逻辑而是确认串口能稳定收发原始字节。Linux 下串口设备通常是/dev/ttyUSB0或/dev/ttyS0。先用命令行确认参数别急着写代码# 查看串口参数波特率、数据位、校验位、停止位 stty -F /dev/ttyUSB0 -a # 设置为 Modbus RTU 常见参数9600 8N1无流控 stty -F /dev/ttyUSB0 9600 cs8 -cstopb -parenb -crtscts # 用十六进制方式发一帧读保持寄存器的请求从站1功能码03起始0数量2 # 01 03 00 00 00 02 CRC实际 CRC 需计算这里演示发送动作 printf \x01\x03\x00\x00\x00\x02\xC4\x0B /dev/ttyUSB0 # 另开终端读返回的原始字节 xxd -g 1 /dev/ttyUSB0stty里cs8是 8 数据位-cstopb是 1 位停止位-parenb是无校验-crtscts关硬件流控。Modbus RTU 默认就是 8N1但现场偶尔遇到 7E1 的老设备参数对不上就是收不到任何回应。printf那行发的是原始字节\x转义在 bash 里需要printf支持注意 CRC 两个字节是低字节在前。xxd -g 1按单字节分组显示方便你肉眼核对报文。这一步能收到正确回应说明物理层和串口参数没问题再往上写 Modbus 解析才有意义。2.2 用 Python 把 Modbus RTU 读寄存器跑成可复用函数命令行验证完落到代码。Python 生态里pymodbus和minimalmodbus都常用但做网关我倾向自己封装串口读写因为要控制超时和重试库的默认行为不一定贴合现场。下面是一个最小可用的 Modbus RTU 读保持寄存器实现import serial import struct import time def crc16_modbus(data: bytes) - bytes: 计算 Modbus CRC16返回低字节在前的两字节 crc 0xFFFF for b in data: crc ^ b for _ in range(8): if crc 0x0001: crc (crc 1) ^ 0xA001 else: crc 1 return struct.pack(H, crc) def read_holding_registers(port, slave_id, start_addr, count, baudrate9600, timeout1.0): 读保持寄存器功能码 0x03 # 构造请求从站地址 功能码 起始地址(2字节) 数量(2字节) req struct.pack(BBHH, slave_id, 0x03, start_addr, count) req crc16_modbus(req) with serial.Serial(port, baudrate, bytesize8, parityN, stopbits1, timeouttimeout) as ser: ser.reset_input_buffer() ser.write(req) # 预期返回长度地址1 功能码1 字节数1 数据2*count CRC2 expected 5 count * 2 resp ser.read(expected) if len(resp) expected: raise TimeoutError(f响应不完整收到 {len(resp)} 字节) # 校验 CRC if crc16_modbus(resp[:-2]) ! resp[-2:]: raise ValueError(CRC 校验失败) # 解析数据区 byte_count resp[2] values struct.unpack( H * (byte_count // 2), resp[3:3 byte_count]) return values if __name__ __main__: # 读从站1起始地址0读2个寄存器 print(read_holding_registers(/dev/ttyUSB0, 1, 0, 2))crc16_modbus里多项式0xA001是 Modbus 规定的反向多项式初值0xFFFF算完用H小端打包因为 Modbus 报文里 CRC 低字节在前。struct.pack(BBHH, ...)用大端因为 Modbus 协议字段本身是大端。ser.reset_input_buffer()很关键清掉上一次残留否则粘包时你会读到上一帧的尾巴。expected长度算错是新手常见问题功能码 03 的响应是「地址 功能码 字节数 N 字节数据 2 字节 CRC」所以是 5 count*2。超时设 1 秒对 9600 波特率足够太短会误判太长会拖慢轮询。这个函数跑通你就有了采集侧的地基。2.3 轮询多个从站时怎么避免总线冲突一条 RS485 总线上挂多个从站是常态。Modbus 是主从协议同一时刻只能有一个主站发问从站不会主动说话。所以轮询必须串行不能并发。常见做法是维护一个从站列表逐个读每个之间留一点间隔import time slaves [ {id: 1, start: 0, count: 4}, {id: 2, start: 0, count: 2}, {id: 3, start: 10, count: 6}, ] def poll_all(port): result {} for s in slaves: try: vals read_holding_registers(port, s[id], s[start], s[count]) result[s[id]] vals except Exception as e: result[s[id]] None print(f从站 {s[id]} 读取失败: {e}) time.sleep(0.05) # 帧间间隔给总线留恢复时间 return resulttime.sleep(0.05)是保守值实际按波特率和帧长算50ms 对 9600 波特率足够。失败不要直接抛异常中断整个轮询记 None 继续下一个否则一个从站掉线会拖垮整条链路。这里有个血泪经验RS485 接线 A/B 反了不会报错只会一直超时排查时先拿万用表量差分电压别死磕代码。3. 数据转换把 Modbus 寄存器变成有业务含义的 MQTT 消息3.1 寄存器到物理量的映射表怎么设计Modbus 读回来的是 16 位无符号整数但业务要的是温度、转速、产量。中间要做数据转换缩放、字节序、位拆分。这一步没有标准答案全靠设备手册。我一般用一张配置表驱动而不是把转换逻辑写死在代码里字段名从站寄存器地址类型缩放系数单位说明temperature10int160.1℃有符号需处理负温speed11uint161rpm直接读status_bits12bitfield--bit0 运行bit1 报警total_count20uint321件两个寄存器拼int16和uint16的区别在于最高位是否当符号位温度这类可能为负的必须按有符号解析否则 -5℃ 会读成 65531。uint32跨两个寄存器字节序有 ABCD 和 CDAB 两种三菱、西门子、汇川各不相同手册上一般写「高字在前」或「低字在前」。bitfield是把一个寄存器的 16 个 bit 当 16 个开关用状态字最常见。这张表建议存成 JSON 或 YAML网关启动时加载改点位不用改代码。3.2 用配置驱动转换避免每接一个设备就改代码把上面的表落成 JSON转换函数读配置执行import json CONFIG { device_id: plc_line1, slave_id: 1, points: [ {name: temperature, addr: 0, type: int16, scale: 0.1, unit: C}, {name: speed, addr: 1, type: uint16, scale: 1, unit: rpm}, {name: status_bits, addr: 2, type: bitfield, bits: { running: 0, alarm: 1, ready: 2}} ] } def convert(raw_values, points): raw_values 是按地址顺序读回的寄存器列表 out {} for p in points: idx p[addr] if p[type] int16: v raw_values[idx] if v 32767: # 处理有符号 v - 65536 out[p[name]] round(v * p[scale], 2) elif p[type] uint16: out[p[name]] raw_values[idx] * p[scale] elif p[type] bitfield: bits raw_values[idx] for bname, bpos in p[bits].items(): out[bname] bool(bits (1 bpos)) return outint16那段if v 32767: v - 65536是补码还原等价于struct.unpack(h, ...)但直接算更直观。round(v * p[scale], 2)保留两位小数避免浮点噪声。bitfield用bits (1 bpos)取某一位返回布尔值。这样一份配置对应一台设备新增设备只加配置不改代码这是网关能规模化的前提。注意addr这里假设读回的列表下标和地址一一对应如果跳着读要改成按地址查字典。3.3 数据转换里最容易错的字节序和缩放字节序翻车率极高。同一个 32 位值ABCD 和 CDAB 读出来能差 65536 倍。判断方法拿一个已知值比如设备显示 1000读两个寄存器看哪个组合能还原。缩放系数也别想当然手册写「0.1℃/bit」就是乘 0.1写「实际值 寄存器值 / 10」也是乘 0.1但写「放大 10 倍存储」就要除 10。我一般会在转换后加一个合理性校验比如温度超出 -50~200℃ 就标记异常防止缩放写错导致大屏显示离谱数字。4. MQTT 消息发布订阅把转换后的数据送出去4.1 MQTT 主题设计别用一个大 topic 塞所有数据MQTT 是发布订阅模型核心是 topic。topic 设计不好后期订阅端会痛不欲生。常见做法是按层级组织factory/{workshop}/{line}/{device_id}/data # 实时数据 factory/{workshop}/{line}/{device_id}/status # 在线状态 factory/{workshop}/{line}/{device_id}/cmd # 下行指令用/分层订阅端可以用通配符匹配单层、#匹配多层。比如订阅factory/workshop1//plc_line1/data就能拿到该设备所有产线数据。别把所有设备数据塞进一个 topic那样订阅端要自己过滤流量也浪费。QoS 选择上实时数据用 QoS 0 够了丢了下一轮补状态和指令用 QoS 1保证至少到达。retain 标志对状态类消息有用新订阅者能立刻拿到最后状态但实时数据不要 retain否则新订阅者会被旧数据刷屏。4.2 用 paho-mqtt 把采集数据发布出去Python 侧用paho-mqtt最省事import paho.mqtt.client as mqtt import json import time BROKER 127.0.0.1 PORT 1883 TOPIC_TPL factory/workshop1/line1/{device}/data client mqtt.Client(client_idgateway_01) client.connect(BROKER, PORT, keepalive60) client.loop_start() # 后台线程处理网络收发 def publish_data(device_id, payload): topic TOPIC_TPL.format(devicedevice_id) # QoS 0不 retain实时数据 client.publish(topic, json.dumps(payload), qos0, retainFalse) # 模拟一轮采集发布 while True: raw read_holding_registers(/dev/ttyUSB0, 1, 0, 3) data convert(raw, CONFIG[points]) data[ts] int(time.time() * 1000) publish_data(CONFIG[device_id], data) time.sleep(1)client.loop_start()启动后台网络线程主线程只管采集和发布不用手动调loop()。json.dumps把字典序列化成字符串订阅端解析方便。ts加毫秒时间戳方便订阅端判断数据新鲜度。qos0对应「最多一次」适合高频实时数据。如果 broker 断线paho会自动重连但重连期间 publish 的消息会丢对可靠性要求高的场景要自己加本地缓存队列。4.3 订阅端怎么验证数据真的通了发布出去不算完得验证。命令行用mosquitto_sub最直接# 订阅所有设备数据-v 显示 topic mosquitto_sub -h 127.0.0.1 -p 1883 -t factory/workshop1/line1//data -v # 订阅单设备只看 plc_line1 mosquitto_sub -h 127.0.0.1 -t factory/workshop1/line1/plc_line1/data-v会同时打印 topic 和 payload方便确认路由对不对。如果收不到先查 broker 是否在跑、端口是否监听、topic 拼写是否一致。MQTT 的 topic 大小写敏感Data和data是两个 topic。还有个坑client_id 重复会导致互相踢下线网关的 client_id 要唯一别用默认随机值又手动写死成一样的。5. 避坑与排查网关跑起来后最容易翻车的 5 个点5.1 串口能发不能收或收到乱码现象write成功但read超时或收到一堆0xFF。原因通常是串口参数不匹配波特率、校验位、A/B 线接反、或者 USB 转串口驱动问题。解决先用stty -a核对参数再用示波器或万用表量 RS485 差分电压A-B 之间应有 2V 左右压差。驱动问题换ftdi或ch340官方驱动别用系统自带。5.2 Modbus 响应 CRC 校验失败现象收到字节数对但 CRC 对不上。原因多是粘包——上一帧的尾巴混进来了或者帧间静默时间不够。解决读之前reset_input_buffer()读的时候按预期长度精确读不要用read_all()。如果还不行在发送前加 10ms 延时给总线恢复时间。5.3 MQTT 发布成功但订阅端收不到现象publish返回成功订阅端没消息。原因可能是 topic 不匹配、QoS 设置导致消息被丢弃、或者 broker 权限限制。解决用mosquitto_sub -t #订阅所有 topic 看有没有数据确认 topic 拼写。如果 broker 配了 ACL检查网关的 username 有没有发布权限。5.4 数据转换后数值离谱现象温度显示 6553.1℃。原因有符号数没处理或者缩放系数用反。解决对照手册确认类型和系数加合理性范围校验超范围就丢弃并告警别让脏数据进大屏。5.5 网关跑几天后内存涨或卡死现象进程内存持续增长或串口无响应。原因串口句柄没释放、MQTT 消息积压、或异常没捕获导致线程挂掉。解决串口用with上下文管理MQTT 发布加异常捕获长时间运行加看门狗定时重启。我一般会加一个心跳 topic网关每分钟发一次订阅端收不到就告警。6. 进阶让网关支持远程配置和跨平台部署6.1 用 MQTT 下行指令动态改采集配置网关部署到现场后改点位不该跑现场。做法是订阅一个配置 topic收到新配置就热加载import json def on_message(client, userdata, msg): if msg.topic.endswith(/config): new_cfg json.loads(msg.payload) # 校验后替换全局配置 global CONFIG CONFIG new_cfg print(配置已热更新) client.subscribe(factory/workshop1/line1/plc_line1/config, qos1) client.on_message on_messageon_message是 paho 的回调收到消息自动触发。配置里可以带点位表、轮询周期、缩放系数。注意热更新时要加锁避免采集线程读到半截配置。校验不能省错误的配置会让网关读错地址。6.2 跨平台部署Linux 工控机和 Windows 现场机怎么统一同一套 Python 代码在 Linux 和 Windows 上跑差异主要在串口设备名和路径。Linux 是/dev/ttyUSB0Windows 是COM3。用配置区分import platform def get_serial_port(cfg): if platform.system() Windows: return cfg[serial_port_win] # 如 COM3 return cfg[serial_port_linux] # 如 /dev/ttyUSB0打包用pyinstaller可以出单文件可执行程序Windows 和 Linux 各打一份。依赖里pyserial和paho-mqtt都跨平台。如果现场是 ARM 开发板注意pyserial的 wheel 要选对架构或者直接源码安装。部署时把配置文件和可执行文件放一起改配置不用重新打包。6.3 一个验证网关稳定性的土办法别等上线才发现问题。我习惯在实验室做 72 小时连续跑接一个 Modbus 从站模拟器比如 Modbus Slave 软件让网关按 1 秒周期采集发布同时用脚本订阅并记录消息数。72 小时后看两个指标消息总数是否符合预期1 秒 1 条72 小时约 25.9 万条以及内存占用是否平稳。如果消息数少了说明有丢包或异常中断内存涨了说明有泄漏。这个土办法帮我提前发现过串口句柄没释放的问题比上线后半夜被叫起来强。做网关这行后悔药就是提前压测希望帮到你。本文还有配套的精品资源点击获取