从ModbusRTU到WebServer：手把手搭建工业数据采集服务

现场工程师最怕的，不是设备不联网，而是设备明明在身边，数据却拿不出来。

很多工厂车间里，PLC、电表、温控仪、变频器都支持 ModbusRTU，打开面板就能看到数值，可一旦要把这些数据送到 MES、ERP 或者网页大屏上，问题就来了：串口线怎么接、波特率对不对、寄存器地址怎么映射、数据到了服务器之后怎么存怎么展示，每一步都可能卡住。

这篇文章不讨论那种“买一台工业网关，配置一下云平台”的商业方案，而是从开发者视角，把从 ModbusRTU 串口采集到 WebServer 对外提供 HTTP 接口的完整链路讲清楚。读完你可以搭建一个最小可用、可扩展的工业数据采集服务，也能搞明白现场调试时最容易踩的坑在哪里。

1. 为什么要自己做 ModbusRTU 数据采集服务

很多人会问：现在工业网关那么成熟，插上电就能用，为什么还要自己写采集程序？

这个问题的答案，取决于你的项目阶段和规模。

如果你负责的是几十台设备的产线级采集，采购工业网关是合理的；但如果你处于设备调试阶段、样机验证阶段，或者需要对采集逻辑做深度定制，比如多套协议转换、特殊的数据清洗规则、跟内部系统深度集成，那么一套自己可控的采集程序反而效率更高、成本更低。

自己做采集服务的核心价值有三点：

第一，掌握了 Modbus 协议的真实细节。设备返回的数据不是一个“数字”那么简单，它涉及功能码、寄存器地址、字节序、数据类型转换、CRC 校验。用网关时这些问题被封装掉了，一旦设备数据不对，你根本不知道是接线问题、波特率问题还是字节序问题。

第二，可以自由对接内部系统。网关厂商的 API 不一定满足你的数据格式要求，自己写服务可以把数据以 JSON 形式直接推给内部 MES，也可以保证插入数据库的数据结构与业务表完全对齐。

第三，排错成本低。现场最贵的不是设备，是等待。自己掌握采集链路每一层的行为，出现问题可以先定位到串口层、协议层还是应用层，不用打一圈电话找厂商支持。

但也要泼一盆冷水：自己写采集程序，不代表要自己从零实现 Modbus 协议栈。现在主流的编程语言都有成熟的 Modbus 库，比如 Python 的pymodbus、Java 的jamod、C# 的NModbus。我们的核心工作是理解工作机制，然后合理使用这些库。

2. ModbusRTU 核心概念与适用场景

2.1 ModbusRTU 是什么

Modbus 是一种应用层报文协议，诞生于 1979 年，最初为 PLC 通信而设计。它有两种常见物理形态：ModbusRTU 走串口（RS232/RS485），ModbusTCP 走以太网。

ModbusRTU 的报文格式非常紧凑，它以二进制形式传输数据，报文结构为：

地址码(1字节) + 功能码(1字节) + 数据段(N字节) + CRC校验(2字节)

一条完整的 RTU 报文最多也就 256 字节。这意味着它在 9600 波特率的 RS485 总线上也能高效工作，这对于长距离、多节点的工业环境非常重要。

2.2 主站与从站的关系

ModbusRTU 网络是典型的主从结构：

• 主站（Master）：发出请求，一个网络上只能有一个。
• 从站（Slave）：响应请求，常见的有 1-247 个编号，每个从站必须有一个唯一的地址。

以西门子 S7-200 SMART 为例，它本身原生支持的是 PPI 协议，但可以通过编程将部分通信口配置为 ModbusRTU 从站模式。实际上，很多国产 PLC、仪表、变频器出厂就支持 ModbusRTU 从站协议，你在设备手册里通常能看到“支持 Modbus RTU”字样。

2.3 寄存器模型的四个区域

Modbus 的核心抽象是“寄存器”。理解这四个区域，你基本就理解了整个 Modbus 数据模型：

现场调试时，90% 的精力都在处理保持寄存器和输入寄存器，因为大多数过程变量，如温度、湿度、压力、流量，都是 16 位数值。

