PLC转Web API框架：工业物联网数据采集的轻量级中间件实践

做工业物联网的人，迟早会遇到这个场景：车间里的PLC跑了几年了，数据都在寄存器里躺着，可MES要产量，看板要状态，云端平台要温度、压力、电流，结果全卡在“怎么把PLC里的数拿出来”这一步。老办法是拉一堆现场总线、组态SCADA，花钱花时间，还往往被供应商绑死。这些年我折腾了一套“PLC转Web API服务器框架”，把PLC的读写能力直接封装成标准HTTP接口，物联网应用像调自己的后端服务一样去读设备、写设备。这篇文章就把这套框架从设计到落地的完整思路、核心代码和踩过的坑一次性讲清楚。

适用人群很明确：正在做物联网毕业设计的学生、工厂里接MES系统的工程师、想快速把老PLC设备接入云平台的集成商。你不需要很懂PLC，只要会点Python或者Node.js，照着这个思路就能在几天内搭出可用的中间件。反过来说，如果你本身就是电气工程师，这套框架能帮你把PLC的数据开放给IT团队，自己不用碰Web开发。

1. 为什么非要加一层“PLC转Web API”中间件

1.1 PLC在物联网场景里的先天短板

PLC最擅长的是逻辑控制和实时响应，但它骨子里是个“封闭”的工业设备。它说话用的是Modbus、S7、FINS、CC-Link这类工业协议，跑在串口或工业以太网上，数据结构完全跟着工艺走，外界根本看不懂。你让物联网平台直接跟PLC对话？大多数平台只会HTTP、MQTT、JSON，Modbus得专门做解析，更别说西门子的S7协议、三菱的MC协议，光是字节序就够折腾一星期。

这里有个核心矛盾：物联网应用需要的是“语义化、标准化、随时可调用”的数据接口，而PLC只提供“位、字节、字、寄存器”这种底层视图。中间这层翻译和封装，就是“PLC转Web API”框架存在的意义。

1.2 Web API能给物联网带来什么

把PLC包装成Web API之后，价值立刻就能落地。第一，任何能发HTTP请求的终端都能接入：手机App、Web看板、Python脚本、Node-RED、云函数，甚至Excel里的Power Query都能直接读PLC数据。第二，API层天然适合做权限控制、日志审计、数据格式转换，这些逻辑放在PLC里很难写，放在中间件里轻松搞定。第三，团队协作变简单了——IT工程师不用懂PLC编程，拿一份点位表就能像开发普通后端一样调接口；电气工程师也不用学Web开发，只管维护PLC程序和数据字典。

很多人在这一步会纠结：直接用OPC UA不香吗？OPC UA确实是工业通讯的标准答案，但很多老PLC不支持，商业OPC UA服务器又往往要授权费，配置门槛也不低。对于中小规模场景，用轻量级中间件走Modbus TCP桥接，反而能在一天内上线，开发成本直接降一个数量级。这就是我选择做这套框架的原因。

2. 框架选型与整体架构设计

2.1 通讯层：优先Modbus TCP，个别情况走厂商协议

底层通讯协议决定了框架的稳定性和开发量。我的建议是：只要PLC支持Modbus TCP，就优先用Modbus TCP。理由很实际：Modbus TCP是事实上的工业以太网标准，几乎所有主流PLC都支持，比如台达、汇川、信捷、永宏、施耐德，西门子部分系列也能通过网关或选件支持。它的报文结构简单、调试工具多，Python的pymodbus、Node.js的modbus-serial都能快速跑通。

如果碰到西门子S7-1200/1500这种原生只开放S7协议、Modbus需要额外授权的情况，就走S7协议。Python有python-snap7，Node.js有nodes7，GitHub上都是现成的。三菱FX/Q系列走MC协议，也有对应库。原则是：先上Modbus，再谈私有协议，别让协议选型卡住整体进度。

2.2 API层：Python FastAPI是我现在的首选

API框架的选择，我重点对比过三组方案，直接上表格：

我常年用FastAPI，原因有三个：第一，Pydantic可以自动做请求和响应的强类型校验，省掉一堆手写的参数判断；第二，asyncio可以同时管理多个PLC的轮询任务，不会因为一台设备断网就卡死整个服务；第三，FastAPI自动生成Swagger文档，把抽象好的点位接口直接扔给前端或物联网平台对接方，沟通成本几乎为零。

