Python多线程读写OPC DA：解决单线程断连与延迟的工程实践

简介这份资源面向工业自动化、物联网与系统集成方向的 Python 开发者聚焦 OPC DA 协议的多线程读写实践帮助读者打通 Python 与 PLC 等工业控制设备之间的数据通道。包内共 6 个文件以 4 个 py 脚本为主体辅以 1 个 dll 动态库和 1 个 md 说明文档压缩包约 90KB体积轻巧便于快速部署。脚本覆盖 OPC 客户端封装、主程序入口以及 MQTT 数据桥接等模块配合 OPCDAAuto.dll 完成 COM 组件注册与服务器连接并借助 threading 模块实现多数据点的并发读写提升采集响应效率。已有 1542 人学习下载适合希望掌握 OPC DA 通信、MQTT 数据转发与工业数据远程管理的中级开发者参考可作为自动化与工业 4.0 场景下的实用脚本模板。1. Python 多线程读写 OPC DA为什么单线程采集总在半夜断连凌晨三点产线数据突然断流第二天查日志发现采集进程还活着但 OPC DA 的读句柄已经僵死——这是我第一次用 Python 单线程轮询 OPC DA 时踩的坑。OPC DA 是基于 COM/DCOM 的老一代工业数据交换规范服务端多为组态软件或 PLC 网关客户端通过 COM 接口读写 Item。Python 本身不原生支持 COM得靠 pywin32 这类桥接库。问题在于OPC DA 的同步读写是阻塞的一个 Item 卡住整个采集循环就跟着卡。多线程不是炫技是把「读」和「写」拆到不同线程避免写操作阻塞读循环同时用线程池控制并发粒度。这篇面向已经能用 Python 连上 OPC DA、但被断连和延迟折磨的工程师从线程模型选型讲到句柄复用和异常恢复每一步都有可复现的代码和参数说明。2. 线程模型选型为什么我最终放弃 ThreadPoolExecutor 裸跑2.1 OPC DA 的 COM 单元模型与 Python 线程的冲突OPC DA 客户端底层是 COMCOM 有套间Apartment概念。Python 的pythoncom默认在主线程初始化的是 STA单线程套间如果你在子线程里直接调win32com.client.Dispatch大概率报CoInitialize has not been called或者拿到一个跨套间失效的接口指针。常见做法是在每个工作线程入口显式调用pythoncom.CoInitialize()退出时pythoncom.CoUninitialize()。但这里有个反直觉的点OPC DA 服务端通常注册为 MTA多线程套间或 Neutral客户端如果也用 MTA跨线程调用反而更稳。我一般会在采集线程里用pythoncom.CoInitializeEx(pythoncom.COINIT_MULTITHREADED)让线程进入 MTA这样多个读线程共享同一个服务端连接时不会因为套间切换产生额外封送开销。import pythoncom import win32com.client import threading class OpcDaWorker(threading.Thread): def __init__(self, prog_id, item_ids, result_queue): super().__init__(daemonTrue) self.prog_id prog_id # 例如 OPC.SimaticNET 或组态软件暴露的 ProgID self.item_ids item_ids # 本线程负责的 Item 列表 self.result_queue result_queue self._stop_event threading.Event() def run(self): # 关键每个线程独立初始化 COM且指定 MTA pythoncom.CoInitializeEx(pythoncom.COINIT_MULTITHREADED) try: server win32com.client.Dispatch(self.prog_id) # 获取 OPC 组对象组是读写操作的容器 group server.OPCGroups.Add(worker_group) group.UpdateRate 500 # 毫秒服务端刷新周期 group.IsActive True items group.OPCItems.AddItems(self.item_ids) while not self._stop_event.is_set(): # 同步读返回 (值, 质量, 时间戳) 的变体数组 values items.Read(1) # 1 表示从缓存读2 表示从设备读 self.result_queue.put((self.item_ids, values)) self._stop_event.wait(0.5) finally: pythoncom.CoUninitialize() def stop(self): self._stop_event.set()这段代码的逻辑说明CoInitializeEx的COINIT_MULTITHREADED参数让线程进入 MTA避免 STA 下的消息泵依赖OPCGroups.Add创建的组是线程内私有的不要跨线程共享组对象Read(1)的 source 参数 1 是缓存读2 是设备读缓存读快但可能拿到旧值设备读慢但实时。参数上UpdateRate设 500ms 是折中设太小服务端压力大设太大缓存值滞后。_stop_event.wait(0.5)代替time.sleep是为了 stop 时能立刻退出而不是等满一个周期。2.2 读线程与写线程的职责分离读和写如果放在同一个线程里顺序执行写操作的阻塞会直接拖慢读周期。我的做法是读线程只负责按固定周期拉数据写线程从队列取指令执行。写线程同样要独立CoInitializeEx并且写操作要用Write而不是Read注意 OPC DA 的写是同步的服务端确认后才返回。写队列用queue.Queue设一个最大长度满了就丢弃最旧的写指令——产线场景里过期的写指令比不写更危险。import queue write_queue queue.Queue(maxsize100) class OpcDaWriter(threading.Thread): def __init__(self, prog_id, item_map): super().__init__(daemonTrue) self.prog_id prog_id self.item_map item_map # {item_id: 服务端Item对象} self._stop_event threading.Event() def run(self): pythoncom.CoInitializeEx(pythoncom.COINIT_MULTITHREADED) try: server win32com.client.Dispatch(self.prog_id) group server.OPCGroups.Add(write_group) group.IsActive True items group.OPCItems.AddItems(list(self.item_map.keys())) while not self._stop_event.is_set(): try: item_id, value write_queue.get(timeout0.5) except queue.Empty: continue # 写操作第二个参数是值同步等待服务端返回 items.Write(value) finally: pythoncom.CoUninitialize() def stop(self): self._stop_event.set()逻辑说明写线程单独建组避免和读组抢服务端资源write_queue.get(timeout0.5)让线程在空闲时也能响应 stopitems.Write是同步的如果服务端无响应会阻塞所以实际部署时我会给写操作加一个看门狗线程超时未返回就重建连接。参数上maxsize100是经验值产线写指令频率一般低于 10Hz100 的缓冲足够应对短时突发。2.3 线程数量与 Item 分组的权衡不是线程越多越好。OPC DA 服务端对并发连接数有限制很多组态软件默认只允许 2 到 4 个客户端连接。我的做法是按 Item 数量分组每组 50 到 200 个 Item每组一个读线程。Item 太少线程多连接开销大Item 太多单线程轮询周期长。下面这个分组函数按 Item 的更新需求分高频 Item 单独一组低频 Item 合并。def group_items(item_configs, high_freq_threshold1000): item_configs: [(item_id, update_rate_ms), ...] 返回: [[item_id, ...], ...] 分组后的列表 high, low [], [] for item_id, rate in item_configs: if rate high_freq_threshold: high.append(item_id) else: low.append(item_id) groups [] # 高频组每组最多50个保证轮询周期 for i in range(0, len(high), 50): groups.append(high[i:i50]) # 低频组每组最多200个 for i in range(0, len(low), 200): groups.append(low[i:i200]) return groups参数说明high_freq_threshold1000表示更新周期小于等于 1 秒的算高频高频组每组 50 个按每个 Item 读耗时 2ms 估算一轮 100ms能满足 1 秒周期低频组每组 200 个一轮 400ms满足 5 秒以上周期。这个估算基于我实测的本地 OPC DA 服务端实际要按你的服务端响应时间调整。3. 句柄复用与连接恢复让采集进程活过一周3.1 为什么每次读写都 Dispatch 会拖垮服务端新手常犯的错是每次读都Dispatch一次或者每个 Item 单独AddItems。OPC DA 的Dispatch会建立新的 COM 连接服务端要为每个连接维护会话频繁创建销毁会导致服务端句柄泄漏表现就是跑几小时后新连接被拒绝。正确做法是每个线程只Dispatch一次组和 Item 句柄在线程生命周期内复用。如果 Item 列表动态变化用OPCItems.AddItems增量添加而不是重建组。# 错误示范每次读都重新连接 def bad_read(prog_id, item_id): server win32com.client.Dispatch(prog_id) # 每次都新建连接 group server.OPCGroups.Add(g) item group.OPCItems.AddItem(item_id) return item.Read(1) # 正确示范连接和句柄复用 class OpcDaSession: def __init__(self, prog_id): self.server win32com.client.Dispatch(prog_id) self.group self.server.OPCGroups.Add(persistent_group) self.group.IsActive True self.items None def ensure_items(self, item_ids): if self.items is None: self.items self.group.OPCItems.AddItems(item_ids) # 增量添加逻辑略核心是不重建组逻辑说明OpcDaSession把连接和组句柄封装在对象里线程内复用ensure_items只在首次调用时AddItems后续增量处理。参数上persistent_group的名字要固定方便服务端识别同一组。3.2 断连检测与自动重连的实现OPC DA 的断连不会抛异常而是读返回的质量位变成BAD或COMM_FAILURE。所以不能靠 try/except 判断断连要检查返回值的质量码。我一般每读 10 次检查一次质量连续 3 次 BAD 就触发重连。重连不是简单重新 Dispatch要先释放旧组和连接再重建否则服务端可能还占着旧会话。def read_with_reconnect(session, item_ids, max_bad3): bad_count 0 while True: try: values session.items.Read(1) # values 是二维数组每行 [值, 质量, 时间戳] quality values[0][1] if isinstance(values, tuple) else values[1] if quality 192: # OPC 质量码192 以下为 Bad bad_count 1 if bad_count max_bad: session.reconnect() bad_count 0 else: bad_count 0 return values except pythoncom.com_error as e: # COM 层异常直接重连 session.reconnect() return None参数说明OPC 质量码 192 是OPC_QUALITY_GOOD的阈值低于 192 包括BAD和UNCERTAINmax_bad3是经验值设 1 太敏感网络抖动就重连设 5 以上断连发现太慢。reconnect方法要先group.Remove再server.Disconnect然后重新走初始化流程。3.3 用看门狗线程兜底写操作超时写操作是同步的服务端不响应就永久阻塞。看门狗线程的思路是写线程每次写之前记录时间戳看门狗每 2 秒检查一次如果某个写操作超过 5 秒没完成就强制重建写线程的连接。Python 没法安全 kill 线程所以看门狗只负责标记和重建旧线程让它自己超时退出。class WriteWatchdog(threading.Thread): def __init__(self, writer, timeout5.0): super().__init__(daemonTrue) self.writer writer self.timeout timeout self.last_write_ts time.time() def run(self): while True: time.sleep(2) if time.time() - self.last_write_ts self.timeout: # 标记写线程需要重建实际重建由主控逻辑执行 self.writer.need_reconnect True self.last_write_ts time.time()逻辑说明看门狗不直接操作 COM只做时间检查need_reconnect标志由主控线程读取后执行重建。参数上timeout5.0是产线经验值一般写操作 100ms 内返回5 秒没返回基本是服务端挂了。4. 避坑与排查多线程 OPC DA 最常见的 5 个翻车现场4.1 现象子线程报 CoInitialize has not been called原因Python 主线程初始化了 COM但子线程没有独立初始化。COM 的套间状态是线程局部的主线程的初始化不会传递给子线程。解决每个工作线程入口调pythoncom.CoInitializeEx出口调CoUninitialize且不要在主线程和子线程之间传递 COM 接口指针。4.2 现象读到的值一直是旧值质量码却是 Good原因Read(1)是缓存读服务端按UpdateRate刷新缓存如果UpdateRate设得比实际变化慢缓存值就滞后。解决对实时性要求高的 Item 用Read(2)设备读或者把UpdateRate调到和采集周期匹配。注意设备读会显著增加服务端负载不要所有 Item 都用。4.3 现象跑几小时后服务端拒绝新连接原因每次读写都Dispatch导致连接句柄泄漏服务端会话数达到上限。解决线程内复用连接和组句柄Item 列表变化时用增量AddItems不要重建组。定期用server.OPCGroups.Count检查组数量正常应该等于线程数。4.4 现象写操作偶尔卡死整个采集进程无响应原因OPC DA 写是同步的服务端处理慢或网络抖动时写线程阻塞如果读线程和写线程共享同一个 COM 连接读也被拖死。解决读写线程分离各自独立连接写操作加看门狗超时重建。不要用threading.Lock去保护 COM 调用锁解决不了服务端阻塞。4.5 现象多线程下数据错乱Item 值张冠李戴原因多个线程共享同一个OPCItems对象Read返回的数组顺序和AddItems的顺序不一定一致尤其是增量添加后。解决每个线程维护自己的 Item 列表和返回值的映射关系不要跨线程共享 Item 对象。用item_id做 key 而不是靠数组下标。5. 进阶用异步回调替代轮询把采集周期压到 100ms 以内轮询的瓶颈在于每个 Item 的读都要走一次 COM 调用Item 多了周期就上不去。OPC DA 支持IOPCDataCallback回调服务端数据变化时主动推给客户端。Python 里用win32com.client.WithEvents可以挂接回调但要注意回调是在 COM 的套间线程里执行的不能在里面做耗时操作否则会阻塞服务端推送。import win32com.client class OpcDaCallback: def OnDataChange(self, transaction_id, group_handle, quality, timestamp, values): # 回调里只做入队不做处理 for item_id, value in zip(self.item_ids, values): self.result_queue.put((item_id, value, quality)) def setup_async_read(prog_id, item_ids, result_queue): server win32com.client.Dispatch(prog_id) group server.OPCGroups.Add(async_group) group.UpdateRate 100 # 服务端推送周期100ms group.IsActive True items group.OPCItems.AddItems(item_ids) callback OpcDaCallback() callback.item_ids item_ids callback.result_queue result_queue # 挂接回调WithEvents 会创建事件接收器 win32com.client.WithEvents(group, callback) # 建立异步订阅 group.OPCItems.SetEnable(True) return server, group逻辑说明WithEvents把OpcDaCallback的方法绑定到组的事件上OnDataChange是 OPC DA 规范定义的回调方法名不能改回调里只put到队列实际处理由消费线程做。参数上UpdateRate100是服务端推送周期设太小服务端 CPU 高设太大实时性差。注意异步模式下Read仍然可用但不要混用否则回调会重复触发。验证异步是否生效可以对比轮询和回调的周期轮询 200 个 Item 一般 400ms 以上回调可以稳定在 100ms 到 150ms。如果回调不触发检查group.IsActive是否为 True以及SetEnable是否调用。我踩过的坑是回调方法名拼错比如写成OnDataChange少了个 e事件接收器静默失败没有任何报错查了半天。最后说个习惯我每次部署多线程 OPC DA 采集都会先跑一个 24 小时的空载测试只读不写观察内存和句柄数是否稳定。句柄数每小时增长超过 10 个基本就是连接泄漏别急着上产线。希望帮到你。本文还有配套的精品资源点击获取