2.4 数据地址与协议地址的偏移陷阱

这是最容易踩坑的地方，必须单独拿出来说。

设备手册上写的地址往往是“数据地址”，而 ModbusRTU 报文里实际传输的是“协议地址”，两者通常存在偏移：

• 保持寄存器：数据地址从 40001 开始，协议地址从 0 开始。所以手册上的 40001，对应协议地址 0。
• 输入寄存器：数据地址从 30001 开始，协议地址从 0 开始。手册上的 30001，对应协议地址 0。

如果你直接用 40001 当作寄存器编号发给设备，可能差 1 个地址位。好在大多数 Modbus 库允许你直接使用地址 0 开始索引，你只需要在处理设备手册时做好映射。

3. WebServer 在数据采集架构中的角色

光有 ModbusRTU 采集，只是解决了“把数据从设备拿出来”的问题。数据出来之后去哪、怎么展示、怎么被其他系统消费，这是第二个问题。

WebServer 在这个架构里的角色可以概括为三层：

3.1 数据接入层

WebServer 提供 HTTP API，让采集程序可以推送数据。采集程序每秒钟读取一次设备数据，然后以 POST 请求上报到 WebServer 的/api/telemetry接口。这样做的好处是采集与存储解耦，即使数据库暂时不可用，采集端可以先缓存数据。

3.2 数据查询层

WebServer 对外提供 REST API，比如GET /api/devices获取设备列表，GET /api/devices/1/telemetry?start=...&end=...查询历史数据。前端大屏、MES 系统、手机 App 都通过这个接口拿数据。

3.3 数据可视化层

虽然现代前端框架（Vue、React）都可以做展示，但如果你不想引入复杂的前端工程，直接用 Flask + 模板渲染或 FastAPI + Jinja2，几分钟就可以生成一个简单的实时监控页面。

这个架构里的关键判断是：不要把 Modbus 数据直接暴露给外部系统。曾见过有人把 ModbusTCP 端口直接映射到公网，任何能访问该端口的人都可以修改现场设备寄存器，这是非常危险的做法。通过 WebServer 做一层隔离，可以控制权限、校验数据、限制频率，是工业网络安全的底线。

4. 数据采集系统整体架构设计

在做具体编码之前，先用一张架构图描述整体设计思路。注意，这里不绘制 Mermaid 图，用文字描述数据流向。

系统整体分为四层：设备层、采集层、服务层、展示层。

设备层是现场传感器和 PLC，它们通过 RS485 总线连接到一台工业计算机或边缘网关。RS485 总线使用两线制差分信号，通信距离可达 1200 米，一条总线上可以挂接最多 32 个标准负载节点。

采集层运行一个守护进程，负责按固定周期扫描总线上每个从站地址，读取指定的寄存器区域，将原始数值转换为工程单位值，然后通过 HTTP 上报给服务层。

服务层是一个 WebServer，提供设备管理、数据接收、数据存储和数据查询功能。这个服务层可以跑在现场工控机上，也可以跑在工厂内网的服务器上。这里使用 SQLite 或 MySQL 作为存储后端。

展示层是浏览器的数据监控页面，定时轮询服务层的接口，将最新数据渲染成表格或曲线。这是 WebServer 最直观的价值：让工程师不用站在设备前面就能看到所有关键参数。

整体数据流如下：

现场设备(ModbusRTU从站) -> RS485 -> 采集程序(Modbus主站) -> HTTP JSON -> WebServer -> 数据库 -> 浏览器监控页面

从开发角度看，这个架构有一个明显特点：每一层之间的接口都比传统串口协议更通用。采集程序不关心 WebServer 的实现语言，WebServer 也不关心采集程序来自哪个厂商。只要约定好 JSON 字段，就能集成。

5. 环境准备与前置条件

为了把重心放在逻辑实现上，我选择 Python 作为示例语言，因为它有一流的 Modbus 库支持，开发效率高。以下环境以通用版本为例，具体版本以实际安装为准。

建议环境如下：

