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

拿到这个需求的时候，我就知道这绝对不是一个单纯的“用Python读PLC数据”的教程级问题。做过工控上位机的人都有这种体会：把三菱PLC的数据读回来，用MC协议通上信，这只是万里长征第一步；真正决定项目成败的，是数据从PLC侧产生开始，到最终落入数据库的整个链条上，每个环节的时序是怎么设计、怎么衔接、怎么应对异常。数据入库时序设计，才是这里的灵魂。

这篇文章我会按照自己在实际项目里的处理方式，把整条链路拆开讲：从MC协议通信层的搭建，到数据解析、采集循环，再到入库策略和多级缓冲，最后附带一份我在现场踩坑之后的排查速查表。无论你是刚接触三菱PLC通信的Python新手，还是已经在做MES、SCADA数据采集的工程师，这篇内容应该都能给你一些参考。

1. 整体设计思路：先搞清数据链路里藏着的三个时钟

1.1 这是一个什么系统，解决什么问题

这里讨论的系统，可以概括为一条单向数据管道：

三菱PLC（FX3U、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-ENET），FX5U和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() # 连接PLC，plc.setaccessopt() 可以设置通信选项 plc.setaccessopt( comm_type="binary", # 或者 "ascii" frame="3E" ) plc.connect("192.168.1.10", 5007) # 读取D100开始的连续20个字，返回类型是一个列表 values = plc.batchread_wordunits(headdevice="D100", readsize=20) print(values) # 读取M100开始的连续10个位元件 bits = plc.batchread_bitunits(headdevice="M100", readsize=10) 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(headdevice="D100", readsize=100) 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) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

这里几个字段值得解释：

• 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( creator=pymysql, maxconnections=20, host="127.0.0.1", port=3306, user="root", password="123456", database="plc_data", charset="utf8mb4" ) 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(maxsize=5000)，当队列满时，新数据就没法再放进去。这时候你必须做取舍：

• 方案一：阻塞采集线程，直到队列有空间。这会导致采集线程卡住，PLC通信超时，最终数据链路阻塞。除非你严格要求数据不能丢，否则不推荐。
• 方案二：丢弃最旧的数据或丢弃新数据，并记录丢数据计数。推荐，这是一种“可控降级”。

我用的是“丢弃新数据+记录日志”方案：

try: queue.put(record, block=False) except queue.Full: dropped_count += 1 # 记录日志告警

这样即使数据库故障，采集线程依然在跑，只是新数据暂时进不了队列。数据链路的“实时性”被破坏，但“可用性”保住了，数据库恢复后队列里的数据可以继续消费。这种丢数据的决策一定要提前和业务方说明，达成共识。

4.5 断线续传：要不要做持久化缓冲

如果数据真的很重要，不允许丢，那就要考虑“断线续传”了——在采集线程和最终数据库之间，再加一层本地持久化缓冲。

常见的做法是在工控机上再跑一个本地SQLite，入库线程先把数据写入本地SQLite（本地磁盘写入，速度极快、极其稳定），然后再由一个同步任务把SQLite里的数据转发到远端MySQL或InfluxDB。远端数据库不可用时，数据就暂时积压在本地SQLite里，等恢复后再继续同步。

这样做的好处非常明显：本地SQLite的可靠性远高于网络数据库，而且写入速度比MySQL快很多，天然适合做这个缓冲层。缺点是架构多了一层，数据会存在“最终一致”的延迟。对于大多数数据采集项目来说，这个延迟完全可接受。

5. 常见问题与排查技巧实录

5.1 通信层典型故障速查表

下面这组问题是实际项目里最高频的，按“一眼能看出原因”到“需要逐步排查”排序。

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秒以内、极端情况下数据库中断后会丢失最近的哪些数据、队列溢出后是什么行为。这些内容才是这套数据采集系统的“设计契约”，比任何代码注释都更能帮助后来接手的人。

如果这篇文章里的某一段做法能帮你在现场少踩一个坑，那这篇内容就没有白写。