1. 项目缘起与整体架构思路
三菱 PLC 在制造业现场保有量极大,尤其是 FX 系列和 Q 系列,很多产线设备、检测工位、包装机械的控制核心都是它。我接触过的项目中,十有八九需要把 PLC 里的实时数据采集上来存进数据库,用于追溯、报表或者 MES 对接。早期大家习惯用组态软件或者触摸屏自带的记录功能,但一旦涉及自定义逻辑、复杂清洗规则、多设备并发,组态软件就捉襟见肘了。用 Python 直接和三菱 MC 协议通信,再配合数据库做时序入库,是我这几年用得最顺手的一套方案。
这个项目的核心目标很明确:让 Python 程序周期性地从三菱 PLC 读取指定软元件的数据,经过必要的解析和清洗后,按照合理的时间节奏写入数据库,保证数据不丢、不重、不乱序。听起来简单,但真正落地时会遇到一堆细节问题——MC 协议是二进制还是 ASCII、软元件地址怎么换算、批量读取和单点读取怎么选、入库频率和 PLC 扫描周期怎么匹配、断线重连后数据怎么补、数据库写入压力怎么控制。这些才是决定项目成败的关键。
适合阅读这篇内容的人,应该是有一定 Python 基础、手头有三菱 PLC 或者准备对接三菱设备的工程师。如果你完全没接触过 PLC,也没关系,我会把 MC 协议的基本概念和软元件寻址方式讲清楚,保证你能跟着思路走。如果你已经是老手,可以直接跳到时序设计和避坑部分,那里有我踩过的坑和总结出来的参数。
整体架构上,我采用的是分层设计:最底层是通信层,负责和 PLC 建立 TCP 连接、收发 MC 协议报文;中间是采集调度层,负责按周期触发读取、管理软元件列表、处理异常重连;上层是数据处理与入库层,负责解析原始字节、做类型转换、按时间窗口批量写入数据库。三层之间通过队列或者回调解耦,避免因为数据库写入慢而阻塞 PLC 读取。这个设计的好处是每一层都可以独立调试和替换,比如通信层换成串口或者换成其他品牌协议,上层逻辑基本不用动。
为什么选择 TCP 而不是串口?因为现在大部分三菱 Q 系列和 FX5U 都自带以太网口,FX3U 加个 ENET 模块也能走以太网。TCP 的速率和稳定性远高于 RS485,而且 Python 的 socket 编程非常成熟。MC 协议本身支持二进制和 ASCII 两种帧格式,二进制效率更高,我一般优先用二进制。至于数据库,MySQL 和 PostgreSQL 都用过,时序数据量大的话 PostgreSQL 配合 TimescaleDB 或者直接用 InfluxDB 会更合适,但考虑到很多工厂的 IT 环境还是 MySQL 为主,这篇内容以 MySQL 为例来展开。
2. MC 协议通信核心细节与实操要点
2.1 MC 协议帧结构与软元件寻址
三菱 MC 协议全称是 MELSEC Communication Protocol,是三菱为自己 PLC 定义的一套应用层协议。它跑在 TCP 或者 UDP 之上,默认端口是 5000,也有用 6000 的。MC 协议有两种帧格式:3E 帧和 4E 帧,3E 帧最常用,结构也简单。一个典型的二进制 3E 帧请求报文包含以下部分:副头部(5000 表示请求)、网络号、PLC 号、请求目标模块 IO 号、请求目标模块站号、请求数据长度、CPU 监视定时器、指令代码、子指令代码、软元件地址、软元件点数。
软元件地址的表示方式是很多人第一次接触时最懵的地方。三菱的软元件用字母加数字表示,比如 D100 表示数据寄存器第 100 号,M50 表示内部继电器第 50 号,X10 表示输入继电器第 10 号。但在 MC 协议报文里,地址需要转换成十六进制,并且要区分位软元件和字软元件。D 寄存器是字软元件,地址直接就是数值;M、X、Y 是位软元件,地址需要按位计算。比如 D100 在报文里就是 0x0064,M50 是 0x0032,但读取位软元件时,返回的数据是按位打包的,每两个字节包含 16 个位的状态。
我刚开始做的时候,在地址换算上栽过跟头。当时读 D 寄存器没问题,读 M 点就总是错位。后来才发现,位软元件的地址在 MC 协议里是按位编号的,但报文里传的是起始位地址,返回的数据需要自己按位解析。比如读取 M0 开始的 16 个位,返回两个字节,第一个字节的 bit0 对应 M0,bit1 对应 M1,以此类推。如果读取的点数不是 16 的整数倍,最后一个字节的高位是无效的,需要自己屏蔽掉。
2.2 Python 通信库选型与连接管理
Python 和三菱 PLC 通信,可选的路子有几条。一是用pymcprotocol这个第三方库,它封装了 MC 协议的 3E 帧和 4E 帧,支持二进制和 ASCII,用起来比较省心。二是用python-snap7,但那个是西门子的 S7 协议,不适用三菱。三是自己用 socket 拼报文,灵活性最高,但开发量大。我一般推荐pymcprotocol,它的 API 设计比较直观,batchread_wordunits和batchread_bitunits两个方法就能覆盖大部分场景。
安装很简单,pip install pymcprotocol就行。连接的时候需要指定 PLC 的 IP 和端口,还有网络号、PLC 号这些参数。大部分情况下网络号和 PLC 号都是 0,站号也是 0。连接建立后,建议做一个心跳检测,定期读一个固定的寄存器,确认连接还活着。我见过太多因为网线松动或者交换机重启导致连接假死的情况,如果不做心跳,程序会一直卡在 recv 上,数据就断了。
连接管理上,我习惯用一个单独的类来封装,内部维护 socket 连接和重连逻辑。重连策略是:检测到异常后,先关闭旧连接,等待一个退避时间(比如 1 秒、2 秒、4 秒递增),然后尝试重新连接。连续失败超过一定次数(比如 10 次),就记录错误日志并告警。重连成功后,要把采集调度层里缓存的待读任务重新激活,避免数据出现大段空白。
注意:三菱 PLC 的 MC 协议连接数有限制,FX5U 一般支持 8 个以内的 TCP 连接,Q 系列多一些但也有限。如果多个程序同时连,可能会被拒绝。建议一个 PLC 只用一个采集程序,需要多路数据就在程序内部做多线程或者异步。
2.3 批量读取与单点读取的取舍
MC 协议支持批量读取,一次可以读连续的一段软元件。比如一次读 D100 到 D199,共 100 个字。批量读取的效率远高于单点读取,因为报文开销是固定的,读 1 个点和读 100 个点的报文长度差不多。我实测过,读 100 个 D 寄存器,批量读取耗时大约 5 到 8 毫秒,而单点读 100 次需要 300 毫秒以上。所以在设计采集方案时,一定要把需要读取的软元件按地址连续性分组,尽量批量读。
但批量读取也有坑。一是地址必须连续,如果需要的软元件分散在 D100、D200、D500,那就得分成三组。二是单次读取的点数不能超过 MC 协议的限制,3E 帧二进制模式下,一次最多读 960 个字或者 7168 个位。超过这个数就得拆分。三是有些 PLC 型号对批量读取的响应时间较长,如果一次读太多,可能会触发 CPU 监视定时器超时。我一般把单次批量读取控制在 200 个字以内,兼顾效率和稳定性。
对于位软元件,批量读取的返回数据是打包的,解析时需要按位展开。我写了一个辅助函数,把返回的字节数组转换成布尔列表,再根据软元件起始地址映射到具体的 M 点或者 X 点。这个函数看起来简单,但边界条件很多,比如点数不是 16 的整数倍时,最后一个字节的有效位要截断。
3. 数据入库时序设计的核心逻辑
3.1 采集周期与 PLC 扫描周期的匹配
时序设计的第一步,是确定采集周期。这个周期不能拍脑袋定,要和 PLC 的扫描周期匹配。PLC 的扫描周期是指 CPU 从读取输入、执行程序、刷新输出的一个循环时间,通常在 1 到 20 毫秒之间,取决于程序复杂度。如果采集周期比扫描周期还短,那读到的数据可能是同一个扫描周期的重复值,没有意义。如果采集周期太长,又可能漏掉快速变化的信号。
我的经验是:采集周期至少是 PLC 扫描周期的 5 到 10 倍。比如扫描周期是 10 毫秒,采集周期设为 50 到 100 毫秒比较合适。对于变化缓慢的温度、压力等模拟量,采集周期可以放宽到 500 毫秒甚至 1 秒。对于高速计数或者位置数据,可能需要 20 到 50 毫秒。但要注意,采集周期越短,网络和数据库的压力越大,需要权衡。
还有一个关键点是采集的触发方式。我一般用固定周期触发,而不是事件触发。固定周期的好处是数据的时间戳均匀,便于后续做趋势分析和报表。事件触发虽然能捕捉突变,但容易造成数据稀疏不均,入库和查询都麻烦。如果确实需要捕捉突变,可以在固定周期的基础上,增加一个变化检测逻辑,当某个关键值变化超过阈值时,额外记录一条。
3.2 时间戳的生成与对齐策略
时间戳是时序数据的灵魂。没有准确的时间戳,数据就是一堆无意义的数字。时间戳的生成有两种方式:一是用 Python 程序所在服务器的系统时间,二是用 PLC 内部的时钟。我强烈建议用服务器时间,因为 PLC 的时钟往往不准,而且不同 PLC 之间可能有偏差。服务器可以配置 NTP 同步,精度有保障。
但用服务器时间也有讲究。采集到数据的那一刻,和写入数据库的那一刻,时间可能差了几十毫秒甚至几秒。如果直接用写入时间做时间戳,数据的时序就会失真。我的做法是:在采集调度层触发读取之前,先记录一个时间戳,这个时间戳代表“本次采集的基准时间”。然后把这个时间戳附加到读取到的每一条数据上。这样即使入库有延迟,时间戳仍然是准确的。
对于批量读取的多个软元件,它们共享同一个时间戳,因为它们在同一个报文里返回,可以认为是同一时刻的值。如果对时间精度要求极高,可以在报文返回后立即记录时间,但差别通常在毫秒级,对大部分工业场景够用了。
时间戳的格式我一般用datetime对象,入库时转成数据库的DATETIME或者TIMESTAMP类型。如果数据量特别大,可以考虑用 Unix 时间戳的整数形式,节省存储空间,查询时再转换。MySQL 的TIMESTAMP精度默认到秒,需要毫秒精度的话要用DATETIME(3)或者TIMESTAMP(3)。
3.3 批量入库与事务控制
数据入库的频率和采集频率不一定一致。如果每采集一次就写一次数据库,数据库的压力会很大,尤其是采集周期在 100 毫秒以下的时候。我的做法是:采集层把数据放入一个内存队列,入库层从队列里批量取数据,攒够一定数量或者达到一定时间间隔后,一次性写入数据库。这个批量大小和时间间隔需要根据数据量和数据库性能来调。
批量入库的好处很明显:减少数据库连接开销、减少事务提交次数、提高写入吞吐量。我实测过,单条插入 MySQL,每秒大概能写 500 到 1000 条;批量插入 100 条一批,每秒能写 5000 到 10000 条。差距是数量级的。但批量也不能太大,否则一次事务占用太多内存和锁资源,反而影响数据库的并发性能。我一般把批量大小设在 100 到 500 条之间,时间间隔设在 1 到 5 秒之间。
事务控制上,我建议用自动提交关闭的方式,显式地commit。如果一批数据写入失败,可以回滚,避免部分写入导致数据不一致。但要注意,如果数据库连接断了,回滚可能失败,需要有补偿机制。我的做法是:入库失败时,把数据重新放回队列头部,等待下次重试。同时记录错误日志,如果连续失败超过阈值,就告警。
注意:MySQL 的
max_allowed_packet参数限制了单次 SQL 语句的大小。如果批量插入的数据量太大,可能会超过这个限制导致报错。建议在数据库配置里把这个参数调大,比如 16MB 或者 64MB。同时,innodb_buffer_pool_size也要根据服务器内存适当调大,提升写入性能。
4. 完整实操流程与关键代码解析
4.1 环境准备与依赖安装
先说一下环境。我用的 Python 版本是 3.9 到 3.11 之间,太老的版本有些库不支持,太新的版本有些库还没适配。操作系统 Linux 和 Windows 都用过,生产环境建议用 Linux,稳定性和资源占用都更好。依赖库主要有三个:pymcprotocol用于 MC 协议通信,pymysql或者mysql-connector-python用于 MySQL 连接,schedule或者直接用threading.Timer做周期调度。如果要用异步,可以上asyncio和aiomysql,但复杂度会高一些。
安装命令很简单:
pip install pymcprotocol pymysql如果网络环境不好,可以换国内源:
pip install pymcprotocol pymysql -i https://pypi.tuna.tsinghua.edu.cn/simple数据库这边,需要提前建好库和表。表结构我一般设计成宽表,每个软元件一列,加上一个时间戳主键。比如:
CREATE TABLE plc_data ( ts DATETIME(3) NOT NULL, d100 FLOAT, d101 FLOAT, d102 FLOAT, m50 TINYINT, m51 TINYINT, PRIMARY KEY (ts) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;如果软元件很多,宽表会有很多列,维护起来麻烦。另一种设计是窄表,每个软元件一行,用device_id和ts做联合主键。窄表更灵活,但查询时需要做行转列,性能差一些。我一般根据软元件数量来选:少于 50 个用宽表,多于 50 个用窄表加分表。
4.2 通信层代码实现
通信层的核心是封装pymcprotocol的Type3E类。下面是我常用的一个简化版本:
import time import logging from pymcprotocol import Type3E class PlcClient: def __init__(self, ip, port=5000): self.ip = ip self.port = port self.plc = None self.connect() def connect(self): try: self.plc = Type3E() self.plc.connect(self.ip, self.port) self.plc.setaccessopt(commtype="binary") logging.info(f"PLC connected: {self.ip}:{self.port}") except Exception as e: logging.error(f"PLC connect failed: {e}") self.plc = None def read_words(self, device, start, count): if self.plc is None: self.connect() if self.plc is None: return None try: return self.plc.batchread_wordunits(headdevice=device, readsize=count) except Exception as e: logging.error(f"Read words failed: {e}") self.plc.close() self.plc = None return None def read_bits(self, device, start, count): if self.plc is None: self.connect() if self.plc is None: return None try: return self.plc.batchread_bitunits(headdevice=device, readsize=count) except Exception as e: logging.error(f"Read bits failed: {e}") self.plc.close() self.plc = None return None这里有几个细节。setaccessopt(commtype="binary")是设置二进制模式,必须在连接后、读取前调用。batchread_wordunits的headdevice参数是字符串,比如"D100",readsize是点数。返回的是一个列表,每个元素是一个整数,对应一个寄存器的值。对于位软元件,batchread_bitunits返回的也是列表,每个元素是 0 或 1。
重连逻辑我放在read_words和read_bits里,一旦读取异常就关闭连接并置空,下次读取时会自动重连。这种懒重连的方式比较简单,但要注意重连频率,避免疯狂重试把 PLC 的连接数耗尽。可以在connect里加一个退避,比如失败后time.sleep(1)再试。
4.3 采集调度与队列管理
采集调度层我一般用一个独立的线程,按固定周期触发读取。读取到的数据放入queue.Queue,入库线程从队列里取。下面是一个简化的调度器:
import threading import queue import time from datetime import datetime class Collector(threading.Thread): def __init__(self, plc_client, data_queue, interval=0.1): super().__init__() self.plc_client = plc_client self.data_queue = data_queue self.interval = interval self.running = True def run(self): while self.running: ts = datetime.now() words = self.plc_client.read_words("D", 100, 10) bits = self.plc_client.read_bits("M", 50, 4) if words is not None and bits is not None: record = {"ts": ts, "words": words, "bits": bits} self.data_queue.put(record) time.sleep(self.interval) def stop(self): self.running = False这里interval是采集周期,单位秒。time.sleep的精度在 Windows 上可能只有 15 毫秒左右,Linux 上好一些。如果对周期精度要求高,可以用time.perf_counter做补偿,或者用asyncio的call_later。但大部分工业场景,几十毫秒的抖动是可以接受的。
队列的作用是解耦采集和入库。如果入库慢了,队列会积压,但采集不会停。但队列也不能无限积压,否则内存会爆。我一般给队列设一个最大长度,比如 10000 条,满了就丢弃最老的数据并告警。丢弃数据是下策,但比程序崩溃好。更好的做法是监控队列长度,超过阈值就告警,让人来排查入库为什么慢。
4.4 入库层批量写入实现
入库层从队列里取数据,攒够一批就写数据库。下面是一个简化的入库线程:
import pymysql import threading import time class DbWriter(threading.Thread): def __init__(self, data_queue, db_config, batch_size=200, flush_interval=2.0): super().__init__() self.data_queue = data_queue self.db_config = db_config self.batch_size = batch_size self.flush_interval = flush_interval self.running = True self.buffer = [] self.last_flush = time.time() def run(self): conn = pymysql.connect(**self.db_config) cursor = conn.cursor() while self.running: try: record = self.data_queue.get(timeout=0.5) self.buffer.append(record) except queue.Empty: pass if len(self.buffer) >= self.batch_size or (time.time() - self.last_flush) >= self.flush_interval: if self.buffer: self.flush(cursor, conn) if self.buffer: self.flush(cursor, conn) cursor.close() conn.close() def flush(self, cursor, conn): sql = "INSERT INTO plc_data (ts, d100, d101, d102, m50, m51) VALUES (%s, %s, %s, %s, %s, %s)" values = [] for r in self.buffer: ts = r["ts"] words = r["words"] bits = r["bits"] values.append((ts, words[0], words[1], words[2], bits[0], bits[1])) try: cursor.executemany(sql, values) conn.commit() except Exception as e: conn.rollback() logging.error(f"DB write failed: {e}") self.buffer.clear() self.last_flush = time.time()executemany是批量插入的关键,它会把多条INSERT合并成一个网络往返,效率比循环单条插入高得多。batch_size和flush_interval是两个调节旋钮,根据实际数据量和数据库性能来调。如果数据量很大,可以适当增大batch_size,但要注意max_allowed_packet的限制。
5. 常见问题与排查技巧实录
5.1 连接类问题速查
| 问题现象 | 可能原因 | 排查方法 | 解决方案 |
|---|---|---|---|
| 连接被拒绝 | IP 或端口错误 | ping PLC IP,telnet 端口 | 确认 PLC 的 MC 协议端口,默认 5000 |
| 连接超时 | 网络不通或防火墙拦截 | 检查网线、交换机、防火墙规则 | 放行端口,确认 PLC 的 IP 设置 |
| 连接数超限 | 多个程序同时连接 | 查看 PLC 连接数设置 | 合并采集程序,或增大 PLC 连接数 |
| 连接假死 | 网线松动或交换机重启 | 心跳检测读固定寄存器 | 加心跳,超时后重连 |
| 读取返回错误码 | 软元件地址错误或点数超限 | 检查地址格式和点数 | 修正地址,拆分批量读取 |
连接类问题里,最常见的是 IP 和端口搞错。三菱 PLC 的 MC 协议端口不一定是 5000,有些项目会改成 6000 或者别的。另外,PLC 的 IP 地址要和采集服务器在同一网段,或者路由可达。如果中间有防火墙,要放行对应端口。
5.2 数据异常类问题排查
数据异常通常表现为读到的值不对、值不变、或者跳变。值不对可能是地址错了,比如把 D100 写成了 D101。值不变可能是 PLC 程序没有在刷新那个寄存器,或者采集周期太短读到了缓存。跳变可能是数据类型解析错了,比如把有符号数当成无符号数,或者把 32 位浮点数的两个寄存器顺序搞反了。
三菱 PLC 的浮点数在 D 寄存器里占两个连续的字,低字在前还是高字在前,取决于 PLC 的参数设置。我遇到过读出来的浮点数完全不对,后来发现是高低字反了。解决方法是:先读两个寄存器,按两种顺序拼成 32 位整数,再转成浮点数,看哪个结果合理。这个坑很隐蔽,但一旦踩过就记住了。
注意:三菱 PLC 的 D 寄存器默认是 16 位有符号整数,范围 -32768 到 32767。如果要存大于 32767 的数,需要用两个寄存器拼成 32 位,或者用浮点数。读取时要注意符号位,Python 的
int是无符号的,需要自己处理负数。
5.3 入库性能与稳定性问题
入库慢是最常见的性能问题。原因可能是批量太小、索引太多、磁盘 IO 瓶颈、或者数据库连接池不够。我一般先看批量大小,如果每次只插几条,那肯定慢。然后看表索引,时序数据表的主键是时间戳,如果还有别的索引,写入时会额外维护,拖慢速度。建议时序表只保留必要的主键索引,查询用的索引可以后续再加。
另一个问题是数据重复。如果采集程序重启,可能会把之前已经入库的数据重新读一遍。解决方法是:在 PLC 里做一个递增的计数器,每次采集时读这个计数器,入库时用计数器做去重。或者用时间戳做唯一约束,重复插入时忽略。MySQL 的INSERT IGNORE或者ON DUPLICATE KEY UPDATE可以实现。
数据丢失也是要关注的。如果采集程序崩溃,队列里的数据就没了。我的做法是:队列用持久化的方式,比如写入本地文件或者 Redis,程序重启后从持久化存储里恢复。但这样会增加复杂度,对于大部分场景,只要保证程序稳定运行,偶尔丢几条数据是可以接受的。如果数据绝对不能丢,那就得上更重的方案,比如 Kafka 或者 RabbitMQ。
5.4 实操心得与避坑清单
- 心跳不能省:我见过太多因为没心跳导致连接假死、数据断了几小时才发现的情况。心跳间隔建议 5 到 10 秒,读一个固定的寄存器,超时 3 次就重连。
- 批量读取要分组:把地址连续的软元件放在一组,一次读回来。地址不连续的不要硬凑,分开读反而快。
- 时间戳在采集时生成:不要等到入库时才取时间,那样会有延迟。采集触发时立即取时间,附加到数据上。
- 队列要有上限:无限队列在入库故障时会吃光内存。设一个上限,比如 10000 条,满了就丢弃并告警。
- 数据库连接要复用:不要每次写入都新建连接,用连接池或者长连接。但长连接要注意空闲超时,定期做 ping。
- 日志要详细:记录每次采集的时间、点数、耗时,每次入库的批量大小、耗时、结果。出问题时,日志是唯一的线索。
- 测试要覆盖异常:拔网线、关 PLC、关数据库、重启程序,这些场景都要测一遍,确保程序能自动恢复。
6. 时序设计的进阶优化方向
6.1 多 PLC 并发采集的时序协调
一个项目里往往不止一台 PLC,可能有多台设备需要同时采集。如果每台 PLC 用一个独立的线程,线程数多了之后,GIL 和上下文切换会成为瓶颈。更好的方式是用asyncio做异步 IO,一个线程就能管理多个连接。但pymcprotocol是同步库,要配合asyncio需要自己封装或者用run_in_executor。另一种方式是每个 PLC 一个进程,进程间用消息队列通信,这样能充分利用多核 CPU。
多 PLC 采集时,时间戳的对齐很重要。如果各台 PLC 的采集时间不一致,后续做关联分析时会很麻烦。我的做法是:用一个全局的调度器,统一触发所有 PLC 的采集,尽量让它们在同一个时间窗口内完成。如果某台 PLC 响应慢,可以给它单独放宽周期,但时间戳仍然用触发时间。
6.2 数据压缩与冷热分离
时序数据的特点是量大、写入密集、查询集中在近期。如果全部存在 MySQL 里,时间长了表会非常大,查询和备份都成问题。我的做法是:热数据(最近 7 天)存 MySQL,冷数据(7 天以上)定期归档到文件或者专门的时序数据库。归档可以用 Python 脚本定时跑,把旧数据导出成 CSV 或者 Parquet,然后从 MySQL 删除。
如果数据量特别大,可以考虑用 TimescaleDB,它是 PostgreSQL 的时序扩展,支持自动分区和压缩。或者用 InfluxDB,专门为时序数据设计,写入和查询性能都很好。但引入新数据库会增加运维成本,要根据团队的技术栈来选。
6.3 采集程序的监控与告警
采集程序跑在生产环境,不能等出问题了才去查。我一般会加几个监控指标:采集成功率、平均采集耗时、队列长度、入库成功率、入库耗时。这些指标可以写到日志里,也可以用 Prometheus 的 Python 客户端暴露出来,配合 Grafana 做可视化。告警规则可以设:采集成功率低于 99%、队列长度超过 5000、入库连续失败 3 次,触发邮件或者钉钉告警。
监控的另一个作用是容量规划。通过观察队列长度和入库耗时的趋势,可以判断当前配置是否够用,什么时候需要扩容。比如队列长度持续增长,说明入库速度跟不上采集速度,需要优化入库或者降低采集频率。
6.4 断线重连后的数据补采
如果 PLC 断线了一段时间,重连后这段时间的数据就缺失了。对于大部分场景,缺失就缺失了,补不回来。但如果 PLC 里有历史数据缓存,比如用 D 寄存器做环形缓冲,那可以在重连后把缺失的数据读回来。这需要 PLC 程序配合,在采集程序里记录上次读取的位置,重连后从那个位置继续读。
另一种补采方式是:在数据库里记录采集的连续性,如果发现时间戳有跳跃,就标记为数据缺失,后续分析时注意。补采的实现复杂度较高,除非业务对数据完整性要求极高,否则不建议做。大部分工业场景,只要保证实时数据不丢,历史数据偶尔缺几秒是可以接受的。
7. 个人实操体会与建议
这套方案我在多个项目里用过,从单台 FX5U 到十几台 Q 系列组网,整体稳定性还是不错的。最关键的经验是:不要追求极致的采集频率,要根据实际需求来定。很多新手一上来就想 10 毫秒采一次,结果数据库扛不住,程序也不稳定。实际上,大部分工业过程的变化速度远没有那么快,100 毫秒甚至 500 毫秒的采集周期完全够用。
另一个体会是:异常处理比正常逻辑更重要。正常读取谁都会写,但断线、超时、数据异常这些情况,才是决定程序能不能长期稳定运行的关键。我现在的习惯是,每写一个功能,先想它可能怎么失败,然后把失败的处理逻辑写好。这样虽然开发慢一点,但上线后省心很多。
最后分享一个小技巧:如果 PLC 的 MC 协议读取总是超时,可以试试把setaccessopt里的commtype从binary改成ascii。虽然 ASCII 效率低一些,但有些老型号的 PLC 对二进制模式支持不好,换成 ASCII 反而稳定。这个是我在一个 FX3U 加 ENET 模块的项目里发现的,当时二进制模式读十次错三次,换成 ASCII 后一次都没错过。