第一次把产线上的PLC数据搬上网页大屏，我用的是一台Windows工控机，跑一个串口转发程序，把读回来的寄存器值写进数据库，再让前端每三秒拉一次。那套东西能跑，但每增加一个新点位就要改一遍代码，换一台PLC品牌又要重新写通讯函数，后来项目里同时出现三台不同品牌的PLC，我彻底下定决心做一个“PLC 转 Web API 服务器框架”。这个框架本身不复杂，但它解决的是物联网项目里最磨人的问题：协议碎片化、数据模型不一致、设备直接暴露太危险。这篇文章把整个框架的设计思路、核心代码、现场踩坑和落地复盘一次讲完，写给正在被PLC协议折磨的物联网工程师、自动化工程师和准备做工业数据采集的开发者。
1. 为什么不直接拿 PLC 当 Web 服务器用？
很多人一开始的想法是：既然PLC能联网，能不能直接在PLC里跑个Web服务，让前端直接读？这个思路我在早期项目里也试过，最后都放弃了。不是说技术上完全不可能，而是PLC的根本任务是实时控制，让它承担Web API服务器的角色，会带来一系列连锁问题。
1.1 现场最常见的三种协议“方言”
PLC行业的通信协议非常碎片化。西门子有S7协议，三菱有MC协议，施耐德有Modbus，AB有CIP，更别提各家还有串口、以太网、总线等一堆物理层变种。同样是读一个温度值，在Modbus TCP里是读保持寄存器，在三菱PLC里可能是读D区数据，在西门子S7-200 SMART里又要走另一套寻址逻辑。而物联网平台、可视化大屏、手机App这些东西，普遍只认HTTP、JSON、MQTT这类干净、标准化的协议。
这里面的差距才是物联网接入PLC时真正的“方言问题”。你想让一套IoT系统同时对接三台不同品牌的PLC，就得写三套不同的驱动，然后把数据统一成一套模型。如果每台PLC都直接暴露给上层，上层代码会被协议绑定得很死，换一台设备就得大改。
1.2 扫描周期和CPU开销经不起折腾
PLC的CPU资源是有明确的实时性要求的。一个扫描周期可能只有几毫秒到几十毫秒，CPU要在这么短的时间里完成梯形图逻辑、运动控制、模拟量处理。如果在PLC里挂一个功能完备的Web服务器，TCP连接管理、HTTP报文解析、JSON序列化这些任务都会抢占扫描周期，直接影响控制质量。
我在现场遇到过一个典型案例：有人试图在一台老款PLC上通过开放Modbus寄存器直接对接物联网平台，结果平台侧一启动高频轮询，PLC的通讯模块就开始频繁报错，连带着伺服轴出现了轻微抖动。后来把通讯任务全部挪到外置网关，PLC这边只用非常低的频率做数据交换，问题立刻消失。这件事给我最大的教训是：对于运行中的产线，任何可能干扰扫描周期的功能都必须拆出去。
1.3 安全边界需要一道“防火墙”
工业现场的安全逻辑和IT系统不太一样。PLC往往是整个产线的核心控制设备，一旦被人通过网络误操作，轻则停线，重则可能造成设备损坏甚至人员安全问题。如果把PLC直接暴露成Web API，意味着任何持有IP和端口的客户端都能尝试读写寄存器。
Web API服务器框架的另一个核心价值，是在PLC和外部网络之间形成一道逻辑隔离层。所有外部请求先经过API层的校验、鉴权和数据范围过滤，再由网关以受控方式访问PLC。PLC只需要信任这一个网关，不需要处理纷繁复杂的外部连接。这样无论是部署在局域网内还是跨网络访问，风险都能收敛到一个可控的范围内。
2. 框架架构：把“采集”和“开放”拆成两层
这个框架最常见的设计失误，是把采集和API服务写在一个回调函数里：来一个HTTP请求就去读一次PLC寄存器。这种写法在Demo里能跑，一上产线就会暴露出实时性差、PLC通讯模块过载、并发请求互相阻塞等问题。我的做法是把整个框架拆成采集层、缓存层和API层，三件事各干各的。
2.1 采集层：用轮询线程统一访问PLC
采集层是唯一直接和PLC打交道的模块。它负责维护与PLC的通讯连接，按照预设的周期读取点位数据，并把结果写入本地缓存。采集层的关键在于“主动轮询，而不是被动响应”。不管上层有多少客户端在请求数据，采集层只按自己的节奏和PLC通讯。
这样设计的理由很朴素：PLC的通讯模块通常支持有限数量的并发连接或事务。如果每个浏览器、每个后端服务都直接连PLC，连接数一多，PLC侧通讯资源马上紧张。而通过采集层做统一访问，PLC始终只面对一个稳定的客户端连接。
2.2 缓存层：让API响应不再依赖实时通讯
缓存层是连接采集层和API层的中间地带。采集线程读到的最新寄存器值、设备状态、最后更新时间、通讯错误信息，都存放在这里。API层在处理客户端请求时，只读缓存，不直接读PLC。
这是一个非常重要的设计取舍。API响应速度不再受PLC通讯延迟影响，客户端查询再频繁也不会额外增加PLC负担。更重要的是，当PLC因为故障重启或者网络断开时，API层依然可以返回最后一份已知数据，并附带“数据时间戳”，让上层系统明确知道当前数据是实时值还是历史快照。
我在这个缓存模型里还加了一个“数据新鲜度”标记。比如超过5秒没更新温度值，就把这条数据标记为陈旧，API返回时增加一个status字段。上层可视化系统看到陈旧的温度值，可以选择变成灰色或者弹出告警，而不是用一个假实时数据误导操作人员。
2.3 API层：把PLC点位映射成业务资源
API层的职责是面向业务。PLC侧的寄存器地址、数据类型这些东西，不应该被上层系统感知。API层要做的是把底层点位映射成有业务含义的资源。比如把“保持寄存器地址100的十六位整数”映射成“产线A当前产量”，把“线圈地址1”映射成“自动模式状态”。
这样一来，上层系统的开发人员只需要面向一套稳定的API契约，完全不需要了解PLC是什么品牌、用了什么协议、寄存器地址是多少。就算底层PLC替换成另一家品牌，只要API层提供的JSON结构不变，上层系统几乎不用改代码。
3. 核心实现：Modbus 转 REST API 的落地代码
下面就进入最实际的环节。我以最常见的Modbus TCP设备为例，展示一个最小可运行方案。完整的工业级框架还会涉及多协议适配、权限控制、数据历史存储等，但核心脉络就是以下这几段代码。注意：不同版本的pymodbus库API略有差异，大家以自己安装版本的官方文档为准。
3.1 选型与依赖
我选择Python实现这套框架，原因是物联网生态里Python的PLC通信库最齐全，而且后续对接MongoDB、InfluxDB、MQTT、可视化后端都非常方便。依赖只需要两个：
pip install pymodbus flask如果是生产环境，建议再加一个gunicorn来部署Flask服务，单纯用Flask自带的开发服务器扛不住真实流量。
3.2 采集线程：维护连接并写入本地快照
采集线程的核心逻辑是：启动时连接PLC，进入循环，按预设间隔读取点位寄存器，将结果更新到本地字典。我把点位配置抽象成一个列表，每个点位包含名称、寄存器地址、读取长度和数据类型。
import threading import time from pymodbus.client import ModbusTcpClient from pymodbus.exceptions import ModbusException class PlcPoller(threading.Thread): def __init__(self, host, port, points, interval=1.0): super().__init__(daemon=True) self.host = host self.port = port self.points = points self.interval = interval self.snapshot = {} self.running = True def run(self): client = ModbusTcpClient(self.host, port=self.port, timeout=3) while self.running: try: if not client.is_socket_open(): client.connect() for point in self.points: rr = client.read_holding_registers( address=point["address"], count=point["length"], slave=point.get("slave", 1) ) if not rr.isError(): self.snapshot[point["name"]] = rr.registers else: self.snapshot[point["name"]] = None self.snapshot["_update_time"] = time.time() except ModbusException as exc: self.snapshot["_last_error"] = str(exc) self.snapshot["_connected"] = False self.backoff_reconnect(client) time.sleep(self.interval)这段代码里最关键的是把snapshot作为唯一数据源，API层永远只读这个字典。你可以看到，即使Modbus通讯抛异常，采集线程也不会崩溃，而是把错误信息记录到快照里，同时进入重连逻辑。
3.3 API层：读取快照并处理写命令
API层我直接用Flask实现最小的几个端点。读数据只访问快照，不需要关心PLC通讯状态，所以响应速度非常快。
import time from flask import Flask, jsonify, request from poller import PlcPoller # 上面定义的采集线程 app = Flask(__name__) # 假设已经配置好点位并启动线程 poller = PlcPoller("192.168.1.10", 502, [ {"name": "temperature_a", "address": 100, "length": 2, "type": "float"}, {"name": "product_count", "address": 120, "length": 1, "type": "int"}, ]) poller.start() @app.get("/api/v1/points") def list_points(): snapshot = dict(poller.snapshot) for key in list(snapshot.keys()): if key.startswith("_"): continue snapshot[key] = { "value": snapshot[key], "update_time": poller.snapshot.get("_update_time"), "freshness": "fresh" if time.time() - poller.snapshot.get("_update_time", 0) < 5 else "stale" } return jsonify(snapshot) @app.post("/api/v1/write") def write_point(): data = request.get_json() point_name = data.get("name") value = data.get("value") # 白名单校验 allowed_write_points = {"target_speed": (0, 3000), "reset_counter": (0, 1)} if point_name not in allowed_write_points: return jsonify({"error": "point not writable or not found"}), 400 low, high = allowed_write_points[point_name] if not (low <= value <= high): return jsonify({"error": "value out of range"}), 400 address, length, dtype = WRITE_MAP[point_name] # 调用Modbus写寄存器 return jsonify({"ok": True})写命令的处理比读取复杂得多。我强烈建议在API层和PLC之间再加一道“命令确认”机制，尤其是涉及设备启停、调速这类关键操作。用户可能需要在网页上先发起一个命令请求，然后在30秒内输入确认码才能生效。这个机制听起来繁琐，但能有效防止误操作。
3.4 代码跑通后的自检清单
代码写完以后，不是启动就完事了。每次部署到现场前，我会按以下清单过一遍：
- 用Modbus Poll工具先离线验证PLC地址和寄存器长度是否正确；
- 检查
snapshot缓存里是否能正常填充数据； - 用一个临时的HTTP请求测试
/api/v1/points返回的数据是否与Modbus Poll一致； - 断开PLC网线，确认API依然返回数据但
freshness标记变为stale； - 重新插上网线，确认采集线程能在预设时间内恢复连接。
这套自检流程看起来简单，却帮我在现场排查掉大量“看起来代码没问题，但上位机就是没数据”的尴尬情况。
4. 现场稳定性：从“能跑”到“跑得稳”的关键细节
如果只是把几个PLC寄存器映射成JSON，框架半天就能搭出来。真正让这个框架从工控机里的试验品变成能稳定跑几个月的工业设备，靠的是下面这几个容易忽视的细节。
4.1 轮询间隔不是越快越好
很多工程师做数据采集时，默认把轮询间隔设成100ms甚至50ms，觉得数据越实时越好。但在PLC通讯这个场景里，这个直觉是有害的。
先看PLC侧：每一次Modbus读请求都要占用通讯模块的处理时间。通讯模块在高速处理读请求的同时，还要处理梯形图程序里的通讯指令、人机界面HMI的实时刷新，以及上位机其他软件的访问。如果外置网关用100ms的间隔疯狂读取，通讯模块的负载会急剧上升，严重时干扰PLC主程序的执行节奏。
再看网络侧：一个Modbus请求平均几十字节，响应也差不多。100ms轮询意味着每秒钟产生10个请求包，看起来不多，但现场如果有多台PLC、多台网关同时如此操作，再加上HMI和其他上位机软件的流量，交换机的负载和出错概率都会上升。
我的实践经验是：纯粹的数据采集与监控场景，1秒的轮询间隔已经足够；需要实时性较高的报警联动场景，可以缩短到500ms；只有像伺服位置跟踪这种强实时场景，才需要考虑更快的采集方案，但这通常不应该依赖TCP协议来完成。记住这句话：轮询间隔是一个需要根据现场需求理性设定的参数，不是越小越专业。
4.2 断线重连一定要用退避策略
PLC重启、网线松动、交换机暂时过载，这些问题在产线上几乎一定发生。重连策略如果设计得不好，比不断线还糟。
举个例子：某个网关程序检测到PLC网络断开，立即进入重连循环，每100ms尝试一次连接。当PLC重新上电时，通讯模块还没准备好，网关不断发起连接请求，PLC又不断拒绝。两边就这样互相耗着，典型的表现是网关日志里刷满连接失败，PLC侧通讯模块也异常繁忙，反而不容易恢复正常。
更合理的做法是采用指数退避：从1秒开始，连续失败则2秒、4秒、8秒逐步增加重连间隔，最多不超过30秒。一旦连接恢复，立即重置为1秒。这个策略的哲学是：PLC刚启动时需要的不是高频连接，而是给它一个缓冲时间，让通讯模块完成初始化。重连期间的采集线程也不要空转，而是把_connected标志置为False，同时保留最后一次成功的快照值。
4.3 日志必须分级、可轮转
我在早期版本里曾经把采集日志写成“无限追加的文本文件”，结果是跑了一周之后日志文件膨胀到几个GB，工控机磁盘报警。后来我改成了分级日志加文件轮转：INFO级别记录正常的轮询、点位更新和API访问摘要；WARNING级别记录单次Modbus通讯失败和重连尝试；ERROR级别才记录连接彻底断开和设备无响应。
日志文件按大小轮转，单个文件达到10MB就自动切成新文件，保留最近10个文件。这样既保证了问题可追溯，又不会让日志把磁盘空间耗尽。还有一个实用的技巧：在日志里为每次异常附加上下文编号，比如[PLC-03][UNIT-02] Connection lost，这样从日志文件里可以快速定位是哪台PLC哪个工位出了问题。
4.4 多客户端并发下的连接策略
当框架接入物联网平台后，API层会面对大量并发请求。这里最忌讳的是每个HTTP请求都单独建立一个到PLC的Modbus连接。正确的做法是保持API层与PLC完全解耦：API层只读缓存快照，所以它可以轻松应对几百甚至上千个并发请求，因为实际每个请求都只是内存字典的一次读取。
我在一个项目里测过：框架部署在一台普通的四核工控机上，采用Flask + gunicorn（4个worker），同时对上提供REST API，对内只维持一条Modbus TCP连接，采集间隔设为1秒。在200个Web客户端并发轮询的场景下，API平均响应时间稳定在5毫秒以内，而PLC侧通讯模块几乎无感。这个结果印证了一个结论：并发能力强不强，取决于是否把“对PLC的访问”降到了最低频率，而不是取决于你的Web服务器性能。
5. 复盘：一个三台PLC实时数据项目的落地全过程
理论讲再多，不如一个完整的落地案例有说服力。这个项目是我去年做的一条小型装配线数据采集改造，规模不大，但非常能说明问题。
5.1 需求拆解
现场有三台设备，分别是：
- 一台西门子S7-200 SMART，控制装配线的传送带逻辑；
- 一台老款三菱FX系列PLC，控制气缸夹具；
- 一台支持Modbus TCP的温控器，控制加热工位。
客户要求：产线上所有设备的产量计数、设备状态、关键温度、报警信号，十分钟内必须出现在车间的大屏看板上；同时，现场管理人员需要通过平板电脑实时查看数据，不需要进控制柜看触摸屏。
这个需求听起来不复杂，但第一版的方案设计曾经尝试让大屏系统直接分别读三台设备。西门子走S7协议、三菱走MC协议、温控器走Modbus，每套协议都单独实现一遍，光是驱动调试就花了两周，而且只要一台设备通讯出问题，大屏对应区域就空白。
5.2 框架侧的选择与部署
最后采用的方案就是我在这篇文章里描述的框架结构。采集层里分别实现了三个驱动：西门子用snap7库，三菱用pymcprotocol库，温控器用pymodbus库。三套驱动各自独立运行，读取的点位统一映射成一套内部点位模型，写入缓存。API层只暴露一套统一的REST接口，大屏和后端只认这套接口。
在部署位置上，用户原计划把网关程序放在云端服务器，通过公网直接访问现场PLC。我没有采用这个方案，原因是公网链路一旦抖动，采集线程会频繁重连，而且PLC的通讯模块将直接暴露在外网环境下，风险太高。最终我们在现场放置了一台小型的边缘计算盒子，靠近PLC侧，通过局域网连接所有设备。采集层的PLC通讯全部在局域网内完成，API层则通过现场已有的工业网关与上层办公网打通。
5.3 排掉的两个坑
第一个坑是寄存器地址错位。三菱FX系列PLC的程序里，D100到D110原本是产量数据的连续寄存器。我用pymcprotocol读取时，发现返回的数据和服务端梯形图里的变量顺序完全对不上。查到最后才发现，三菱PLC的数据寄存器在程序里被重新映射过，D105对应的实际物理地址和梯形图工程里显示的窗口地址并不一致。这在老旧设备上非常常见，因为程序几经修改，早期的注释早就失效了。
第二个坑出现在西门子S7-200 SMART侧。它的自带的以太网通讯端口在默认配置下，同时只能保持有限数量的活跃连接。我最初测试时用了一个调试工具和一个监视工具同时连接PLC，PLC拿不到“操作许可”，导致状态读取一直在切换。西门子的解决方法是设置PG/PC的“唯一客户端”属性，现场实施的工程师不一定都会配。最终我在采集驱动里做了互斥逻辑，确保同时只有一条S7连接处于活跃状态，其他调试工具一律在需要时临时占用。
5.4 上线后的实际表现
框架连续运行了三个月，没有出现一次需要人工干预的重启。车间大屏的数据延迟控制在3秒以内，温控器数据1秒刷新一次，产量数据在交接班时清零操作通过API写命令完成。客户满意度较高，但真正让我觉得这套框架值回票价的是后来的扩展：第二期项目要增加两台同样型号的温控器，我只需要在配置里增加两个新的点位，重新导入，前端的看板配置一下，整个接入过程不到半小时。
6. 最后，写给准备自己做这套框架的人
前面几节已经把框架的结构、代码和经验都讲完了。最后分享几个不在任何代码注释里、但每一个都真金白银换来的建议。
6.1 点位表内容管理比代码本身更重要
代码写得再漂亮，点位表维护混乱一样会让项目崩溃。PLC程序升级之后，变量地址变了；设备改造后，某个寄存器从只读变成了可写；车间新增了一台设备，点位表里却没有模板可以用。这些问题远比通讯函数bug频繁。
我的做法是在项目中专门准备一份点位表文档，每行记录：点位名称、所属设备、寄存器地址、数据类型、读写属性、单位、更新频率、备注。每次PLC程序发生变化，第一件事是更新点位表，再改框架的配置。这个顺序不能颠倒，否则调试的时候你根本不知道是通讯出错了，还是地址对不上。
6.2 写命令必须设计得比读取“重”
数据读取是只读行为，出错最多就是显示错误；写命令是修改行为，出错可能直接影响设备动作。所以我把写命令设计得很“重”：客户端提交写请求后，接口并不会立刻执行PLC写入，而是返回一个待确认的token，要求客户端在30秒内携带这个token发送确认请求，框架才会真正下发Modbus写指令。
这个机制还会自动记录所有写操作的日志：谁在什么时间提交了什么命令，确认token是什么，最终执行结果如何。一旦现场发生问题，这套审计日志的价值无可替代。
6.3 下一步可以扩展的方向
这个框架现在已经能满足基本的工业数据采集和监控需求。如果你需要继续扩展，我建议按这个优先级考虑：
- 增加MQTT发布能力，把点位快照周期性地推送到物联网平台的MQTT Broker；
- 增加历史数据存储，用轻量级的时序数据库或者直接落到关系库里，为后续数据分析做准备；
- 增加WebSocket或SSE通道，让大屏在这类场景可以做到“秒级实时推送”而不需要HTTP轮询；
- 为每个点位增加限值配置和告警规则，让网关具备简单的边缘计算能力。
我个人在这套框架上线之后最大的体会是：PLC接入物联网，真正的难点从来不是“把数据读出来”，而是“怎么把数据稳定、安全、有组织地交出去”。想清楚这个原则，再去做设计，你会少走很多弯路。