2.3 核心架构：采集、缓存、API三层分离

整个框架拆成三个逻辑层：

• 采集层：负责和PLC通讯，执行Modbus/S7协议读写，屏蔽底层协议差异。
• 缓存层：维护统一的数据字典（点位名、寄存器地址、类型、缩放系数、状态），保存最近一次读取的快照值，同时记录点位的时间戳和健康状态。
• API层：对外暴露RESTful接口，读点位、写点位、批量读、订阅变更，全部走HTTP/JSON。

这三层分离有什么好处？采集层坏了不牵连API层，API层压力再大也不会直接冲击PLC扫描周期；缓存层让API响应不再受制于PLC的响应速度——读到的永远是最近一轮轮询的快照，而不是现场阻塞式请求。尤其当物联网平台每秒要刷新几十个点位时，直接打PLC会明显增加通讯负载，但打内存缓存就非常轻松。

3. 核心实现：从点位表到可调用API的四个步骤

3.1 画出PLC侧的地址映射表

这是最需要耐心的一步，也是后面所有开发的基础。要把所有需要暴露给物联网的点位整理成一张表，至少包含：点位名称（英文ID）、中文描述、数据类型（bool/int16/uint16/float32/string）、PLC寄存器地址（如400001）、数据寄存器类型（线圈/离散输入/保持寄存器/输入寄存器）、读写权限（RO/RW）、缩放系数（比如温度实际值=原始值×0.1）。这张表建议直接从PLC程序里导出，千万别凭记忆手敲。

比如一个恒温车间项目，我当时的点位表大致是：

这套表既是PLC程序注释的补充，也是API文档的源头。后面写配置文件和Swagger描述，都从这里来。

3.2 实现Modbus TCP采集模块

我用Python的FastAPI+pymodbus做了第一版，核心采集模块就一百多行。先封装一个ModbusClient管理器，负责建立连接、读数据、断线重连。这里有一个极其容易踩的坑：pymodbus 3.x和2.x的API变化很大，网上很多代码是2.x的写法，复制下来直接跑不通。我建议直接用pymodbus 3.x的异步客户端：

python复制from pymodbus.client import AsyncModbusTcpClient

class PlcConnector:
    def __init__(self, host: str, port: int = 502, timeout: float = 3.0):
        self.client = AsyncModbusTcpClient(host, port=port, timeout=timeout)
        self.connected = False

    async def ensure_connected(self):
        if not self.connected:
            self.connected = await self.client.connect()
        return self.connected

    async def read_holding_register(self, address: int, count: int = 1):
        if not await self.ensure_connected():
            raise ConnectionError("PLC连接失败")
        result = await self.client.read_holding_registers(address, count)
        if result.isError():
            raise RuntimeError(f"读取寄存器失败: {result}")
        return result.registers

这段代码加了ensure_connected，每次读写前检查连接状态。关键点在于：千万不能每次请求都新建一个TCP连接，必须复用连接，否则PLC的502端口会被大量TIME_WAIT连接塞爆，现场直接卡死。我处理过好几次这类故障，最后发现都是代码里每次读都new了一个client，连接数瞬间上千。

3.3 设计RESTful接口：读、写、批量、状态四大类

API设计要克制，不要一上来就搞几十个端点。我通常只保留四个核心路由：

• GET /api/plcs/{plc_id}/points/{point_id}：读取单个点位。
• GET /api/plcs/{plc_id}/points/batch?ids=a,b,c：批量读取点位，用于看板和物联网平台定时拉取。
• POST /api/plcs/{plc_id}/points/{point_id}/write：写入控制指令，body里带value。
• GET /api/plcs/{plc_id}/health：返回PLC连接状态、最后通讯时间、错误次数。

响应格式统一用JSON，带code、message、data三个字段。data里除了value，还必须带timestamp和quality。quality字段特别重要，0表示数据新鲜，1表示数据过期，2表示点位未配置。物联网平台拿到quality后就不需要自己瞎猜数据是否有效。FastAPI里可以直接这样定义：

python复制class PointData(BaseModel):
    value: float | bool | str
    timestamp: datetime
    quality: int = 0

@app.get("/api/plcs/{plc_id}/points/{point_id}", response_model=PointData)
async def read_point(plc_id: str, point_id: str):
    return api_service.read_point(plc_id, point_id)