• 操作系统：Windows 10/11 或 Ubuntu 20.04/22.04
• Python：3.9 及以上
• Modbus 库：pymodbus3.x 版本
• Web 框架：Flask2.x 或FastAPI
• 串口调试工具：Modbus Poll（主站模拟）、Modbus Slave（从站模拟）
• 数据库：SQLite 内置即可，生产环境可换 MySQL/PostgreSQL

安装依赖：

pip install pymodbus flask requests

如果你的机器没有真实串口设备，也不要紧。可以安装虚拟串口软件（Windows 下使用 VSPD，Linux 下使用socat）创建一对互联的虚拟串口，再配合 Modbus Slave 模拟从站设备。

# Ubuntu 下用 socat 创建虚拟串口对 sudo apt install socat socat -d -d pty,raw,echo=0 pty,raw,echo=0

运行后会输出两个/dev/pts/XX设备，一个给从站模拟器用，一个给采集程序用。

6. 核心流程拆解：从轮询到上报

整个采集过程可以拆成三个子流程：设备发现、周期轮询、数据上报。

6.1 设备发现与配置管理

在一个 RS485 总线上，并非所有从站地址都是连续存在的。如果采集程序盲目扫描 1-247 所有地址，每条总线上都会产生大量超时等待，效率非常低。

更实际的做法是使用配置文件或数据库中的设备表来管理从站列表。每个设备需要知道的关键信息包括：

• 从站地址（Slave ID）
• 寄存器区域类型（保持寄存器/输入寄存器）
• 起始地址和读取长度（决定一次读取返回多少数据）
• 数据类型的解析规则（16位无符号/32位浮点/字节序）

示例如下：

{ "devices": [ { "slave_id": 1, "name": "1号温控仪", "register_type": "holding", "address": 0, "quantity": 10, "data_type": "float32", "byte_order": "ABCD", "scale": 0.1 }, { "slave_id": 2, "name": "S7-200SMART", "register_type": "holding", "address": 0, "quantity": 20, "data_type": "uint16", "byte_order": "AB", "scale": 1.0 } ] }

这个设计看似简单，实际上解决了项目中的大痛点：设备参数变了，不用改代码，只改配置。

6.2 周期轮询策略

采集周期是该设计的核心权衡。

• 温度、液位等慢变量，1-5 秒采集一次即可。
• 电机转速、压力控制等快变量，可以缩短到 500 毫秒。
• 不是越快越好，因为 RS485 是半双工通信，同一时间总线上只能有一个设备发送数据。采集频率过高会导致总线冲突和响应超时。

推荐实现方式是单线程串行轮询，每个从站之间加入短暂延时（比如 20-50ms），等待总线上数据稳定。不要轻易用多线程并发轮询同一总线，那会导致总线上的报文交织，除非你非常清楚自己的调度逻辑。

6.3 数据上报与本地缓存

采集到的数据要先在内存中组装成标准格式，然后批量上报。批量上报比一条一条上报效率高得多，示例如下：

{ "timestamp": "2025-01-15T10:30:00+08:00", "reports": [ {"device_id": 1, "values": {"temperature": 25.3, "humidity": 48.1}}, {"device_id": 2, "values": {"voltage": 220.5, "current": 5.2}} ] }

上报失败时，采集程序应把数据暂存在本地队列或磁盘文件中，等网络恢复后再补发。这避免了因为网络抖动丢失关键数据。

7. 完整示例代码实现

下面进入工程核心部分。这里的代码省略掉繁琐的异常分支，但保留了完整的主流程，读者可以直接复制修改。

7.1 ModbusRTU 采集端实现

创建modbus_collector.py，实现一个独立运行的数据采集脚本。

