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 字节），0x0000 表示成功，其他值就是错误码。

为什么要懂这个？因为当你用库读不到数据的时候，抓包一看，可能是软元件代码写错了，或者点数超了限制。比如 D 寄存器一次最多读 960 个字（3E 帧二进制），你写 1000 就会报错。这些细节，库的文档不一定写清楚。

2.2 Python 库的选型对比

Python 操作三菱 MC 协议，常用的库有这么几个：

我的经验是：采集频率在 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(headdevice="D100", readsize=10) plc.close()

看着简单，但生产环境不能这么写。每次读都 connect/close，开销大不说，网络一抖就抛异常。正确做法是长连接 + 异常重连 + 心跳保活。

2.3 时序设计的核心参数

时序设计说白了就是回答几个问题：多久读一次？一次读多少？读失败等多久重试？数据攒多久写一次库？

采集周期：取决于工艺要求。如果是监控温度、压力这种慢变量，1 秒一次足够；如果是抓取瞬间的触发信号，可能要 100ms 甚至更快。但要注意，MC 协议单次请求的往返时间（RTT）在局域网内大概 5~20ms，所以理论上限也就 50Hz 左右，再快就不现实了。

单次读取点数：D 寄存器一次最多 960 字，M 继电器一次最多 7168 位。实际项目中，我一般把需要连续读取的地址合并成一块，减少请求次数。比如要读 D100~D120 和 D200~D210，与其发两次请求，不如一次读 D100~D210（110 个字），虽然多读了一些无用数据，但省了一次网络往返，整体更快。

重试策略：读失败不要立刻重试，容易雪崩。我一般用指数退避：第一次失败等 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__(daemon=True) 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(commtype="binary") 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( headdevice=tag["device"], readsize=tag["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_path="buffer.db", max_memory=10000): self.mem_queue = queue.Queue(maxsize=max_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_size=500, flush_interval=1.0): super().__init__(daemon=True) 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(timeout=0.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 通信类问题速查

4.2 那些文档里不会写的坑

坑一：PLC 的以太网模块有连接数限制。FX3U-ENET 模块一般只支持 8 个以内的并发连接，Q 系列多一些但也有限。如果你开了多个 Python 进程同时连，很容易连满。解决办法是一个 PLC 只用一个连接，所有采集需求走同一个连接复用。

坑二：MC 协议的监视定时器别设太小。这个参数是 PLC 等待请求完成的超时，单位 250ms。设成 0x0001（250ms）在网络稍慢时就容易超时。我一般设 0x0010（4 秒），给足余量。

坑三：批量读的地址要连续。MC 协议批量读是按起始地址 + 点数来的，中间不能跳。如果你要读 D100 和 D200，只能分两次请求，或者一次读 D100~D200（101 个字），把中间没用的也读回来。后者更快，但要注意别读到未定义的区域，有些 PLC 会报错。

坑四：时间戳的时区问题。Python 的time.time()是 UTC 时间戳，存数据库时如果直接转字符串，可能和本地时间差 8 小时。建议统一用 UTC 存储，展示时再转本地时区，避免跨时区部署时混乱。

坑五：数据库连接会断。MySQL 默认 8 小时空闲就断开连接，采集程序跑一晚上，第二天早上发现入库全失败。解决办法是加连接保活（conn.ping(reconnect=True)）或者定期重连。

4.3 性能调优的几个实操技巧

技巧一：用二进制格式而不是 ASCII。MC 协议的 3E 帧支持二进制和 ASCII 两种格式，二进制效率高得多，数据量小一半。pymcprotocol 里用setaccessopt(commtype="binary")设置。

技巧二：合并读取请求。前面说过，把相邻地址合并成一次读。我实测过，读 10 个分散地址（10 次请求）和读 1 个连续块（1 次请求），耗时差 5~8 倍。

技巧三：入库用executemany而不是循环execute。这个差别巨大，1000 条数据，循环 execute 要 2~3 秒，executemany 只要 50ms 左右。

技巧四：数据库索引别建太多。每个索引都会拖慢写入。plc_data表建两个索引就够了，多了写入性能直线下降。

技巧五：采集和入库用不同的进程。Python 有 GIL，多线程在 CPU 密集场景下跑不满多核。如果采集频率很高，建议用multiprocessing把采集和入库分成两个进程，各自跑满一个核。

4.4 数据完整性保障

数据采集最怕的就是丢数据。我一般从三个层面保障：

第一层：采集层重试。读失败重试 3 次，还不行标记 quality=1（超时），数据照样入库，只是标记为异常。这样至少知道那个时刻数据有问题，而不是完全缺失。

第二层：缓冲层落盘。内存队列满了或者进程退出时，数据落 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 程序变更，采集端的地址映射表也要同步更新，最好做个版本校验机制，不匹配就告警。这个教训值好几千块的现场差旅费。