写入接口是另一个要小心的地方：控制器写入是有风险的，不支持随意写。我的做法是，配置表里把每个点位的权限标清楚，API层校验RW；写入前还会做数值钳位，比如温度设定值只允许在0~100之间，防止上位机误操作把参数写飞。这一步很多教程不会提，但在工厂里真出了问题，责任边界就靠这个钳位逻辑来兜底。

3.4 缓存与推送：让数据“等”API，而不是API“等”数据

直接把API请求转发给PLC的做法，在点数少、频率低的demo里没问题，但一旦接入物联网大屏，每秒几十次请求，PLC会吃不消。我的做法是引入一个后台轮询任务，每500ms把点位表里所有RO点扫一遍，写入内存字典。API读取时只取缓存，写入时则先更新缓存，再异步下发PLC。

如果物联网平台需要“变化上报”而不是定时拉取，我会在框架里加一个WebSocket通道：客户端订阅点位ID后，后台轮询发现数值变化超过死区（deadband），就主动推送。死区设置很有讲究：温度类可以设0.1，开关量只要有变化就推，电流可以设0.5A。死区设小了，网络流量爆炸；设大了，事件实时性下降。

4. 部署中必须处理的三个关键细节

4.1 轮询周期怎么定：一千个人有一千种设备

轮询周期没有统一定论，但可以给一个参考值：常规工况用500ms轮询；需要做高速曲线显示的用200ms；用Web API做控制回路的，千万别用轮询，直接走专用的长连接或OPC UA。为什么？API框架本身有事件循环和序列化开销，轮询越密，CPU占用越高，而且PLC通讯本身也有吞吐上限。

我遇到过一个实际案例：某车间要求所有温度、压力、流量每秒采集一次，60个点，我一开始把轮询设成100ms，结果模块CPU占用跑到80%，网络包满天飞，PLC端都开始丢响应。后来改成500ms+补偿曲线，效果反而更好。所以合理做法是：先把采集任务按PLC分组，每个PLC一个独立task，task内部按时延要求调度不同优先级的点位。高频点位单独一条快路，低频点位走慢路。

4.2 多线程并发读写：一个容易爆掉的坑

PLC TCP连接不是线程安全的，尤其Modbus协议，同一时刻只能有一个请求在线上跑。如果你用FastAPI的async直接并发去读写同一个PLC，光请求重试就会把通讯链路打乱。我的经验是：每个PLC实例维护一个asyncio.Lock，读写前先加锁。虽然会略微拉低响应，但能保证通讯稳定。

python复制class PlcGateway:
    def __init__(self, connector):
        self.connector = connector
        self.lock = asyncio.Lock()

    async def read_point(self, addr):
        async with self.lock:
            return await self.connector.read_holding_register(addr)

还有一个细节：PLC的写操作比较复杂，一般流程是先写预设值、再触发一个“执行写”的位。这类多步操作必须包在同一个锁里，否则中间插入了别的写请求，状态就乱了。我建议把这类写操作封装成“事务”，锁里完成所有步骤。

4.3 断线重连和数据补传：工业现场永远不能信一次连接

PLC在车间里重启、断电、网络瞬断都是常态，框架必须把这当成正常路径而不是异常路径。断线重连的策略我总结成三条：第一，TCP连接断开后，指数退避重连，初始1秒、最大30秒，不要疯狂重试；第二，重连成功后，立刻执行一次全量点读，把缓存刷新；第三，断线期间的WebSocket订阅用户，要收到离线通知，UI上显示设备离线，而不是把旧数据一直当新鲜的展示。

数据补传这件事很多人会忽略。我的做法是：缓存里保存每个点位的last_success_time和last_value，如果轮询某一点失败，不覆盖旧值，而是把quality字段置为1。API层给到物联网平台时，平台就能根据quality决定是做插值还是置报警。数据补全的活儿，交给时序数据库或边缘规则引擎，框架本身不需要做大量历史存储。

5. 真实项目中踩过的坑与排查心得

5.1 Modbus地址偏一位，读出来全是离谱值