# -*- coding: utf-8 -*- """ 基于 ModbusRTU 的工业数据采集脚本 功能：周期读取多个从站寄存器数据，并通过 HTTP 上报到 WebServer """ import json import logging import time import threading from datetime import datetime import requests from pymodbus.client import ModbusSerialClient logging.basicConfig(level=logging.INFO, format='%(asctime)s [%(levelname)s] %(message)s') logger = logging.getLogger("collector") # 载入设备配置 def load_config(path="device_config.json"): with open(path, "r", encoding="utf-8") as f: return json.load(f) def parse_registers(register_data, data_type, byte_order="AB"): """ 将原始寄存器整型数组按指定格式转为工程值 这里主要处理 uint16 / float32 两种常见类型 """ if data_type == "uint16": return [reg for reg in register_data] elif data_type == "float32": # pymodbus 的 register data 每项是 16bit # 两个寄存器组成一个 32 位浮点数 from struct import unpack values = [] for i in range(0, len(register_data) - 1, 2): ab = register_data[i] cd = register_data[i + 1] if byte_order == "ABCD": payload = ab.to_bytes(2, "big") + cd.to_bytes(2, "big") elif byte_order == "CDAB": payload = cd.to_bytes(2, "big") + ab.to_bytes(2, "big") else: payload = ab.to_bytes(2, "big") + cd.to_bytes(2, "big") (val,) = unpack(">f", payload) values.append(round(val, 4)) return values return register_data class ModbusCollector: def __init__(self, config): self.config = config self.serial_conf = config["serial"] self.devices = config["devices"] self.web_url = config["webhook_url"] self.client = None def connect(self): self.client = ModbusSerialClient( port=self.serial_conf["port"], baudrate=self.serial_conf["baudrate"], bytesize=self.serial_conf["bytesize"], parity=self.serial_conf["parity"], stopbits=self.serial_conf["stopbits"], timeout=self.serial_conf["timeout"], ) if not self.client.connect(): raise ConnectionError(f"无法连接串口 {self.serial_conf['port']}") def read_device(self, device): slave_id = device["slave_id"] address = device["address"] quantity = device["quantity"] register_type = device.get("register_type", "holding") if register_type == "holding": result = self.client.read_holding_registers(address, quantity, slave=slave_id) else: result = self.client.read_input_registers(address, quantity, slave=slave_id) if result.isError(): logger.error("从站 %s 读取失败: %s", slave_id, result) return None raw = result.registers values = parse_registers( raw, device.get("data_type", "uint16"), device.get("byte_order", "AB") ) return values def poll_once(self): reports = [] for device in self.devices: try: values = self.read_device(device) if values is None: continue mapped = {} keys = device.get("value_keys", []) for idx, val in enumerate(values): if idx < len(keys): key = keys[idx] scale = device.get("scale", 1.0) mapped[key] = round(val * scale, 4) reports.append({ "device_id": device["slave_id"], "device_name": device["name"], "values": mapped }) except Exception as exc: logger.exception("从站 %s 读取异常: %s", device["slave_id"], exc) return reports def upload(self, reports): if not reports: return body = { "timestamp": datetime.now().isoformat(), "reports": reports } try: resp = requests.post(self.web_url, json=body, timeout=5) if resp.status_code != 200: logger.warning("数据上报失败，HTTP %s", resp.status_code) except Exception as exc: logger.warning("数据上报异常: %s", exc) def run_forever(self, interval=2.0): logger.info("采集程序启动，周期 %.1f 秒", interval) while True: start = time.time() reports = self.poll_once() self.upload(reports) elapsed = time.time() - start sleep_time = max(0.1, interval - elapsed) time.sleep(sleep_time) if __name__ == "__main__": cfg = load_config("device_config.json") collector = ModbusCollector(cfg) collector.connect() collector.run_forever(interval=2.0)

这段代码里有几个关键设计值得解释。

首先是parse_registers函数，它解决了 Modbus 数据解析中最容易踩的坑——数据类型转换。很多设备存储的不是标准的 16 位整数，而是两个寄存器拼成一个 32 位浮点数。字节序不同，解析结果完全不同。比如ABCD是标准大端序，CDAB则是字序颠倒。

然后是poll_once方法，它一次轮询所有设备，把每个设备的数据组装成独立字典，不混在一起。这样 WebServer 端可以根据device_id快速判断数据来源。

upload方法使用requests.post上报 JSON 数据，超时设为 5 秒。这里要注意，采集程序不能因为上报失败就停止采集，所以异常被捕获并打印日志，主循环继续运行。

