三菱PLC数据采集到数据库的时序设计：Python+MC协议实战

拿到这个需求的时候我就知道这绝对不是一个单纯的“用Python读PLC数据”的教程级问题。做过工控上位机的人都有这种体会把三菱PLC的数据读回来用MC协议通上信这只是万里长征第一步真正决定项目成败的是数据从PLC侧产生开始到最终落入数据库的整个链条上每个环节的时序是怎么设计、怎么衔接、怎么应对异常。数据入库时序设计才是这里的灵魂。这篇文章我会按照自己在实际项目里的处理方式把整条链路拆开讲从MC协议通信层的搭建到数据解析、采集循环再到入库策略和多级缓冲最后附带一份我在现场踩坑之后的排查速查表。无论你是刚接触三菱PLC通信的Python新手还是已经在做MES、SCADA数据采集的工程师这篇内容应该都能给你一些参考。1. 整体设计思路先搞清数据链路里藏着的三个时钟1.1 这是一个什么系统解决什么问题这里讨论的系统可以概括为一条单向数据管道三菱PLCFX3U、Q系列、L系列均可通过内置以太网口或扩展以太网模块对外提供MC协议通信服务。上位机是一台Windows或Linux工控机安装了Python环境通过pymcprotocol这类库按轮询方式从PLC中读取D寄存器、M软元件的数据做必要的解析和单位换算后写入到MySQL、SQLite或时序数据库比如InfluxDB中供后续的看板展示、报表统计或算法分析使用。这套方案解决的典型问题包括车间里PLC没有配套的SCADA系统但业务方又需要实时掌握产线数据MES系统需要采集设备状态和产量数据但不想采购昂贵的网关硬件又或者你只是想把PLC里的温度、压力、转速等参数存档方便事后回溯分析。1.2 为什么选Python加MC协议而不是Modbus TCP先说结论在三菱PLC场景下MC协议是比Modbus TCP更好的选择Python则是这套场景下开发效率最高的语言。MC协议是三菱自家定义的通信协议全称MELSEC Communication Protocol走以太网时通常使用3E帧。它最大的好处是原生支持三菱全系软元件地址比如D寄存器、M继电器、X/Y输入输出、ZR文件寄存器等不需要像Modbus那样做地址映射表。你用Modbus TCP去读三菱D100往往要先把D100换算成Modbus地址400101之类的中间错一位就全错。而MC协议直接给定住址代码里写“D100”读出来的就是D100心智负担小很多。Python的优势就更直接了开发速度快pymcprotocol这样的库封装了协议细节你不用去啃MC协议的报文格式和二进制编码规范几行代码就能连上PLC读写数据后续要对接数据库、WebAPI、消息队列Python的生态都是现成的。和C#、Java比在这种中小型数据采集项目里Python的交付效率是碾压级的。1.3 时序设计的核心矛盾三个时钟速率不匹配这是整篇文章最重要的一节也是大多数人做数据入库时翻车的根本原因。任何一条“PLC数据入库”的链路里都至少存在三个不同的时钟PLC侧的数据产生时钟。这由PLC的扫描周期决定FX3U通常在几毫秒到十几毫秒的量级Q系列更快。也就是说D寄存器里的值是每几毫秒就可能变化一次的。Python采集线程的轮询时钟。受限于网络往返延迟RTT、Python解释器的执行效率、GIL的影响你实际能做到的轮询周期通常稳定在几十毫秒到几百毫秒之间。数据库的写入时钟。单条INSERT一条条提交几千行数据可能要好几秒钟但用批量提交executemany几千行可能只需要几百毫秒。三个时钟速率不一致就带来了一个最基本的问题PLC已经更新了100次数据你的Python只读了1次数据库只写了1次。那99次的变化被“天然丢弃”了。这几乎是所有数据采集系统的宿命你能做的不是让三个时钟完全对齐而是让“数据的流向”可控、可预期。也就是说你要明确告诉业务方这个系统采样的实际频率是多少、PLC数据的刷新率是多少、入库的吞吐量是多少然后再在代码层面通过缓冲、批量、超时丢弃等机制保证系统在任何工况下都不会崩、不会丢、不会堵。这就是时序设计的本质不是在追求极致速度而是在管理吞吐节奏和异常行为。2. MC协议通信层的搭建与实操要点2.1 通信前的准备确认PLC侧参数动手写代码之前必须先确认PLC侧的硬件和软件配置这是很多新手最容易忽略的一步。你需要确认以下信息PLC型号FX3U、FX5U还是Q系列、L系列。不同系列打开以太网功能的方式不太一样FX3U需要扩展以太网模块比如FX3U-ENETFX5U和Q系列则很多型号自带网口。PLC的IP地址和端口号三菱PLC的MC协议以太网通信默认端口号是5007这个端口在GX Works2或GX Works3的以太网模块参数里配置。通信帧格式MC协议3E帧有两种编码方式一种叫二进制码一种叫ASCII码。pymcprotocol默认支持二进制格式如果PLC侧配置成了ASCII格式也没关系库里有参数可以切换。软元件地址范围搞清楚你要读的数据存放在哪些软元件区域比如D100到D200是温度数据、M100到M150是设备状态这些都要和PLC程序作者确认清楚最好形成一份地址映射表。确认的方式很简单打开GX Works2在工程里找到“PLC参数”或“以太网模块设置”截图确认上面的IP、端口、通信协议设置保存后下载到PLC并重启。2.2 Python侧依赖库与最小通信Demo通信库我推荐使用pymcprotocol它是对MC协议3E帧的一个纯Python实现无需安装额外的本地依赖pip安装即可。pip install pymcprotocol安装完成后可以先用下面这个最小示例验证通路import pymcprotocol # 创建一个3E帧的以太网客户端对象 plc pymcprotocol.Type3E() # 连接PLCplc.setaccessopt() 可以设置通信选项 plc.setaccessopt( comm_typebinary, # 或者 ascii frame3E ) plc.connect(192.168.1.10, 5007) # 读取D100开始的连续20个字返回类型是一个列表 values plc.batchread_wordunits(headdeviceD100, readsize20) print(values) # 读取M100开始的连续10个位元件 bits plc.batchread_bitunits(headdeviceM100, readsize10) print(bits) # 断开连接 plc.close()就这么简单。batchread_wordunits是读字元件的方法batchread_bitunits是读位元件的方法。这里的headdevice直接写成“D100”“M100”这种三菱原生地址格式不需要做任何偏移换算。2.3 MC协议通信层的三个隐藏坑代码看着简单但实际跑起来以下三个问题我几乎在每次项目中都会遇到。第一个坑是数据类型。batchread_wordunits读出来的数据默认是无符号16位整数uint16。三菱的D寄存器本质上是16位存储单元如果PLC里存放的是有符号整数或者浮点数你必须自己做转换。比如PLC里用32位浮点数存储温度那实际上是占用了D100和D101两个连续的寄存器低16位在D100、高16位在D101。Python这边需要把两个uint16拼成32位再用struct.unpack把它解析成float。import struct # 假设读回的两个寄存器值是 raw_low, raw_high # 注意三菱是低字在前即低位寄存器是D100高位寄存器是D101 packed struct.pack(HH, raw_low, raw_high) value struct.unpack(f, packed)[0]第二个坑是批量读取大小。MC协议虽然支持一次读多个地址但单次读取的寄存器数量不是无限的超过上限比如一次读1000个字会报错具体限制和PLC型号与帧格式有关就需要拆成多次批量读。所以设计地址映射表的时候尽量把连续地址放在一起减少通信次数。第三个坑是超时和重连机制。pymcprotocol.connect默认超时时间可能不够长。PLC在重启、断电、网络抖动的时候Python端的连接会假死。你需要给通信层封装一层“心跳检测”和“自动重连”逻辑而不是在业务代码里裸调batchread。心跳检测最简单的方式就是周期性地读取一个固定地址例如D0里的运行状态字如果连续几次失败或超时就调用close()然后重新connect()。3. 数据解析与采集循环的精细设计3.1 先做一张软元件映射表再谈采集我见过太多项目代码里直接写plc.batchread_wordunits(D100, 1)然后下一个地方又写plc.batchread_wordunits(D200, 1)读回来之后零零散散塞进数据库。这种代码写到最后就是维护噩梦。PLC程序一改地址你得满代码地找D100、D200改到怀疑人生。正确做法是做一张“软元件映射表”用Python的数据结构描述所有采集点的信息# tag_mapping.py TAG_MAPPING [ # 每个元素代表一个采集点 { name: line1_temperature, # 入库字段名 device: D100, # 起始软元件地址 type: float32, # 数据类型 unit: ℃, # 工程单位 scale: 0.1, # 换算系数 offset: 0, # 换算偏移 }, { name: line1_speed, device: D102, type: int16, unit: rpm, scale: 1, offset: 0, }, ]有了这张表你的采集代码就可以写成完全通用的形式遍历配置表、根据类型解析字节、做工程换算、组装成记录、推入队列。PLC地址一变你只需要改配置表不用改代码。这一点在后续系统维护期价值是无法估量的。3.2 采集周期的确定不是越快越好采集周期怎么定很多人的第一反应是“越快越好”。但这里有几个硬约束。第一个约束是PLC侧通信性能。MC协议单次轮询数据量越大PLC的响应时间就越长。如果你一次读200个寄存器PLC要花几毫秒到几十毫秒才能响应。轮询周期如果设得太短比如10ms一次PLC根本忙不过来通信错误率会急剧上升。第二个约束是网络往返时间RTT。工控网一般都在局域网内RTT在1ms以内但有些项目PLC通过工业交换机和多个路由器连接RTT可能飙到10ms以上。RTT越大你可以达到的稳定轮询频率就越低。第三个约束是数据入库的吞吐量。如果数据库写入速度跟不上采集速度队列会不断堆积最终导致内存暴涨或数据延迟越来越大。结合这几个约束我的经验值是常规设备监控场景轮询周期设在200ms到500ms是性价比最高的区间。快速响应场景比如报警联动的数据点可以针对个别关键地址单独开一路100ms的快速轮询但绝不要让所有采集点都跑100ms。实时性要求极高小于50ms的场景说实话不适合用Python轮询方案应该考虑PLC直接向上位机主动发送或者换用实时性更强的通信方式。你可以先用一个简单的往返测试脚本估算一下当前网络条件下单次批量读的耗时import time import pymcprotocol plc pymcprotocol.Type3E() plc.connect(192.168.1.10, 5007) # 测试读取100个字的耗时 start time.perf_counter() plc.batchread_wordunits(headdeviceD100, readsize100) cost_ms (time.perf_counter() - start) * 1000 print(f单次批量读耗时: {cost_ms:.2f} ms) plc.close()如果单次耗时在5ms以内你的采集周期可以轻松设在200ms如果单次耗时到了50ms那采集周期至少要放宽到500ms以上。3.3 采集线程与解析逻辑的分离采集循环不要把所有事情都揉在一起。我推荐的模式是一个“采集线程”专门负责和PLC通信只做两件事批量读数据然后把原始字数组按地址映射表解析成带标签名的字典列表。一个“处理线程”从队列里拿数据做工程换算、加时间戳、组装成入库记录批量写入数据库。这两个线程之间用一个queue.Queue解耦。这样即使数据库写入偶尔变慢采集线程也不会被阻塞太久反过来数据库写入异常也不会直接导致PLC通信中断。这里有一个非常重要的架构理念采集线程永远不要直接读写数据库。否则一旦数据库连接卡死你的采集线程也会卡死最终PLC通信也跟着超时整个数据管道就瘫痪了。3.4 时间戳的坑到底用哪边的钟时间戳的选取是数据入库项目里最容易出问题的地方而且出了问题往往很久才发现。如果你用time.time()在Python侧打时间戳那么每条数据的入库时间表示的是“上位机接收到数据的时刻”。这个时间戳的意义在于记录数据的到达时间但它并不严格代表数据在PLC侧产生的时间。尤其在轮询周期较大的时候你读到的那条数据极可能是几百毫秒之前PLC就已经产生好的。所以在具体设计时要先想清楚业务需求如果数据库里的数据用于趋势展示、报表统计大多数情况下用“上位机接收时间”就够了因为PLC侧的精确产生时间很难获取到除非PLC程序主动记录时间戳。如果你的PLC程序里维护了一个运行时间计数器比如用D寄存器存储开机以来的毫秒数你可以把它读出来换算成PLC侧时间但这里涉及PLC时钟同步问题大概率你会花掉大量时间在时钟校准上。我的建议是默认使用上位机接收时间作为时间戳但必须把“采集周期”的数值一并存下来。这样在回溯数据时你可以知道相邻两条数据之间的理论时间间隔避免把数据误判成均匀时间序列。4. 数据入库与多级缓冲的时序设计4.1 数据库选型和表结构设计数据入库的选型要结合数据量、查询频率来定。数据量不大一天几十万条以内、查询场景简单用SQLite就够了。SQLite是文件型数据库部署零成本Python内置sqlite3模块最适合做单机版数据采集程序。数据量较大、需要多台设备集中查询报表用MySQL或PostgreSQL。MySQL访问用pymysql库配合DBUtils的连接池使用。如果是高频采集加实时看板建议直接上时序数据库InfluxDB它本身就是为时序数据设计的写入吞吐和查询性能都比关系型数据库好几个量级。无论用哪种数据库表结构都有一些通用字段设计经验CREATE TABLE plc_data ( id BIGINT AUTO_INCREMENT PRIMARY KEY, device_id VARCHAR(32) NOT NULL, tag_name VARCHAR(64) NOT NULL, tag_value DOUBLE NOT NULL, quality INT DEFAULT 0, collect_time DATETIME(3) NOT NULL, create_time DATETIME(3) DEFAULT CURRENT_TIMESTAMP(3), INDEX idx_device_tag_time (device_id, tag_name, collect_time) ) ENGINEInnoDB DEFAULT CHARSETutf8mb4;这里几个字段值得解释collect_time采集时间戳也就是第3.4节说的“上位机接收时间”。quality质量戳0代表正常非0代表异常比如通信中断后补采的数据、解析告警的数据这个字段在分析历史数据时非常关键方便你过滤掉“脏数据”。create_time入库时间和collect_time区分开。这两个字段的差值就可以用来衡量“数据从采集到入库的延迟”是排查性能瓶颈的重要指标。4.2 连接池与批量写入别一条一条插入在入库环节最容易犯的错误是“每采集到一条数据就执行一条INSERT”。一次轮询如果解析出100个采集点就意味着100次INSERT这个效率非常低。我实测下来同样是写入1000条数据到MySQL逐条INSERT耗时可能超过5秒而使用executemany批量提交只要200到500毫秒性能差距在十倍以上。所以入库代码必须使用批量写入加连接池。连接池可以用dbutils库的PooledDB它避免了每次写入都重新建立TCP连接的巨大开销。批量写入的伪代码思路如下import pymysql from dbutils.pooled_db import PooledDB # 初始化连接池 pool PooledDB( creatorpymysql, maxconnections20, host127.0.0.1, port3306, userroot, password123456, databaseplc_data, charsetutf8mb4 ) def batch_insert(records): records: 一组待入库的字典列表 conn pool.connection() cursor conn.cursor() sql INSERT INTO plc_data (device_id, tag_name, tag_value, quality, collect_time, create_time) VALUES (%s, %s, %s, %s, %s, %s) data [ ( r[device_id], r[tag_name], r[tag_value], r[quality], r[collect_time], r[create_time], ) for r in records ] cursor.executemany(sql, data) conn.commit() cursor.close() conn.close()每批攒够比如200条或者距离上次写入超过1秒就执行一次批量写入。这套机制下MySQL的写入吞吐量轻松可以做到每秒万条以上远高于采集速度。4.3 多级缓冲三速率模型的落地解法前面我们讲了三个时钟速率不匹配的问题那到底怎么落地答案是分层缓冲。第一级缓冲是“通信层的单次批量读”。这一步已经把PLC侧的快速变化压缩成了周期采样点。200ms轮询一次读回100个点天然就是一批100条的数据而不是每个点单独做一次通信。第二级缓冲是“内存队列”。采集线程解析完原始数据后把带时间戳的记录推入queue.Queue入库线程从这个队列里消费。这个队列起到了削峰填谷的作用——数据库偶尔变慢时队列可以暂存数据不至于立刻丢数据。第三级缓冲是“批量攒批”。入库线程不是拿到一条写一条而是把队列里的数据攒成一批达到阈值或者到达时间阈值才批量写入。这就是刚才4.2说的机制。在这个三级缓冲模型下整个系统的数据流是这样的PLC数据按扫描周期更新毫秒级。Python采集线程按轮询周期读取百毫秒级每次读取产生一批记录。记录进入内存队列待入库线程消费消费速率取决于数据库性能。入库线程攒批后批量写入秒级一批。这个模型的最大价值在于解耦了三者之间的速率依赖。只要队列不溢出任何一方的短暂抖动都不会撕裂整个链路。4.4 队列溢出的取舍宁可丢弃不可阻塞内存队列不是无限大的必须有一个上限。假设你的采集周期是200ms每轮产出的记录数是一批那每秒就产生约5批数据。如果数据库停了10秒钟队列里会积压约50批数据。如果你一个批次是100条50批就是5000条记录每条记录按100字节算也就是500KB内存还可以接受。但如果数据库停了10分钟积压的就是30万条内存占用30MB以上。如果数据库停了1小时这个数字就是1800万条任何机器都扛不住。所以队列必须设置最大长度比如queue.Queue(maxsize5000)当队列满时新数据就没法再放进去。这时候你必须做取舍方案一阻塞采集线程直到队列有空间。这会导致采集线程卡住PLC通信超时最终数据链路阻塞。除非你严格要求数据不能丢否则不推荐。方案二丢弃最旧的数据或丢弃新数据并记录丢数据计数。推荐这是一种“可控降级”。我用的是“丢弃新数据记录日志”方案try: queue.put(record, blockFalse) except queue.Full: dropped_count 1 # 记录日志告警这样即使数据库故障采集线程依然在跑只是新数据暂时进不了队列。数据链路的“实时性”被破坏但“可用性”保住了数据库恢复后队列里的数据可以继续消费。这种丢数据的决策一定要提前和业务方说明达成共识。4.5 断线续传要不要做持久化缓冲如果数据真的很重要不允许丢那就要考虑“断线续传”了——在采集线程和最终数据库之间再加一层本地持久化缓冲。常见的做法是在工控机上再跑一个本地SQLite入库线程先把数据写入本地SQLite本地磁盘写入速度极快、极其稳定然后再由一个同步任务把SQLite里的数据转发到远端MySQL或InfluxDB。远端数据库不可用时数据就暂时积压在本地SQLite里等恢复后再继续同步。这样做的好处非常明显本地SQLite的可靠性远高于网络数据库而且写入速度比MySQL快很多天然适合做这个缓冲层。缺点是架构多了一层数据会存在“最终一致”的延迟。对于大多数数据采集项目来说这个延迟完全可接受。5. 常见问题与排查技巧实录5.1 通信层典型故障速查表下面这组问题是实际项目里最高频的按“一眼能看出原因”到“需要逐步排查”排序。故障现象可能原因快速排查方法connect超时PLC IP填错/PLC未开机/端口不对先ping IP再用telnet IP 5007测端口通不通connect成功但读不到数据帧格式配置错binary/ascii不匹配用plc.setaccessopt()切换格式重试读回来的数据全部是0或异常大数地址映射错、数据类型解析错用GX Works2在线监视D寄存器对比批量读超过上限报错单次readsize超限拆小批量每次读几百个字以内偶发超时/连接被重置网络抖动或PLC负载过高降低轮询频率检查交换机是否丢包停电重启后程序无法自动恢复缺少自动重连逻辑加心跳检测和重连定时器5.2 数据入库层典型故障MySQL写入缓慢先看是不是没用连接池再看是不是逐条INSERT。改用连接池加executemany后基本都能解决。数据延迟越来越大队列积压的典型症状。先确认数据库写入是否正常再检查是否连接池不够用。如果数据库正常但队列还是膨胀那要降低采集频率或者扩大批量阈值。时间戳错乱如果多台采集程序同时往一个数据库写不同机器之间的系统时钟不一致会导致时间戳错乱。解决办法是统一用NTP同步工控机时间或者让时间戳以PLC侧时间为准。数据重复排查是否在程序重启时有重复补采逻辑。如果程序崩溃前已经写了一批数据但没提交重启后重新补采就会造成重复这时候要设计一个“数据批次号”字段来幂等去重。5.3 现场排查流程的思考框架我在现场处理问题时习惯按“先链路、再数据、再容量”的顺序排查先确认链路。从Python端能否正常batchread如果能通信层没问题如果不能按上面的速查表逐步排查网络、端口、格式。再确认数据。读回来的数值和GX Works2在线监视的数值是否一致如果不一致解析逻辑有问题。最后确认容量。数据入库速度是否大于采集速度队列有没有积压数据库连接池是否打满按这个顺序走完90%的问题都能定位出来。5.4 实战中的两个小技巧第一个技巧是“先落本地再上云”。我在一个产线项目里就是让Python先写本地SQLite再由另一个进程同步到中央MySQL。后来MySQL服务器经历了一次故障现场数据完全没有丢SQLite里积压了两小时的数据恢复后自动补传效果非常理想。第二个技巧是把采集程序的运行状态也暴露成数据点。比如把“队列积压量”“累计丢数据条数”“最近一次通信耗时”“PLC连接状态”也写成一条记录定期写入数据库或推送成告警。这样数据采集系统本身的状态也是可观测的。排查问题时你不必再靠猜。6. 关于这套方案我自己的几点体会这套“Python加MC协议加数据入库”的架构我前前后后在好几个产线项目中落地过。最大的感受是不要一开始就去优化性能先把数据链路的完整性和可观测性搭起来。很多工程师拿到需求第一句话是“我要把采集周期做到50ms”这是一个常见的误区。我觉得正确的打开方式是先按500ms的轮询周期把整条链路跑通确认数据能稳定入库然后用压力和故障测试去检验这套链路在异常情况下的表现最后才去调整轮询周期和批量参数把性能榨到业务需要的水平。还有一个建议在项目交付时一定要把“时序设计”的约束写进交接文档。比如系统采样周期是200ms、数据入库延迟在1秒以内、极端情况下数据库中断后会丢失最近的哪些数据、队列溢出后是什么行为。这些内容才是这套数据采集系统的“设计契约”比任何代码注释都更能帮助后来接手的人。如果这篇文章里的某一段做法能帮你在现场少踩一个坑那这篇内容就没有白写。