第一次把产线上的PLC数据搬上网页大屏我用的是一台Windows工控机跑一个串口转发程序把读回来的寄存器值写进数据库再让前端每三秒拉一次。那套东西能跑但每增加一个新点位就要改一遍代码换一台PLC品牌又要重新写通讯函数后来项目里同时出现三台不同品牌的PLC我彻底下定决心做一个“PLC 转 Web API 服务器框架”。这个框架本身不复杂但它解决的是物联网项目里最磨人的问题协议碎片化、数据模型不一致、设备直接暴露太危险。这篇文章把整个框架的设计思路、核心代码、现场踩坑和落地复盘一次讲完写给正在被PLC协议折磨的物联网工程师、自动化工程师和准备做工业数据采集的开发者。1. 为什么不直接拿 PLC 当 Web 服务器用很多人一开始的想法是既然PLC能联网能不能直接在PLC里跑个Web服务让前端直接读这个思路我在早期项目里也试过最后都放弃了。不是说技术上完全不可能而是PLC的根本任务是实时控制让它承担Web API服务器的角色会带来一系列连锁问题。1.1 现场最常见的三种协议“方言”PLC行业的通信协议非常碎片化。西门子有S7协议三菱有MC协议施耐德有ModbusAB有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, interval1.0): super().__init__(daemonTrue) self.host host self.port port self.points points self.interval interval self.snapshot {} self.running True def run(self): client ModbusTcpClient(self.host, portself.port, timeout3) while self.running: try: if not client.is_socket_open(): client.connect() for point in self.points: rr client.read_holding_registers( addresspoint[address], countpoint[length], slavepoint.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 gunicorn4个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侧。它的自带的以太网通讯端口在默认配置下同时只能保持有限数量的活跃连接。我最初测试时用了一个调试工具和一个监视工具同时连接PLCPLC拿不到“操作许可”导致状态读取一直在切换。西门子的解决方法是设置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接入物联网真正的难点从来不是“把数据读出来”而是“怎么把数据稳定、安全、有组织地交出去”。想清楚这个原则再去做设计你会少走很多弯路。