最后，主程序中collector.connect()之后直接进入run_forever循环。如果串口连接失败，程序会抛出异常并退出，方便在 systemd 或 Windows 服务中实现自动重启。

7.2 设备配置文件

创建device_config.json：

{ "serial": { "port": "COM3", "baudrate": 9600, "bytesize": 8, "parity": "E", "stopbits": 1, "timeout": 2 }, "webhook_url": "http://127.0.0.1:5000/api/telemetry", "devices": [ { "slave_id": 1, "name": "1号温控仪", "register_type": "holding", "address": 0, "quantity": 2, "data_type": "float32", "byte_order": "ABCD", "scale": 0.1, "value_keys": ["temperature", "target_temp"] }, { "slave_id": 2, "name": "S7-200SMART模拟从站", "register_type": "holding", "address": 0, "quantity": 10, "data_type": "uint16", "byte_order": "AB", "scale": 1.0, "value_keys": ["v0", "v1", "v2", "v3", "v4", "v5", "v6", "v7", "v8", "v9"] } ] }

配置项中比较重要的是parity：这里用的是"E"偶校验。S7-200 SMART 的 ModbusRTU 从站通信参数默认是 9600, 8, 偶校验, 1 停止位。其他设备可能不同，必须逐台核对。

7.3 WebServer 接收端实现

创建app.py，实现数据接收、存储和查询接口。

# -*- coding: utf-8 -*- """ 基于 Flask 的工业数据 WebServer 功能：接收采集程序上报的数据，存储到 SQLite，并提供查询接口 """ import sqlite3 from datetime import datetime from flask import Flask, request, jsonify, render_template_string app = Flask(__name__) DB_PATH = "industry_data.db" def get_db(): conn = sqlite3.connect(DB_PATH) conn.row_factory = sqlite3.Row return conn def init_db(): conn = get_db() conn.execute(""" CREATE TABLE IF NOT EXISTS telemetry ( id INTEGER PRIMARY KEY AUTOINCREMENT, device_id INTEGER NOT NULL, device_name TEXT, timestamp TEXT NOT NULL, json_data TEXT NOT NULL ) """) conn.execute(""" CREATE TABLE IF NOT EXISTS latest_values ( device_id INTEGER PRIMARY KEY, device_name TEXT, timestamp TEXT, json_data TEXT ) """) conn.commit() conn.close() @app.route("/api/telemetry", methods=["POST"]) def receive_telemetry(): """接收采集程序上报的数据""" data = request.get_json() if not data: return jsonify({"error": "empty body"}), 400 ts = data.get("timestamp", datetime.now().isoformat()) reports = data.get("reports", []) conn = get_db() for report in reports: device_id = report.get("device_id") device_name = report.get("device_name", f"device_{device_id}") json_data = report.get("values", {}) conn.execute( "INSERT INTO telemetry (device_id, device_name, timestamp, json_data) VALUES (?,?,?,?)", (device_id, device_name, ts, str(json_data)) ) conn.execute(""" INSERT INTO latest_values (device_id, device_name, timestamp, json_data) VALUES (?,?,?,?) ON CONFLICT(device_id) DO UPDATE SET device_name=excluded.device_name, timestamp=excluded.timestamp, json_data=excluded.json_data """, (device_id, device_name, ts, str(json_data))) conn.commit() conn.close() return jsonify({"status": "ok", "received": len(reports)}), 200 @app.route("/api/devices", methods=["GET"]) def list_devices(): """获取设备列表及最新值""" conn = get_db() rows = conn.execute("SELECT * FROM latest_values").fetchall() conn.close() result = [] for row in rows: result.append({ "device_id": row["device_id"], "device_name": row["device_name"], "timestamp": row["timestamp"], "values": row["json_data"] }) return jsonify(result) @app.route("/api/devices/<int:device_id>/history", methods=["GET"]) def get_history(device_id): """获取设备历史数据""" limit = request.args.get("limit", default=100, type=int) conn = get_db() rows = conn.execute( "SELECT timestamp, json_data FROM telemetry WHERE device_id=? ORDER BY id DESC LIMIT ?", (device_id, limit) ).fetchall() conn.close() result = [] for row in reversed(rows): result.append({"timestamp": row["timestamp"], "values": row["json_data"]}) return jsonify(result) @app.route("/", methods=["GET"]) def index(): """简单监控页面""" html = """ <!DOCTYPE html> <html> <head><title>工业数据采集监控</title></head> <body> <h1>工业数据采集监控</h1> <div id="data">加载中...</div> <button onclick="loadData()">刷新</button> <script> async function loadData() { const resp = await fetch('/api/devices'); const items = await resp.json(); document.getElementById('data').innerHTML = items.map(item => `<pre>${JSON.stringify(item, null, 2)}</pre>`).join(''); } loadData(); setInterval(loadData, 5000); </script> </body> </html> """ return render_template_string(html) if __name__ == "__main__": init_db() app.run(host="0.0.0.0", port=5000, debug=False)

