Python 采集三菱 PLC 数据并可靠写入数据库的时序设计

1. 项目背景与整体设计思路1.1 为什么会有这个需求在工厂自动化场景里三菱 PLC 的占有率非常高尤其是 FX 系列和 Q 系列产线上跑个几年甚至十几年都不稀奇。设备在跑数据在产生但很多现场的数据是用完即弃的——HMI 上显示一下操作工看一眼过去了就没了。等到品质部门要追溯某个批次的工艺参数或者设备部门想分析某台机器的稼动率才发现历史数据根本没存下来。我接触过好几个这样的项目甲方一开始的需求都很朴素能不能把 PLC 里的几个寄存器值定时读出来存到数据库里听起来简单但真做起来坑比想象的多。用 Python 通过 MC 协议Mitsubishi Communication Protocol三菱的开放式通信协议跟 PLC 通信本身不算难难的是时序设计——什么时候读、读多快、读失败了怎么办、数据怎么保证不丢不重、数据库写入会不会拖慢采集节奏。这个项目要解决的核心问题就是用 Python 稳定地从三菱 PLC 采集数据并可靠地写入数据库整个链路的时序要经得起产线 7×24 小时运行的考验。适合谁来参考如果你是会一点 Python、懂一点 PLC、正在做设备数据采集或者 MES 对接的工程师这篇内容应该能帮你少走弯路。如果你是完全的新手也没关系我会把关键概念用大白话讲清楚。1.2 整体架构怎么搭先说结论我推荐的架构是**采集层 缓冲层 入库层三层分离**而不是读一个写一个的直连模式。为什么因为直连模式有个致命问题数据库一旦卡顿比如索引重建、大查询、网络抖动采集线程就会被阻塞PLC 那边的读取节奏就乱了。如果 PLC 侧有看门狗或者通信超时机制甚至可能触发报警。三层分离之后采集只管采集数据先扔到内存队列或者本地缓冲入库层慢慢消费两边互不干扰。具体来说采集层Python 进程通过 MC 协议通常走以太网FX3U 加 ENET 模块或者 Q 系列自带网口周期性读取指定软元件D 寄存器、M 继电器、X/Y 等。缓冲层用queue.Queue做内存队列或者用 SQLite 做本地落盘缓冲防止进程崩溃丢数据。入库层独立的消费者线程或进程从队列取数据批量写入 MySQL / PostgreSQL / SQL Server / InfluxDB 等目标库。这个架构的好处是解耦。采集频率可以很高比如 100ms入库可以批量攒着写比如 1 秒一批两边各自优化。1.3 通信方式的选择逻辑三菱 PLC 的通信方式有好几种串口RS-232/RS-485、以太网MC 协议、CC-Link 等。为什么选 MC 协议走以太网串口方式比如 FX2N 走 485速率低、距离短、一台电脑能接的设备数量有限而且 Python 操作串口还要处理各种超时和帧格式调试起来烦。以太网 MC 协议就不一样了速率快、支持多设备、Python 有现成的库比如pymcprotocol、mcprotocol开发效率高很多。注意FX3U 本身没有网口需要加 FX3U-ENET-ADP 或 FX3U-ENET-L 模块。Q 系列和 iQ-R 系列一般自带以太网口直接支持 MC 协议。选型的时候要确认清楚别买回来发现接不上。MC 协议又分两种帧格式3E 帧和4E 帧。3E 帧用得最广兼容性好4E 帧支持更大的数据量和更灵活的寻址。一般项目用 3E 帧就够了我下面也以 3E 帧为主来讲。2. 核心细节解析与实操要点2.1 MC 协议 3E 帧的报文结构要写好通信代码得先搞明白 MC 协议 3E 帧长什么样。很多人直接用库不关心底层结果一出问题就抓瞎。我建议至少把请求帧的结构过一遍。一个典型的 3E 帧请求报文二进制格式大致是这样字段长度字节说明副头部2固定 0x5000网络号1通常 0x00PLC 号1通常 0xFF站内请求目标模块 IO20x03FF请求目标模块站号10x00请求数据长度2后续数据的字节数监视定时器2单位 250ms比如 0x0010 4 秒指令2如 0x0401批量读、0x1401批量写子指令2如 0x0000按字、0x0001按位首软元件号3起始地址低字节在前软元件代码1如 0xA8 D 寄存器0x90 M 继电器软元件点数2要读多少个响应帧会在指令后面加一个结束代码2 字节0x0000 表示成功其他值就是错误码。为什么要懂这个因为当你用库读不到数据的时候抓包一看可能是软元件代码写错了或者点数超了限制。比如 D 寄存器一次最多读 960 个字3E 帧二进制你写 1000 就会报错。这些细节库的文档不一定写清楚。2.2 Python 库的选型对比Python 操作三菱 MC 协议常用的库有这么几个库名优点缺点适用场景pymcprotocol纯 Python支持 3E/4EAPI 简洁性能一般高频率下 CPU 占用偏高中小型项目采集频率 10Hzmcprotocol功能类似文档稍全社区活跃度一般一般采集python-mcprotocol轻量功能较少简单读写自己 socket 实现完全可控性能最好开发量大要处理各种边界高频采集、特殊需求我的经验是采集频率在 5Hz 以下直接用 pymcprotocol 就行省事。如果频率要求到 10Hz 以上或者要同时采集几十台设备建议自己用 socket 封装把连接复用、批量读取、异常重连都控制在自己手里。pymcprotocol 的基本用法大概是这样from pymcprotocol import Type3E plc Type3E() plc.connect(192.168.1.10, 5000) # PLC 的 IP 和端口 d_values plc.batchread_wordunits(headdeviceD100, readsize10) plc.close()看着简单但生产环境不能这么写。每次读都 connect/close开销大不说网络一抖就抛异常。正确做法是长连接 异常重连 心跳保活。2.3 时序设计的核心参数时序设计说白了就是回答几个问题多久读一次一次读多少读失败等多久重试数据攒多久写一次库采集周期取决于工艺要求。如果是监控温度、压力这种慢变量1 秒一次足够如果是抓取瞬间的触发信号可能要 100ms 甚至更快。但要注意MC 协议单次请求的往返时间RTT在局域网内大概 5~20ms所以理论上限也就 50Hz 左右再快就不现实了。单次读取点数D 寄存器一次最多 960 字M 继电器一次最多 7168 位。实际项目中我一般把需要连续读取的地址合并成一块减少请求次数。比如要读 D100~D120 和 D200~D210与其发两次请求不如一次读 D100~D210110 个字虽然多读了一些无用数据但省了一次网络往返整体更快。重试策略读失败不要立刻重试容易雪崩。我一般用指数退避第一次失败等 100ms第二次等 200ms第三次等 400ms最多重试 3 次还不行就标记连接断开触发重连流程。入库批量单条 insert 效率极低MySQL 每秒几百条就到顶了。用批量插入executemany或者拼多值 INSERT一次 500~1000 条性能能提升几十倍。所以入库层要攒数据攒够一批或者等够时间比如 1 秒就写一次。2.4 数据模型的设计数据库表怎么设计直接影响查询效率和存储成本。我见过有人把每个寄存器单独建一列结果 PLC 程序一改地址表结构就得跟着改维护起来要命。我的建议是用设备ID 测点ID 时间戳 值的纵表结构CREATE TABLE plc_data ( id BIGINT AUTO_INCREMENT PRIMARY KEY, device_id VARCHAR(32) NOT NULL, tag_name VARCHAR(64) NOT NULL, ts DATETIME(3) NOT NULL, value DOUBLE, quality TINYINT DEFAULT 0, INDEX idx_device_ts (device_id, ts), INDEX idx_tag_ts (tag_name, ts) );这样加测点只是加配置不用改表。quality字段用来标记数据质量0 正常1 超时2 越界等方便后续排查。如果数据量特别大比如每秒几万点可以考虑时序数据库如 InfluxDB 或 TDengine写入性能比关系库强很多。但如果是中小项目MySQL 加好索引完全够用。3. 实操过程与核心环节实现3.1 环境准备与依赖安装先把环境搭起来。Python 版本建议 3.8 以上我用的是 3.10稳定。依赖库pip install pymcprotocol pymysql如果要用连接池再加dbutils要读配置文件加pyyaml或configparser。提示生产环境建议用虚拟环境venv 或 conda别把系统 Python 搞乱了。另外Windows 上跑采集程序的话记得把电源计划设成高性能别让系统休眠把程序挂了。3.2 采集层的实现采集层的核心是一个循环读数据 → 打时间戳 → 放入队列。但要做得好得处理连接管理、异常、超时。import time import queue import threading from pymcprotocol import Type3E class PlcCollector(threading.Thread): def __init__(self, ip, port, tags, interval, data_queue): super().__init__(daemonTrue) self.ip ip self.port port self.tags tags # [{name: temp1, device: D100, size: 1}, ...] self.interval interval self.data_queue data_queue self.plc None self.running True def connect(self): try: self.plc Type3E() self.plc.connect(self.ip, self.port) self.plc.setaccessopt(commtypebinary) return True except Exception as e: print(f连接失败: {e}) return False def run(self): while self.running: if self.plc is None: if not self.connect(): time.sleep(2) continue try: ts time.time() for tag in self.tags: values self.plc.batchread_wordunits( headdevicetag[device], readsizetag[size] ) for i, v in enumerate(values): self.data_queue.put({ tag: f{tag[name]}_{i} if tag[size] 1 else tag[name], ts: ts, value: v, quality: 0 }) except Exception as e: print(f采集异常: {e}) try: self.plc.close() except: pass self.plc None time.sleep(0.5) continue time.sleep(self.interval)这段代码有几个关键点时间戳统一取一次采集周期内读的所有点用同一个时间戳。这样后续做趋势分析时同一时刻的数据是对齐的。如果每个点单独取时间会有毫秒级偏差做相关性分析时很别扭。异常后置空连接捕获异常后把self.plc置为 None下一轮循环会重新连接。这比在 except 里直接重连更安全避免异常嵌套。daemon 线程设成守护线程主程序退出时自动结束不用手动 join。3.3 缓冲层的设计缓冲层用queue.Queue最简单但有个问题进程崩溃时内存里的数据就丢了。如果数据不能丢得用持久化缓冲。我的做法是内存队列 定期落盘 SQLite双保险import sqlite3 import queue class PersistentBuffer: def __init__(self, db_pathbuffer.db, max_memory10000): self.mem_queue queue.Queue(maxsizemax_memory) self.db_path db_path self._init_db() def _init_db(self): conn sqlite3.connect(self.db_path) conn.execute( CREATE TABLE IF NOT EXISTS buffer ( id INTEGER PRIMARY KEY AUTOINCREMENT, tag TEXT, ts REAL, value REAL, quality INTEGER ) ) conn.commit() conn.close() def put(self, item): try: self.mem_queue.put_nowait(item) except queue.Full: # 内存满了落盘 self._flush_to_disk([item]) def _flush_to_disk(self, items): conn sqlite3.connect(self.db_path) conn.executemany( INSERT INTO buffer (tag, ts, value, quality) VALUES (?,?,?,?), [(i[tag], i[ts], i[value], i[quality]) for i in items] ) conn.commit() conn.close()这样即使进程挂了重启后可以从 SQLite 里把没入库的数据捞出来补上。3.4 入库层的实现入库层从队列取数据攒批写入。关键参数是批大小和超时时间攒够 500 条就写或者距上次写入超过 1 秒也写两者取先到。import pymysql import time class DbWriter(threading.Thread): def __init__(self, buffer, db_config, batch_size500, flush_interval1.0): super().__init__(daemonTrue) self.buffer buffer self.db_config db_config self.batch_size batch_size self.flush_interval flush_interval self.running True def run(self): conn pymysql.connect(**self.db_config) batch [] last_flush time.time() while self.running: try: item self.buffer.mem_queue.get(timeout0.1) batch.append(item) except queue.Empty: pass now time.time() if len(batch) self.batch_size or (batch and now - last_flush self.flush_interval): self._write_batch(conn, batch) batch [] last_flush now def _write_batch(self, conn, batch): sql INSERT INTO plc_data (device_id, tag_name, ts, value, quality) VALUES (%s,%s,%s,%s,%s) rows [(self.db_config[device_id], i[tag], time.strftime(%Y-%m-%d %H:%M:%S, time.localtime(i[ts])), i[value], i[quality]) for i in batch] try: with conn.cursor() as cur: cur.executemany(sql, rows) conn.commit() except Exception as e: print(f入库失败: {e}) conn.rollback() # 失败的数据重新放回缓冲或者落盘 for item in batch: self.buffer.put(item)这里有个细节入库失败的数据要回退到缓冲不能直接丢。但要注意别无限重试导致死循环可以加个重试计数器超过阈值就落盘告警。3.5 参数计算与调优实例举个实际例子。假设一条产线有 3 台 PLC每台要采集 200 个 D 寄存器采集周期 500ms。单次读取时间估算200 个字一次请求搞定没超 960 上限。局域网 RTT 约 10ms加上 PLC 处理时间单次约 15ms。3 台设备串行读一轮约 45ms远小于 500ms 周期余量充足。数据量估算3 台 × 200 点 × 2 次/秒 1200 点/秒。批量 500 条写一次约 2.4 次/秒的写入频率MySQL 毫无压力。队列容量假设入库偶尔卡顿 10 秒需要缓冲 12000 条。内存队列设 20000 比较稳妥超了就落盘。数据库增长1200 点/秒 × 86400 秒 约 1 亿条/天。这个量级 MySQL 单表扛不住必须按天分表或者用分区表。我一般用plc_data_20240101这种命名每天建一张新表历史表可以压缩归档。注意分表之后查询要跨表 union比较麻烦。如果查询需求复杂建议直接上时序数据库省心。4. 常见问题与排查技巧实录4.1 通信类问题速查现象可能原因排查方法解决连接超时IP/端口错、网络不通ping 测试、telnet 端口检查网线、确认 PLC IP连接被拒绝PLC 未开 MC 协议、端口被占查 PLC 参数设置在 GX Works 里开启以太网通信读到全 0软元件地址错、PLC 未运行用 GX Works 在线监控对比核对地址和软元件代码读数据错位大小端搞反、字/位混淆抓包看原始字节确认二进制/ASCII 格式偶发超时网络抖动、PLC 负载高记录超时频率加重试、降低采集频率结束代码非 0点数超限、地址越界看响应帧结束代码减少单次点数、检查地址范围4.2 那些文档里不会写的坑坑一PLC 的以太网模块有连接数限制。FX3U-ENET 模块一般只支持 8 个以内的并发连接Q 系列多一些但也有限。如果你开了多个 Python 进程同时连很容易连满。解决办法是一个 PLC 只用一个连接所有采集需求走同一个连接复用。坑二MC 协议的监视定时器别设太小。这个参数是 PLC 等待请求完成的超时单位 250ms。设成 0x0001250ms在网络稍慢时就容易超时。我一般设 0x00104 秒给足余量。坑三批量读的地址要连续。MC 协议批量读是按起始地址 点数来的中间不能跳。如果你要读 D100 和 D200只能分两次请求或者一次读 D100~D200101 个字把中间没用的也读回来。后者更快但要注意别读到未定义的区域有些 PLC 会报错。坑四时间戳的时区问题。Python 的time.time()是 UTC 时间戳存数据库时如果直接转字符串可能和本地时间差 8 小时。建议统一用 UTC 存储展示时再转本地时区避免跨时区部署时混乱。坑五数据库连接会断。MySQL 默认 8 小时空闲就断开连接采集程序跑一晚上第二天早上发现入库全失败。解决办法是加连接保活conn.ping(reconnectTrue)或者定期重连。4.3 性能调优的几个实操技巧技巧一用二进制格式而不是 ASCII。MC 协议的 3E 帧支持二进制和 ASCII 两种格式二进制效率高得多数据量小一半。pymcprotocol 里用setaccessopt(commtypebinary)设置。技巧二合并读取请求。前面说过把相邻地址合并成一次读。我实测过读 10 个分散地址10 次请求和读 1 个连续块1 次请求耗时差 5~8 倍。技巧三入库用executemany而不是循环execute。这个差别巨大1000 条数据循环 execute 要 2~3 秒executemany 只要 50ms 左右。技巧四数据库索引别建太多。每个索引都会拖慢写入。plc_data表建两个索引就够了多了写入性能直线下降。技巧五采集和入库用不同的进程。Python 有 GIL多线程在 CPU 密集场景下跑不满多核。如果采集频率很高建议用multiprocessing把采集和入库分成两个进程各自跑满一个核。4.4 数据完整性保障数据采集最怕的就是丢数据。我一般从三个层面保障第一层采集层重试。读失败重试 3 次还不行标记 quality1超时数据照样入库只是标记为异常。这样至少知道那个时刻数据有问题而不是完全缺失。第二层缓冲层落盘。内存队列满了或者进程退出时数据落 SQLite。重启后先消费 SQLite 里的积压数据。第三层入库层幂等。用(device_id, tag_name, ts)做唯一索引重复插入用INSERT IGNORE或ON DUPLICATE KEY UPDATE避免重试导致数据重复。ALTER TABLE plc_data ADD UNIQUE KEY uk_device_tag_ts (device_id, tag_name, ts);这样即使因为重试导致同一条数据被写两次数据库也会自动去重。4.5 监控与告警程序跑起来不是就完事了得有监控。我一般加几个关键指标采集成功率每分钟统计一次低于 95% 就告警。队列积压量超过阈值说明入库跟不上要查数据库。入库延迟数据产生到入库的时间差超过 10 秒要关注。连接状态PLC 连接断开要立即告警。这些指标可以写个简单的 HTTP 接口暴露出来用 Prometheus 抓取或者直接写日志用 ELK 分析。小项目的话写个定时任务检查异常发邮件或钉钉就行。def health_check(): stats { queue_size: buffer.mem_queue.qsize(), collect_ok_rate: collector.ok_count / max(collector.total_count, 1), last_write_ts: writer.last_write_ts, } if stats[queue_size] 50000: send_alert(队列积压严重) if time.time() - stats[last_write_ts] 30: send_alert(入库超过30秒无写入)这套东西搭下来基本能保证采集链路稳定运行。我在一个汽车零部件厂的项目里这套架构连续跑了两年多除了几次网络故障和数据库维护没出过大问题。关键就是别把鸡蛋放一个篮子里采集、缓冲、入库各司其职任何一环出问题都不至于全盘崩溃。最后分享一个我踩过的坑有次现场调试采集程序跑得好好的突然所有数据都变成 0。查了半天发现是 PLC 程序被人下载了新版本D 寄存器的地址偏移变了。所以PLC 程序和采集配置要版本对应每次 PLC 程序变更采集端的地址映射表也要同步更新最好做个版本校验机制不匹配就告警。这个教训值好几千块的现场差旅费。