Modbus线圈和保持寄存器的地址，有0基和1基两种写法。PLC组态软件里可能显示“400101”，但报文里实际地址是0x0064。很多刚开始用Modbus库的人直接把100传进去，差一位，读出来的值完全是乱的，或者是相邻的寄存器。排查方法很简单：先读一个已知固定值的寄存器，对比一下，确认偏移量。这个坑几乎每个做Modbus的人都遇到过，不丢人，但很浪费时间。

5.2 功能码选错，报错毫无头绪

读线圈用功能码01，读离散输入用02，读保持寄存器用03，读输入寄存器用04。我见过有人用03去读开关状态，结果返回的都是0或65535。书写代码前务必确认当前点位表的寄存器类型，别再一股脑“保持寄存器”打天下。

5.3 32位浮点的高低字节顺序搞反

很多PLC的32位float存储方式和Python默认解析方式不一致，有AB（大端）和BA（小端）两种。同一条数据，字节序错了读出来就是天文数字。解决办法是：在配置表里给每个点位加一个byteorder字段，解析时按字段转换。调试时先用PLC软件写一个已知数（如3.14），再读回来对比，几秒钟就能确定字节序。

5.4 端口502被占用：设备IP冲突比协议还难搞

现场调试时，PLC和电脑直连或者走交换机，IP配错、端口被占、防火墙拦截都会导致连接超时。我遇到过最经典的是现场有两台PLC用了相同的IP，框架里配置的IP指向了错误的设备，数据一切正常但就是读的不是目标设备。排查技巧：连接前先ping PLC的IP，再用Modbus Poll或者python脚本连接读一次设备ID，确认目标身份无误再开始批量开发。

5.5 API超时和504：缓存救了整个系统

曾有同事把API层改成直接实时读PLC，一上线，前端的看板每隔几秒就504。后来把缓存加回来，API响应时间稳定在5ms以内。这里面有个原则：凡是“读”接口，一律从缓存走；凡是“写”接口，也先用缓存语义做校验再异步下发。别让PLC成为API的瓶颈，这条经验值价值十万。

6. 框架进阶：让PLC真正融入物联网生态

6.1 点位管理：从硬编码到配置化

框架跑起来之后，最痛苦的不是开发，而是加点位。如果一个一个在代码里写死，随着设备增加会崩溃。我后来把点位表挪到了YAML/JSON配置文件里，或者直接存数据库，框架启动时加载。每次PLC程序升级，只需更新配置文件或DB里的点位表，API和新点位立即生效，不用改一行代码。这里的关键是配置项的规范化，至少包括plc_id、register_type、address、data_type、scale、offset、deadband、permission。

6.2 对接MQTT、时序库和云平台

Web API只是中间层，真正的物联网应用往往还需要把数据推送到MQTT broker、写入时序数据库（如InfluxDB/TDengine）、上传到云平台。我的框架在API层之外加了一个sink模块，专门对接这些下游。sink模块订阅内部的缓存变更事件，按规则转发。这样，同一个点位数据可以同时去大屏、去历史库、去报警规则引擎，而PLC侧只承担一次轮询。

6.3 安全防护：别裸奔

把PLC暴露成Web API后，安全就绝对不能忽略。第一，加上API Key或JWT认证，不要让任何人拿到IP就能写寄存器；第二，写操作必须做白名单，不允许客户端传任意地址和任意值；第三，前置Nginx做TLS终止和限流。曾经有人觉得工厂内网无所谓，结果勒索病毒一进来，整个产线瘫痪。设备层不该裸奔，API层更不该。

7. 最后的实操建议

这套“PLC转Web API服务器框架”我在三个项目里实际落地过，从一条产线扩展到全车间，从单台PLC扩展到几十台同时接入。最大的体会是：不要追求一步到位的“万能网关”，先把PLC的读写能力以API的方式稳定输出，比什么都重要。你后续想接MES、接云平台、接可视化看板，只要API在，都是几天的工作量；反过来，API设计得花里胡哨，采集层却在现场频频断线，用户对你框架的评价就两个字——不好使。

最后分享一个小技巧：调试时不要一上来就连真实PLC。先用一个Modbus仿真从站（Modbus Slave工具，或者干脆在树莓派上用pymodbus起一个模拟Server），把点位表和数据都模拟出来，信号联调完，再到现场切换真实PLC地址。你会发现，80%的问题其实跟现场设备没关系，都是自己在协议和地址上翻了车。先仿真，再上线，永远比在车间边改边试要快得多。