这个 WebServer 做了三件核心事情：

一是提供 POST/api/telemetry接口，这是采集程序的数据入口。数据落库时，同时写入telemetry历史表和latest_values最新值表。历史表记录每一次上报，最新值表只保留每个设备最后采集到的值，这样查询设备状态时不需要扫描历史表，效率高很多。

这里用了 SQLite 的ON CONFLICT(device_id) DO UPDATE语法，这是 SQLite 3.24 以上版本的特性，作用是重复上报时更新而不是插入新记录。

二是提供两个 GET 接口：/api/devices返回所有设备的当前最新值，/api/devices/<id>/history返回指定设备的历史数据。这两个接口将来要对接前端大屏或 MES 系统，语义简单，字段稳定。

三是提供一个极简的 HTML 监控页面。页面每 5 秒轮询一次最新值，可以看到数据实时刷新。如果你只想验证链路通不通，打开这个页面就够了，不需要任何前端工程。

7.4 运行采集服务

完成以上代码后，分别启动两个终端：

第一个终端启动 WebServer：

python app.py

第二个终端启动采集程序：

python modbus_collector.py

启动后观察终端日志。正常情况会看到类似输出：

2025-01-15 10:30:01 [INFO] 采集程序启动，周期 2.0 秒 2025-01-15 10:30:03 [INFO] 数据上报成功 2025-01-15 10:30:05 [INFO] 数据上报成功

然后在浏览器访问http://127.0.0.1:5000/，页面会显示设备的最新数据。

如果想验证 API 是否正常，可以使用 curl：

curl http://127.0.0.1:5000/api/devices

8. 运行结果与效果验证

数据链路是否打通，不能只看采集程序有没有输出日志。建议从三个维度验证。

8.1 验证数据入库

登录 SQLite 数据库，检查数据是否持续写入：

sqlite3 industry_data.db

SELECT device_id, device_name, timestamp, json_data FROM telemetry ORDER BY id DESC LIMIT 5;

如果查询结果持续有新记录出现，说明采集和上报链路正常。

8.2 验证最新值表

SELECT * FROM latest_values;

这张表应该每个设备只有一条记录，但 timestamp 字段会持续更新。

8.3 验证历史查询接口

curl "http://127.0.0.1:5000/api/devices/1/history?limit=10"

返回内容应该是按时间升序排列的 JSON 数组，每个元素包含 timestamp 和 values。

如果采集程序启动后没有任何数据，不要急着改代码，先按下面的排查顺序逐层确认。

9. 常见问题与排查思路

ModbusRTU 调试是典型的“分层排查”工作。下面整理高频问题，按从物理层到应用层的顺序排列：

这里特别强调数据跳变的问题。现场如果看到某个温度值在正常范围和一个巨大数值之间反复跳变，首先要怀疑的不是程序，而是总线物理层。排查时可以先用短网线直连设备测试，如果数据正常，再逐步排查线路问题。

10. 最佳实践与工程建议

10.1 配置与代码分离

上面例子中，串口参数、设备地址、寄存器映射全部放在 JSON 配置文件中。实际项目中，设备台账可能经常变，比如新增一台仪表、修改一个寄存器地址。如果这些信息硬编码在 Python 文件里，每次变更都要改代码、重启服务；放在配置文件里，甚至可以实现热加载，运维成本低很多。

