简介这份资源是面向工业自动化与数控系统开发者的OPC UA客户端实例程序基于西门子官方提供的客户端源代码实现用于与SINUMERIK 840Dsl数控系统进行数据交互完成数据采集与远程监控。资源包共68个文件约1.29MB以C#源码文件18个.cs和PNG图片23个为主另含6个ico图标、6个resx资源文件、4个dll依赖库、2个csproj工程文件及sln解决方案、README说明等整体为可直接编译运行的Visual Studio工程结构。目前已有647人学习下载。通过该实例读者可以理解OPC UA的服务器、客户端、节点与信息模型等核心概念掌握订阅、发布、服务调用及身份验证与安全通信的实现方式并参考OpcUaHelper封装类完成自定义监控与控制应用。对于希望深入理解OPC UA协议与西门子840Dsl集成、需要可运行代码作为排错与二次开发起点的工程师这份资源具有较高的参考价值。1. opcua实例程序从一台设备到另一台设备中间到底发生了什么很多人第一次接触 opcua 实例程序是因为现场有一台设备要往上位机送数据或者两个不同厂商的系统要互通。翻了一圈资料发现 OPC UA 这个词反复出现但真到动手时官方文档给的是规范开源库给的是 API中间那段“到底怎么把两台机器连起来”的实例代码反而没人完整讲。我最初也踩过这个坑以为装个库、填个 IP 就能通结果证书报错、端点选错、命名空间对不上折腾了两天才跑通第一行数据。opcua 实例程序要解决的核心问题就一个让客户端能稳定读写服务端的节点值。它适合三类人——做设备数据采集的、做上位机集成的、做跨系统数据交换的。你不需要先啃完几百页规范但必须理解地址空间、会话、订阅这三件事否则写出来的代码只能跑在你自己电脑上。下面按“先跑通、再调参、后避坑”的顺序把一套可复现的实例程序拆开讲。2. 先分清服务端和客户端opcua 实例程序的最小骨架2.1 为什么实例程序一定要拆成两个进程常见做法是把服务端和客户端写在同一个脚本里启动一个线程跑服务端主线程跑客户端。这在演示时没问题但真实项目里必须拆开原因有三个第一服务端通常对应真实设备或网关生命周期比客户端长第二客户端可能同时连多个服务端混在一起会让连接管理变复杂第三证书和端点配置是分开的混写容易把安全策略搞乱。我一般会建两个文件server.py和client.py。服务端负责暴露变量节点客户端负责连接、读取、订阅。下面先给服务端最小实现用的是 Python 生态里比较常见的 asyncua 库其他语言思路一致。# server.py import asyncio from asyncua import Server async def main(): server Server() await server.init() # 端点必须和客户端一致端口冲突时改这里 server.set_endpoint(opc.tcp://0.0.0.0:4840/freeopcua/server/) # 命名空间用来隔离自定义节点避免和标准节点混在一起 uri http://examples.freeopcua.github.io idx await server.register_namespace(uri) # 创建对象节点相当于给变量分组 obj await server.nodes.objects.add_object(idx, MyDevice) # 添加两个变量温度、状态 temp await obj.add_variable(idx, Temperature, 25.0) status await obj.add_variable(idx, Status, True) # 允许客户端写入 await temp.set_writable() await status.set_writable() async with server: while True: await asyncio.sleep(1) if __name__ __main__: asyncio.run(main())这段代码的逻辑很直白初始化服务端、注册命名空间、挂两个变量、设置可写、然后保持运行。参数上最需要注意的是set_endpoint里的地址0.0.0.0表示监听所有网卡端口 4840 是 OPC UA 的常用端口但并非强制。命名空间 URI 不要用http://opcfoundation.org/UA/那是标准命名空间自定义节点必须用自己的 URI。2.2 客户端连接的三步端点、会话、节点客户端代码比服务端多一步“找节点”。很多人卡在这里是因为不知道节点 ID 长什么样。在 asyncua 里节点 ID 由命名空间索引和标识符组成比如ns2;i1。命名空间索引是服务端注册时动态分配的不是固定的所以客户端不能硬编码最好通过 URI 反查。# client.py import asyncio from asyncua import Client async def main(): # 端点必须和服务端完全一致包括路径 url opc.tcp://127.0.0.1:4840/freeopcua/server/ async with Client(urlurl) as client: # 通过 URI 反查命名空间索引避免硬编码 uri http://examples.freeopcua.github.io idx await client.get_namespace_index(uri) # 按路径找节点Objects - MyDevice - Temperature obj await client.nodes.objects.get_child([f{idx}:MyDevice]) temp await obj.get_child([f{idx}:Temperature]) status await obj.get_child([f{idx}:Status]) # 读取当前值 print(Temperature:, await temp.read_value()) print(Status:, await status.read_value()) # 写入一个新值 await temp.write_value(30.5) print(After write:, await temp.read_value()) if __name__ __main__: asyncio.run(main())这里的关键参数是get_child里的路径写法{idx}:MyDevice表示命名空间索引加浏览名。如果服务端改了变量名客户端必须同步改否则会抛BadNoMatch。另外read_value返回的是 Python 原生类型不是字符串直接打印就能看到数值。2.3 订阅模式为什么轮询读值在生产环境会翻车轮询读值写起来简单但设备一多、频率一高服务端和网络都扛不住。OPC UA 的订阅机制是让服务端在值变化时主动推给客户端客户端只需要处理回调。下面这段代码把温度变量订阅起来变化超过 0.5 才触发。# subscriber.py import asyncio from asyncua import Client async def main(): url opc.tcp://127.0.0.1:4840/freeopcua/server/ async with Client(urlurl) as client: uri http://examples.freeopcua.github.io idx await client.get_namespace_index(uri) obj await client.nodes.objects.get_child([f{idx}:MyDevice]) temp await obj.get_child([f{idx}:Temperature]) # 创建订阅周期 500ms sub await client.create_subscription(500, temp) # 数据变化回调 async def handler(node, val, data): print(Value changed:, val) await sub.subscribe_data_change(temp) # 保持运行等待推送 await asyncio.sleep(30) if __name__ __main__: asyncio.run(main())订阅的坑在于create_subscription的第二个参数是发布周期单位毫秒不是采样周期。采样周期在subscribe_data_change里可以单独设。如果服务端变量没有设置set_writable或者值没变回调不会触发这不是代码问题是机制如此。3. 把实例程序跑在真实设备上端点、证书、命名空间怎么配3.1 端点地址的三种写法和对应场景端点地址不是随便填的它决定了客户端能不能连上。常见写法有三种写法适用场景注意点opc.tcp://127.0.0.1:4840本机调试只能本机访问跨机必失败opc.tcp://0.0.0.0:4840服务端监听客户端不能填 0.0.0.0要填实际 IPopc.tcp://192.168.1.10:4840/xxx局域网设备路径必须和服务端注册时一致我见过最常见的翻车是服务端监听0.0.0.0客户端也填0.0.0.0结果连不上。0.0.0.0是监听地址不是连接地址。客户端必须填服务端所在机器的实际 IP 或主机名。3.2 证书报错时先关安全策略再逐步加回来OPC UA 默认要求安全通道但很多开源库的示例默认是None安全策略。如果你从示例切到真实设备第一件事就是确认安全策略。报错通常是BadSecurityChecksFailed或BadCertificateUntrusted。处理顺序我一般这样先把客户端和服务端都设成SecurityPolicyType.None确认能通再打开Basic256Sha256让双方交换证书最后把证书加入信任列表。不要一上来就开最高安全级别否则你连报错都看不懂。# 客户端指定安全策略 from asyncua import Client from asyncua.crypto.security_policies import SecurityPolicyBasic256Sha256 client Client(urlopc.tcp://192.168.1.10:4840/) await client.set_security( SecurityPolicyBasic256Sha256, certificateclient_cert.pem, private_keyclient_key.pem, server_certificateserver_cert.pem )证书文件格式必须是 PEM私钥不能带密码否则库会报解析失败。如果服务端用的是自签名证书客户端必须显式信任不能靠系统证书链。3.3 命名空间索引会变别在客户端硬编码命名空间索引是服务端启动时按注册顺序分配的今天ns2明天加了一个模块可能就变成ns3。硬编码ns2的客户端在服务端升级后必挂。正确做法是用 URI 反查前面客户端代码里已经用了get_namespace_index这里再强调一次任何生产代码都不应该出现写死的命名空间索引。如果服务端是第三方设备你拿不到 URI那就用浏览的方式找节点。先读Server_NamespaceArray把 URI 列表打出来再根据 URI 找索引。这个变量在标准节点里所有 OPC UA 服务端都有。4. 避坑opcua 实例程序最常见的五个翻车点4.1 连接超时但端口是通的现象telnet能通客户端连不上报TimeoutError。 原因服务端端点路径和客户端不一致或者服务端只监听了 IPv6。 解决检查set_endpoint的完整路径包括末尾斜杠把监听地址改成0.0.0.0强制 IPv4。4.2 读到的值一直是 None现象节点找到了read_value返回None。 原因变量没有初始值或者服务端还没启动数据更新循环。 解决在add_variable时给一个初始值确认服务端async with块已经进入。4.3 写入报 BadUserAccessDenied现象读正常写报权限错误。 原因变量没有调用set_writable()。 解决服务端对每个需要写的变量显式设置可写不要指望默认值。4.4 订阅回调不触发现象订阅创建成功但值变化时没有输出。 原因发布周期设得太长或者变量值变化幅度没超过死区。 解决把发布周期降到 500ms 以内检查subscribe_data_change是否传了死区参数。4.5 多客户端同时写同一个变量现象两个客户端交替写值跳来跳去。 原因OPC UA 不提供跨客户端的写锁。 解决在应用层加互斥逻辑或者让服务端只接受一个客户端的写请求。5. 进阶用 opcua 实例程序做数据采集时我固定会加的三个习惯第一个习惯是给每个节点加时间戳。OPC UA 的read_value只返回值不返回时间。要拿时间戳得用read_data_value里面包含SourceTimestamp和ServerTimestamp。做历史数据存储时这两个时间戳必须一起存否则事后对不上。dv await temp.read_data_value() print(dv.Value.Value, dv.SourceTimestamp, dv.ServerTimestamp)第二个习惯是给订阅加队列。默认回调是同步处理的如果处理逻辑慢会丢数据。用sub.subscribe_data_change时可以在回调里把数据扔进asyncio.Queue另起一个协程慢慢消费。这样即使数据库写入卡顿也不会影响订阅接收。第三个习惯是每次改完服务端节点先跑一遍浏览脚本。浏览脚本不需要复杂逻辑就是把Objects下的所有子节点递归打印出来确认命名空间、浏览名、节点 ID 都对得上。这个脚本我用了很多年每次换设备或升级服务端先跑它能省掉大量猜节点的时间。async def browse(node, depth0): for child in await node.get_children(): print( * depth, await child.read_browse_name()) await browse(child, depth 1)最后说一个我自己的教训早期做实例程序时总想把所有配置写在一个文件里结果换一台设备就要改十几处。后来改成配置文件加环境变量端点、命名空间 URI、证书路径全部外置代码只负责逻辑。这个习惯让我在后续对接不同厂商设备时基本不用改代码只改配置就能跑。希望帮到你。本文还有配套的精品资源点击获取