简介这份实例源码以OPC UA通信协议为核心面向需要实现PLC与PC机数据交换的工业自动化开发者也适合正在学习OPC UA技术的入门者目标是解决现场设备与上位机之间的互联互通问题。压缩包共285个文件大小仅926KB其中C源文件111个、头文件47个、Python脚本21个另有CMake构建配置、Markdown/RST说明文档、Shell脚本与截图便于直接阅读和编译。代码涵盖服务器端的信息模型构建、节点与属性定义、连接监听以及客户端的连接、浏览、读写、订阅和身份验证、数据加密等安全机制同时涉及TCP/IP传输、二进制编解码与发布/订阅等关键实现完整展示了OPC UA通信链路的各个技术环节。已有265人学习下载该资源通过改编其中示例可将OPC UA快速集成到实际PLC上位机项目中缩短开发周期也能加深对OPC UA工作机制的理解。1. OPC UA 不是把网线插上就完事PLC 与 PC 机通信这件事为什么值得重新做一遍做 PLC 与 PC 机通信老路数是 IP 直连加私有协议或者 Modbus 轮询。前两年还能对付等到设备一多、数据要往上送、甲方要求记录每一次写入操作的时候地址表维护就成了灾难。OPC UA 解决的是这个问题它给设备侧的数据做统一建模PC 端按节点去读不管对面是西门子、汇川还是数控机床访问方式一致语义也不靠人记。这个标题里的“实例源码”本质上是把最常用的三条路径写给你节点浏览、数据订阅、写入控制。适合正在做数据采集、上位机监控或虚拟调试的工程师。新手照着代码能把 S7-1200/1500 连起来读数熟手看订阅参数、安全策略和节点映射的边界才知道项目怎么不翻车。2. 先看穿黑匣子节点模型、订阅机制和方法调用决定你的实例源码该怎么写OPC UA 和 Modbus 最大的区别不在速度而在数据组织方式。Modbus 给你一张地址表OPC UA 给你一棵树。你代码里写的每一个路径都是在这棵树上定位节点。如果没理解这棵树的规则就算拿到源码换一台 PLC、换一个程序版本节点 ID 变了你就抓瞎。2.1 地址空间与节点 IDPLC 里的一个 DB 块怎么变成 PC 端的完整路径OPC UA 的地址空间从 Objects 节点开始往下挂。设备、数据块、变量、方法都在这棵树上。每个节点有唯一的 NodeId常见两种格式一种是ns3;i1001数字索引另一种是ns3;sDataBlock1.Value1字符串路径。实际项目里字符串格式好用因为可读性强出错能直接看出来。和西门子 PLC 对接时S7-1200/1500 内建的 OPC UA 服务器会把 DB 块导出成树形节点。注意一个关键选项DB 块的访问方式。优化访问的 DB 块导出的节点路径是符号名比如DataBlock1.Temperature非优化访问的 DB 块在 OPC UA 里只有绝对地址节点比如ns3;sDB1.DBX0.0。我一般强烈建议用优化访问虽然 TIA 里新建 DB 默认就是优化但老项目移植过来经常是非优化。NodeId 格式对照表格式示例含义适用场景ns3;i1001数字节点 ID某些 PLC 厂家只导出数字索引ns3;sDB1.Value字符串节点 ID西门子优化 DB、支持符号名导出ns0;i85OPC UA 标准节点服务端自带的对象和类型定义理解这一点之后你拿到实例源码时第一件事不是跑起来而是先用浏览工具看目标 PLC 的地址空间长什么样。后面第 4 章给的浏览代码就是干这个用的。2.2 订阅机制采样间隔和发布间隔是两回事别只调一个OPC UA 数据订阅有两个时间参数混为一谈是新手最常犯的错。采样间隔 SamplingInterval 是服务端去数据源取值的周期发布间隔 PublishingInterval 是服务端把一包数据发给客户端的周期。你可以让服务端每 10ms 采样一次但每 500ms 才把数据发给你。反过来也可以采样很慢、发布很快那客户端只会反复收到相同的值。对 PLC 数据采集我常用的配置是采样间隔 100ms、发布间隔 200ms或者采样 100ms、发布 500ms。不要追求过低因为 PLC 的 OPC UA 服务器预算是有限的采样间隔压到 10msCPU 负荷明显上升服务端响应变慢客户端反而更容易触发超时。订阅队列也有讲究。每个订阅有一个队列队列长度 QueueSize 决定服务端缓存多少包。常见的坑是客户端处理慢队列满了然后旧数据被丢弃你看到的数值会突然跳变。如果你读的是温度、压力这种连续量跳一下问题不大如果是计件数、位置信号丢一帧可能导致逻辑判断错位。所以要么把队列调大要么写消费代码时保证处理速度。事件和数据的区别也在这里。数据订阅说的是变量值变化事件订阅说的是服务端主动通知。PLC 侧报警、急停信号这类适合做成事件连续变化的工艺参数适合走数据订阅。实例源码里如果两类都涉及务必分开建两个订阅不要共用一个。2.3 方法调用与事件PC 端主动触发 PLC 程序靠的不是写变量很多人想实现“PC 端按一个按钮PLC 跑一段程序”。最容易想到的方式是写一个 bool 变量PC 置 TruePLC 程序里检测上升沿。这能做但不干净你要定义哪个位是触发、哪个位是完成、还要处理复位时序。OPC UA 的方法节点是更可靠的做法。PLC 项目里把某段逻辑封装成一个方法导出后客户端直接调用服务端执行完返回结果状态清晰而且所有调用都有记录可查。不用再管理临时变量的生命周期。方法调用的坑主要在两个地方一是 PLC 侧方法封装时参数类型必须明确比如输入是 INT、输出是 REAL客户端传错类型服务端直接拒绝二是方法执行时间。OPC UA 方法调用是同步等待的如果 PLC 里那个程序块运行超过几十毫秒客户端设置的回调超时就可能被触发。长任务一定要异步化或者拆成“启动”和“查询状态”两个方法。实例源码里涉及到方法调用我一般建议先把最简单的无参方法跑通再逐步加参数。3. 从 PLC 配置到 PC 端 SDK跑通 OPC UA 通信的最小架构原理说完了开始落地。一个能跑的 OPC UA 通信实例由三部分组成PLC 内建的 OPC UA 服务器、PC 端的 SDK 客户端、中间的以太网链路。很多人第一步就错了——在 PLC 上电之后没有主动开启服务器功能。3.1 PLC 侧准备西门子 S7-1200/1500 开启 OPC UA 服务器西门子 S7-1200 从较新固件开始内建 OPC UA 服务器S7-1500 全系支持但默认不开启。在 TIA Portal 里需要做三层配置第一在设备组态的“OPC UA”选项里勾选“启用 OPC UA 服务器”。第二设置安全策略。调试期可以直接选“无安全策略”跑通了再换 Basic256Sha256 签名加密。第三给服务器设置用户名和口令客户端连接时必须带身份验证。不设口令也能连但工业现场不允许留这种口子。还有两个容易被忽略的设置一是“端口号”默认 4840不要改改了 PC 端连接字符串也要跟着改二是“最大连接数”默认值通常够用但如果你同时有 HMI、仿真软件和 PC 客户端三个连接最好确认一下当前值。配置完成后在 TIA 的在线诊断里能看到 OPC UA 服务器是否处于运行状态。还有一个验证方法用 PC 上的 UaExpert 直接连 PLC 的 IP 加 4840 端口能浏览节点就是通了。这一步不要跳过省得后面分不清是 PLC 没配好还是 PC 代码有问题。国产 PLC 的情况不同。汇川 H5U、AM 系列等新平台的 PLC 不少也集成了 OPC UA 服务器配置项一般叫“OPC UA 服务器使能”或“开放网关注册”选完型号后记得下载到 PLC。老一些的日系 PLC 不带内建服务器常见做法是加一个 OPC UA 网关盒子把私有协议翻译过来PC 端代码写法完全一样只是连接地址从 PLC IP 变成网关 IP。另外如果你在做仿真联调Tecnomatix Process Simulate 这类工具可以通过 OPC UA 直接订阅虚拟 PLC 的变量配置思路和真实 PLC 一致这也是个很好的练手环境。3.2 PC 端 SDK 怎么选open62541、python-asyncua 与 UA.NETStandard 的取舍PC 端选 SDK我的判断标准是场景是快速验证还是长时间运行的正式上位机。快速验证和开发调试我用 Python 的 asyncua 库它覆盖了浏览、订阅、读写、方法调用代码量最少十分钟能跑通。正式上位机是 Windows 平台的 .NET 环境UA.NETStandard 是主流选择性能稳定和 WinForm/WPF 界面集成顺手。open62541 是 C 语言库适合做嵌入式网关或者你自己要写一个轻量级服务器端。如果团队里有 Rust 背景rust-opcua 也值得关注内存安全和并发模型在长时间运行的采集服务上有优势但生态不如前两者成熟。选型对比表SDK语言适合场景学习成本asyncuaPython原型验证、数据采集脚本、快速联调低UA.NETStandardC#Windows 上位机、组态软件中open62541C网关、嵌入式设备、自建服务端高rust-opcuaRust高并发采集服务、长时间运行中高我一般建议先拿 asyncua 把通信链路验证完确认节点路径、数据类型、订阅参数都对了再决定正式程序用什么写。很多项目甲方要求源码就是 Python 脚本那就直接用 asyncua 做成常驻服务配 systemd 或 Windows 计划任务管生命周期。3.3 最小连接例程3 秒钟连上 PLC 并读一个变量先装依赖pip install asyncua然后是最小的连接和读取代码import asyncio from asyncua import Client async def read_value(): # url 格式opc.tcp://PLC的IP:端口 # 端口默认 4840调试期用 None 安全策略 client Client(opc.tcp://192.168.0.10:4840) # 连接超时设长一点PLC 首次响应可能超过默认值 client.session_timeout 30000 try: await client.connect() # 读取 PLC 中 DB 块变量的节点 node client.get_node(ns3;sDB_Data.Temperature) value await node.read_value() print(fTemperature: {value}) finally: await client.disconnect() asyncio.run(read_value())连接字符串里的 IP 换成你 PLC 的实际地址。client.session_timeout 30000这个参数很关键默认值偏小PLC 处理慢一点就直接断连。节点路径先用 2.1 里说的浏览工具确认不要凭记忆写否则报BadNodeIdUnknown。还有一点连接建立了不代表身份验证过了。如果 PLC 侧设置了用户名口令需要再加一行client.set_user(opcua_user) client.set_password(your_password)这个例子跑通说明链路没问题。接下来才是真正的实战——把实例源码里的浏览、订阅、写入拆开来看。4. 实例源码拆解节点浏览、数据订阅与写命令的完整实现拿到一份 OPC UA 通信实例源码先别着急改 IP 就跑。源码的骨架通常是四条线连接管理、节点浏览、订阅采集、写入控制。其中连接管理是基础浏览是工具订阅是常态写入是动作。我从实际项目里按这个顺序拆给你看。4.1 先浏览地址空间把真实节点 ID 换成代码里的路径很多人直接去网上抄一个节点路径跑不通就说 OPC UA 坑多。其实大多数情况下是节点路径不对。不同 PLC 品牌、不同固件版本、不同项目配置导出的地址空间都不一样。所以第一步永远是浏览。下面这段代码把地址空间从根节点开始全部打印出来from asyncua import Client, Node async def browse_recursive(node: Node, depth: int 0): if depth 5: # 限制深度防止打印过于深层的系统节点 return children await node.get_children() for child in children: name await child.read_browse_name() node_id child.nodeid print( * depth f{name.Name} - {node_id}) await browse_recursive(child, depth 1) async def browse_all(): client Client(opc.tcp://192.168.0.10:4840) await client.connect() root client.get_root_node() # 地址空间的根 await browse_recursive(root) await client.disconnect()depth限制是有意义的OPC UA 服务器会导出大量的系统类型节点不限制会刷屏。运行后你会看到打印结果里有Objects、DeviceSet、PLC_1、DataBlocks这样的树形结构。找到你关心的数据块把那一行显示的 NodeId 记下来。比如NodeId(ns3, sDB_Data.Temperature)这就是 3.3 代码里get_node的参数格式。不会有哪两个项目地址空间完全一样。所以这个浏览代码是整套源码里的基础工具每次换现场、换 PLC 程序版本先运行一遍确认节点没变再往下走。4.2 数据订阅把 PLC 变量变成 PC 端的事件流读一次值只是验证链路。真正的数据采集是订阅PLC 侧数值一变化服务端主动推送PC 端不用一直发请求。asyncua 里订阅的写法from asyncua import Client import asyncio class SubHandler: def datachange_notification(self, node, value, data): # 每次数据变化触发回调 print(f{node} - {value.value}) async def subscribe_values(): client Client(opc.tcp://192.168.0.10:4840) await client.connect() # 创建订阅发布间隔 200ms subscription await client.create_subscription(200, SubHandler()) # 订阅两个变量 node1 client.get_node(ns3;sDB_Data.Temperature) node2 client.get_node(ns3;sDB_Data.Pressure) await subscription.subscribe_data_change(node1) await subscription.subscribe_data_change(node2) await asyncio.sleep(60) # 让订阅保持运行 60 秒 await subscription.delete() await client.disconnect()create_subscription(200, SubHandler())的第一个参数是发布间隔 200ms第二个参数是回调对象。datachange_notification在值发生变化时被调用注意传入的value是 DataValue 对象要用value.value拿实际数值。这个实例里值得学的不是 API 用法而是参数匹配。如果你把发布间隔设成 1000msPLC 那边数值变了以后最坏情况要 1 秒才能到 PC 端这在监控画面里能看出来延迟。反过来发布间隔设成 50ms服务端压力陡增。工业现场的推荐值需要平滑曲线的场合用 100ms 到 200ms只要状态量报警量的场合用 500ms 到 1s 就足够。订阅回调里绝对不要干重活。我在正式项目里看到过回调里写数据库、发消息队列、刷新界面结果回调阻塞订阅队列溢出数据开始丢。回调里只做赋值和计数器真正的数据处理放到独立线程。4.3 写入与控制通过节点写入触发 PLC 动作订阅解决的是数据上行写入解决的是控制下行。写入在 OPC UA 里是对节点执行write_value操作但比 Modbus 写寄存器多了类型校验。from asyncua import Client from asyncua.common.variant import Variant, VariantType async def write_plc(): client Client(opc.tcp://192.168.0.10:4840) await client.connect() node client.get_node(ns3;sDB_Control.TargetSpeed) # 写入一个 float 值PLC 侧该变量类型必须是 REAL val Variant(1500.0, VariantType.Float) status_code await node.set_value(val) if status_code.is_bad(): print(f写入失败: {status_code}) else: print(写入成功) await client.disconnect()重点在VariantType.Float。OPC UA 要求写入的数据类型必须和节点定义的类型严格一致。PLC 侧变量是 REAL你就得传 Float是 INT你就得传 Int16。很多人习惯直接传一个 Python 的 intasyncua 在某些版本里会自动推断类型推断错了就报BadTypeMismatch。实例源码里我见过最省事的做法是封装一层async def safe_write(client, node_id, value, var_type): node client.get_node(node_id) return await node.set_value(Variant(value, var_type))这样调用时显式传类型哪里错了点哪里。尤其是布尔量PLC 里的 BOOL 对应VariantType.Boolean不要传成整数。写入前先读一下节点当前类型data_type await node.read_data_type() print(data_type) # 拿到的是 VariantType 枚举这是一道后悔药写入失败的时候能省半小时排查时间。4.4 连接管理断线重连、会话超时和日志输出跑现场的人都有过这种经历通信程序跑了一晚上第二天早上发现数据停了。不是 PLC 挂了是 OPC UA 会话超时断开了而你的程序没有重连逻辑。OPC UA 的会话有生命周期。客户端和服务器之间有心跳超过会话超时时间没有消息交互服务器释放会话。asyncua 里有session_timeoutPLC 侧也有自己的超时配置。重连的常规做法是捕获ConnectionError和UaServiceError退避重试。import asyncio, logging from asyncua import Client logging.basicConfig(levellogging.INFO) logger logging.getLogger(opcua_reconnect) async def run_with_reconnect(): url opc.tcp://192.168.0.10:4840 retry_delay 5 # 初始重试间隔 5 秒 while True: client Client(url) client.session_timeout 30000 try: await client.connect() logger.info(OPC UA 连接成功) # 在这里启动你的订阅逻辑或读取循环 await asyncio.sleep(10) retry_delay 5 # 连接成功后复位重试间隔 except Exception as e: logger.error(f连接异常: {e}) await asyncio.sleep(retry_delay) retry_delay min(retry_delay * 2, 60) # 指数退避最大 60 秒 finally: await client.disconnect() asyncio.run(run_with_reconnect())重连间隔用指数退避避免 PLC 还没恢复时客户端疯狂重连占满连接数。日志是必须加的输出至少记录连接成功、连接失败、重连次数这几个事件。现场排查通信问题时没有日志全靠猜有日志一眼看到底是网络断了、PLC 重启了还是会话超时。5. PLC 与 PC 联调的 5 个常见坑从连不上到数据错位的排查顺序把网上各种实例源码里最常出问题的点位汇总一下。这五条是我实际项目里踩过最多遍的也是和其他工程师交流时出现频率最高的。每条按现象、原因、解决写。5.1 现象客户端连接超时报 BadConnectionClosed 或干脆无响应原因是多层的。最常见的是 PLC 侧 OPC UA 服务器没启用或是启用了但 PLC 程序没有下载到设备。还有就是 PC 与 PLC 不在同一网段。西门子 PLC 默认地址常见是 192.168.0.1你 PC 如果是 192.168.1.x连不通很正常。解决顺序写死先在 PC 上pingPLC IP确认链路通然后用 UaExpert 试连如果 UaExpert 能连说明服务器正常问题在 SDK 代码如果 UaExpert 也连不上去 TIA 里检查 OPC UA 服务器勾选和下载状态同时看 Windows 防火墙是否放行了 4840 端口。这是最快的二分定位法。5.2 现象连接报 BadSecurityModeRejected原因只有一个客户端请求的安全策略和服务端配置不一致。PLC 侧设置的是 Basic256Sha256你代码里用的 None必然被拒。解决调试期先把 PLC 侧安全策略改为 None代码里也保持默认不加密两边拉齐之后再把安全策略一项一项升级。正式项目里从 None 切到 Basic256Sha256 时需要在服务端配置信任的客户端证书这个流程各厂家差异很大有的 PLC 是把 CA 证书导入到设备里有的是首次连接时弹窗确认。不管哪种都要在文档里记录证书指纹别等半年后证书过期了才来找你。5.3 现象节点能浏览但数值读出来是错的错得还很有规律比如显示的温度一直是 0或者压力值比现场仪表大一倍。这种往往是数据类型转换问题。OPC UA 服务端传回的是 Int16你按 Int32 解释符号位就错了传回的是 Float你按 Double 解析数值就乱。解决在读取代码里先打印node.read_data_type()确认服务端的真实类型然后显式用 VariantType 包装。还有一个隐蔽点PLC 里有些变量是缩放后的值程序里写的是 0 到 100 的百分比内部实际存的是 0.0 到 1.0 的 REAL。你看到数值和现场表不一致不是通信的问题是量纲没换算。这个靠代码排查不出来得看 PLC 程序。5.4 现象订阅不推送或者推送极其慢先把两个时间概念摆清楚发布间隔是客户端创建订阅时指定的采样间隔是服务端读取数据源的周期。如果你创建订阅时给的发布间隔是 100ms但 PLC 服务端最低采样间隔是 500ms那你最快还是 500ms 收到一次值。西门子很多服务端会对采样间隔做下限限制你设了 10ms 它也会钳到 100ms 以上。解决把采样间隔参数设为 -1表示由服务端自行决定。然后从 100ms 开始慢慢调发布间隔观察回调频率变化。如果始终没有推送检查你有没有await subscription.subscribe_data_change(node)——创建订阅之后不订阅变量订阅就是空壳。5.5 现象程序跑得好好的PLC 一重新下载就全断了PLC 程序更新后OPC UA 服务器会重新初始化节点路径可能不变但节点 ID 的命名空间索引可能变化。尤其是有多个硬件组态的项目ns 索引经常跳。你的代码写死了ns3下载前是对的下载后就坏。解决用浏览代码导出一份节点清单每次 PLC 程序变更之后重新跑一遍对比节点路径和 ns 索引变化。我习惯的做法是启动时先浏览一遍把路径映射成新 ns 之后缓存起来而不是把ns3写死在配置里。这样 PLC 固件升级或者程序重新下载后代码不需要改。6. 进阶验证用 Wireshark 抓 OPC UA 报文确认通信不是“能连就行”能连上、能读数、能订阅这只是第一步。要确认通信模型是真的在按你设的参数工作得看报文不能只看应用层的打印结果。6.1 抓 OpenSecureChannel 与 ReadResponse 的时序在 PC 上用 Wireshark选择通信用的网卡过滤条件填tcp.port 4840。跑通 3.3 的最小连接代码你能看到完整的握手过程先是 HEL 消息客户端声明缓冲区大小服务端回 ACK然后是 OpenSecureChannel建立安全通道再往后是 ReadRequest 和 ReadResponse 成对出现。如果抓包只看到了 HEL 和 ACK没有 OpenSecureChannel说明安全策略协商失败回到 5.2 去检查。如果 ReadRequest 有但 ReadResponse 没有可能是服务端处理超时看 PLC 的 CPU 负荷和通信资源占用。如果用的是订阅而不是周期读你会注意到数据包不是均匀出现的而是值变化的时刻才来一包。发布间隔理论上决定了批量推送的节奏抓包能直接验证你设的 200ms 发布间隔是否真的生效。6.2 批量读写把单点采集升级成整包传输OPC UA 支持一次请求读写多个节点。批量读是降低通信开销的最有效手段尤其是几十个变量同时要刷新的场景。async def batch_read_nodes(client, node_ids): nodes [client.get_node(nid) for nid in node_ids] values await client.read_values(nodes) return dict(zip(node_ids, values)) node_ids [ ns3;sDB_Data.Temperature, ns3;sDB_Data.Pressure, ns3;sDB_Data.Speed, ns3;sDB_Data.Torque, ] values await batch_read_nodes(client, node_ids) print(values)read_values一次调用请求服务端返回所有节点当前值。实例源码里如果把逐点read_value改成这种批量接口通信周期可以从几百毫秒降到几十毫秒。代价是响应包变大但普通 48 字节的包装几十个浮点数绰绰有余。6.3 一个收尾建议:给每个节点建立一个“数据字典”我做 OPC UA 项目到现在最大的教训就是节点路径永远不写死在调用代码里而是集中到一个 JSON 配置或者 Python 字典里。每个现场、每台设备都有自己的节点清单运行前加载配置启动时全量浏览校验。这样换设备不用改代码只改配置。PLC_NODES { temperature: ns3;sDB_Data.Temperature, pressure: ns3;sDB_Data.Pressure, speed_setpoint: ns3;sDB_Control.TargetSpeed, machine_status: ns3;sDB_Control.MachineState, }配这个字典前一定先用第四章的浏览代码确认路径不要凭记忆填。通信代码写得好不好不在于用了多少高级 API而在于现场出问题时你能不能在五分钟内定位到是网络、配置、还是数据语义的问题。这比多接一百个节点都值钱。希望这些经验能帮你少踩几个坑。本文还有配套的精品资源点击获取