10.2 数据采集与业务处理解耦

采集程序只负责“把数据读出来并上报”，不应该负责复杂的数据清洗、告警判断、趋势分析。这些逻辑应该放在 WebServer 或独立的数据处理服务中。这样做的原因是，如果采集程序业务逻辑过重，一旦业务变更，就可能影响采集稳定性，而采集是核心链路，不能随便宕。

10.3 增加数据质量标记

工业数据的价值在于可信。一个更完整的采集系统应该为每条数据增加质量标记，比如quality: "good"或quality: "estimated"。这样 WebServer 端可以识别异常数据，不会把传感器断线时的默认值当作真实值展示。

10.4 日志策略

采集程序的日志至少要覆盖以下信息：启动参数、每个从站的读取结果、上报成功与否、异常堆栈。日志文件要按日期切割，保留至少 30 天，便于远程排查。

建议日志格式包含时间、级别、从站地址和操作类型：

2025-01-15 10:30:01 [INFO] [slave=1] 读取成功，寄存器数量=2 2025-01-15 10:30:01 [ERROR] [slave=2] 读取失败，超时

10.5 安全边界

WebServer 如果部署在工厂内网，也要设置基本的访问控制。至少要做到：

• app.run不要开 debug 模式。
• 如果服务需要跨网段访问，使用反向代理（Nginx）并配置防火墙规则。
• 不要将 Modbus 串口映射到网络端口。如果需要远程写寄存器，应通过 WebServer 提供经过鉴权的写入接口，而不是直接暴露 ModbusTCP。

10.6 采集周期与线程模型

在 RS485 总线上，单线程轮询是更稳妥的选择。不要轻易引入多线程并发读写同一串口，因为 ModbusRTU 是半双工协议，并发会导致报文冲突。如果采集压力确实很大，建议拆分为多条独立串口总线，每个串口一个采集线程，互不干扰。

10.7 测试环境与生产环境

强烈建议在任何真实设备接入之前，先用模拟器打通全链路。使用 Modbus Slave 软件模拟几个从站，设置不同的寄存器值，然后用你的采集程序去读。这样可以将“设备问题”和“程序问题”分开，减少一次性的现场调试时间。

11. 总结与后续学习方向

现在你已经掌握了一套完整的 ModbusRTU 工业数据采集实现方案。从现场设备的寄存器读取，到 JSON 上报，再到 WebServer 的存储和查询，整个链路的关键点都已经梳理清楚。

这篇文章真正想传达的判断是：工业数据采集的难点，不是 Modbus 协议本身，而是对现场的了解程度和数据链路的整体把控。写一个 Modbus 读寄存器的小程序只需要几十行代码，但要让它在恶劣的工业环境下稳定运行，你需要掌握通信参数匹配、字节序解析、异常处理、上报策略、数据存储和展示这些完整环节。

接下来的进阶方向有三个：

第一，把 WebServer 换成性能更高的异步框架，比如 FastAPI，并使用 PostgreSQL 或时序数据库（如 InfluxDB、TDengine）存储历史数据。这会提升大数据量下的查询性能。

第二，考虑把采集程序做成 Windows 服务或 Linux systemd 服务，让它在开机时自动启动、崩溃时自动重启，这是产品化部署的基本要求。

第三，在 WebServer 中增加设备写入能力。Modbus 不只是读数据，还可以写寄存器控制设备。设计一个经过权限校验的写入接口，可以让你在网页上远程修改设备参数，但这一步务必谨慎，最好只开放给内网管理端。

建议你把这套代码跑通之后，用真实验证一下自己的理解：模拟一个 S7-200 SMART 从站，设置几个寄存器，然后尝试修改配置文件中不同的数据类型和字节序，观察采集结果的变化。只要亲手跑一遍，你就会对 ModbusRTU 的容错和数据模型有更实际的理解。

想把这篇文章里的思路用到实际项目中，建议收藏备用，照着从模拟环境开始，一步步搭出自己的采集服务。