简介这份资源是面向工业自动化开发者与西门子PLC工程师的OPC UA客户端源码包用于与S7-1200、S7-1500、S7-300、S7-400等系列PLC进行安全的数据交换。源码围绕连接管理、节点读写、数据订阅、错误处理及二进制编解码等核心环节展开适合已具备一定C#基础、希望深入理解OPC UA通信机制并二次开发上位机或集成自动化系统的技术人员。压缩包共72个文件约1.06MB以29个cs源码文件为主体辅以19个png界面截图、10个resx资源文件、5个dll依赖库及3个csproj工程文件另含sln解决方案与可执行程序结构完整便于直接编译调试。目前已有3385人学习下载。通过研读源码读者可掌握客户端与S7 PLC建立连接、读写变量、订阅实时数据及异常处理的实现思路并据此扩展为更复杂的工业自动化应用。1. 西门子OPC UA 客户端源码从协议栈到产线数据落地的完整路径产线数据采不上来十有八九卡在协议这一层。PLC 侧数据明明在刷新上位机就是读不到或者读到了却对不上号——这种场景做工业数据采集的人都不陌生。西门子 OPC UA 客户端源码这个方向解决的就是让上位机、边缘网关或数据平台以标准 OPC UA 协议稳定读写西门子 S7-1200/1500 及部分 S7-300/400 系列 PLC 的变量。它适合三类人做产线数据采集的自动化工程师、需要把 PLC 数据接入 MES 或数据库的软件开发者、以及想自研轻量采集网关替代商业组件的团队。源码在手意味着你可以改超时、改重连策略、改订阅周期而不是被商业组件的黑匣子行为卡住。2. 先搞清楚 OPC UA 客户端在西门子体系里到底连什么2.1 西门子 PLC 侧的服务端角色与端点差异很多人第一次接触会混淆OPC UA 客户端连的不是 PLC 本身而是 PLC 里运行的 OPC UA 服务端。S7-1500 从固件 V2.0 起内置 OPC UA 服务端S7-1200 从固件 V4.4 起也支持但两者能力有差异。S7-1500 支持订阅Subscription和监控项MonitoredItem可以做到变化上报S7-1200 早期固件只支持读写不支持订阅只能轮询。这个差异直接决定你的客户端架构如果目标设备是 S7-1200 且固件偏低源码里就必须实现轮询调度器而不是依赖订阅回调。端点Endpoint是另一个容易翻车的点。PLC 的 OPC UA 服务端默认监听 4840 端口但西门子允许配置多个端点安全策略也不同。常见的有 None、Sign、SignAndEncrypt 三种。产线内网调试阶段很多人图省事用 None但一旦上生产环境安全策略不匹配会直接导致连接被拒。源码里必须把端点发现GetEndpoints和策略协商做进去不能写死。# 使用 asyncua 库发现西门子 PLC 的 OPC UA 端点 from asyncua import Client async def discover_endpoints(url): client Client(urlurl) # 不传安全策略先拿端点列表 endpoints await client.connect_and_get_server_endpoints() for ep in endpoints: print(fEndpoint: {ep.EndpointUrl}) print(f SecurityMode: {ep.SecurityMode}) print(f SecurityPolicy: {ep.SecurityPolicyUri}) print(f UserToken: {[t.TokenType for t in ep.UserIdentityTokens]}) await client.disconnect()这段代码的逻辑是先不协商安全策略直接向服务端请求端点列表把每个端点支持的 SecurityMode、SecurityPolicyUri 和用户令牌类型打印出来。参数上url 填opc.tcp://PLC_IP:4840如果 PLC 配了多个端点这里会全部列出。拿到列表后再决定用哪个端点、配哪种策略。注意connect_and_get_server_endpoints不会建立正式会话只是拿元数据所以即使策略不匹配也能执行。2.2 节点 ID 的三种写法与西门子变量映射OPC UA 的节点 IDNodeId是寻址核心。西门子 PLC 里的变量映射到 OPC UA 地址空间后节点 ID 通常长这样ns3;sDB1.Temperature。其中 ns 是命名空间索引西门子一般把用户变量放在 ns3 或 ns4具体取决于 PLC 配置。s 表示字符串标识符后面跟的是符号名。也有用数字标识符的比如ns3;i1001但西门子默认用符号名可读性好但解析慢。源码里处理节点 ID 时最常见的坑是命名空间索引写死。不同 PLC、不同项目ns 索引可能不同。正确做法是先读服务端的 NamespaceArray找到对应命名空间的索引再拼节点 ID。另一个坑是符号名里的引号和点号在字符串拼接时容易出错建议用库提供的 NodeId 构造方法而不是手拼字符串。# 动态解析命名空间索引避免写死 ns3 async def resolve_node(client, namespace_uri, symbol_name): # 读取服务端命名空间数组 ns_array await client.get_namespace_array() try: ns_index ns_array.index(namespace_uri) except ValueError: raise RuntimeError(f命名空间 {namespace_uri} 不存在) # 用 NodeId 构造避免手拼字符串 from asyncua import ua node_id ua.NodeId(symbol_name, ns_index) return client.get_node(node_id)逻辑说明先拿服务端的命名空间数组用 URI 反查索引再用ua.NodeId构造节点。参数上namespace_uri 一般是http://www.siemens.com/s7-1500或类似具体看 PLC 配置symbol_name 是带引号的符号名比如DB1.Temperature。这样写的好处是换一台 PLC 只要 URI 不变代码不用改。如果 URI 也变了那就得改配置但至少不会因为 ns 索引变化而静默读错变量。3. 用源码跑通第一个读写会话连接、读、写、断3.1 最小连接与读取单个变量的完整代码先把最小闭环跑通再谈订阅和批量。下面这段代码用 Python 的 asyncua 库连接西门子 PLC读一个变量写一个变量然后断开。选 asyncua 是因为它纯 Python、跨平台、源码可读适合做二次开发。如果你用 C#OPC Foundation 的官方栈更合适但源码量大改起来门槛高。import asyncio from asyncua import Client async def main(): url opc.tcp://192.168.0.10:4840 client Client(urlurl) # 如果 PLC 开了匿名登录不设用户否则设用户名密码 # client.set_user(operator) # client.set_password(password) try: await client.connect() print(连接成功) # 读变量 node client.get_node(ns3;sDB1.Temperature) value await node.read_value() print(fTemperature {value}) # 写变量 setpoint client.get_node(ns3;sDB1.Setpoint) await setpoint.write_value(25.5) print(写入完成) finally: await client.disconnect() asyncio.run(main())逻辑上connect()会完成端点发现、策略协商、会话建立。get_node()只是构造节点对象不发起网络请求read_value()才真正读。write_value()同理。参数上url 里的 IP 换成你的 PLC 地址节点 ID 换成实际变量。如果 PLC 开了安全策略Client构造时要传security_policy和security_mode还要加载证书。匿名登录只适合调试生产环境必须配用户。3.2 批量读取与写入的性能取舍单点读写延迟在 10 到 50 毫秒量级取决于网络和 PLC 负载。如果采 100 个变量逐个读就是 1 到 5 秒产线节拍根本等不起。OPC UA 提供了 ReadRequest 批量读一次请求可以带多个节点。asyncua 里用read_values传节点列表。# 批量读取一次请求拿多个变量 nodes [ client.get_node(ns3;sDB1.Temperature), client.get_node(ns3;sDB1.Pressure), client.get_node(ns3;sDB1.Flow), ] values await client.read_values(nodes) for n, v in zip(nodes, values): print(f{n} {v})批量读的关键参数是单次请求的节点数上限。西门子 PLC 对单次请求的节点数有限制常见是 100 到 500 个超了会返回错误。稳妥做法是分片每片 50 到 100 个。写入同理用write_values批量写但要注意写入顺序和 PLC 扫描周期的关系连续写同一 DB 块的不同变量可能被 PLC 程序覆盖建议一次写一个逻辑组。3.3 会话保持与断线重连的基本策略产线网络不会永远稳定。客户端必须处理断线重连否则一次网络抖动就丢数据。asyncua 的 Client 有connect和disconnect但没有自动重连。源码里要自己包一层重试循环。async def connect_with_retry(url, max_retries5, backoff2): for attempt in range(max_retries): try: client Client(urlurl) await client.connect() return client except Exception as e: wait backoff ** attempt print(f连接失败 {e}{wait} 秒后重试) await asyncio.sleep(wait) raise RuntimeError(重试次数用尽)参数上max_retries 建议 5 到 10backoff 用 2 的指数退避避免频繁重连打爆 PLC。重连后要重新建立订阅因为旧订阅在会话断开时已经失效。如果源码里用了订阅重连逻辑里必须包含订阅重建否则会出现“连上了但没数据”的玄学问题。4. 订阅与监控项让数据变化主动上报而不是轮询4.1 创建订阅与监控项的参数怎么设订阅是 OPC UA 相比 Modbus 轮询的最大优势。客户端向服务端注册一个订阅服务端按发布周期检查监控项有变化就推送。核心参数有三个PublishingInterval发布周期、SamplingInterval采样周期、QueueSize队列深度。PublishingInterval 是服务端检查变化的频率SamplingInterval 是服务端从 PLC 读值的频率。通常 SamplingInterval 小于等于 PublishingInterval。# 创建订阅并添加监控项 subscription await client.create_subscription(period500, handlerhandler) node client.get_node(ns3;sDB1.Temperature) handle await subscription.subscribe_data_change(node)period 是 PublishingInterval单位毫秒500 表示每 500 毫秒检查一次。handler 是回调对象收到通知时触发。subscribe_data_change默认 SamplingInterval 等于 PublishingIntervalQueueSize 默认 1。如果变量变化很快QueueSize 要调大否则会丢中间值。但 QueueSize 太大会占 PLC 内存一般 10 到 100 够用。4.2 回调处理与数据落库的线程安全回调是在 asyncio 事件循环里执行的如果回调里做耗时操作比如写数据库会阻塞整个循环导致后续通知延迟。正确做法是回调里只把数据放进队列另起消费者协程处理。import asyncio from collections import deque data_queue asyncio.Queue(maxsize10000) class SubHandler: def datachange_notification(self, node, val, data): # 只入队不做耗时操作 try: data_queue.put_nowait((str(node), val)) except asyncio.QueueFull: print(队列满丢弃一条) async def consumer(): while True: node_str, val await data_queue.get() # 这里做落库或转发 print(f落库: {node_str} {val})参数上队列 maxsize 根据数据量和落库速度调太小会丢太大吃内存。消费者协程可以批量落库比如攒 100 条或每 1 秒写一次减少数据库压力。注意回调里不要抛异常抛了会被库吞掉排查起来很痛苦。4.3 订阅失效与重建的触发条件订阅不是一劳永逸的。会话断开、PLC 重启、网络切换都会导致订阅失效。源码里要监听订阅状态或者定期检查subscription是否还活着。asyncua 的订阅对象有dead属性但更可靠的做法是设一个心跳监控项如果超过 N 个周期没收到任何通知就主动重建订阅。async def monitor_subscription(subscription, timeout10): last asyncio.get_event_loop().time() while True: await asyncio.sleep(1) now asyncio.get_event_loop().time() if now - last timeout: print(订阅疑似失效重建) await subscription.delete() # 重新创建订阅和监控项 return # 收到通知时更新 last这里简化处理实际源码里last 要在回调里更新。timeout 设成 PublishingInterval 的 3 到 5 倍。重建订阅时要先删旧的再建新的避免服务端资源泄漏。5. 避坑与排查西门子 OPC UA 客户端最常见的五个翻车点5.1 连接被拒但 ping 得通现象TCP 能通4840 端口也开着但connect()报 BadSecurityPolicyRejected 或 BadCertificateUntrusted。原因PLC 端配了 SignAndEncrypt客户端用 None 去连或者客户端证书没导入 PLC 信任列表。解决先用端点发现拿到支持的策略客户端配对应策略和证书。西门子 PLC 的证书管理在 TIA Portal 的 OPC UA 设置里把客户端证书导入信任列表。5.2 读到的值全是 0 或不变现象连接成功读值不报错但值一直是 0 或初始值。原因节点 ID 写错了命名空间索引读到了同名的空节点或者变量在 PLC 里没使能 OPC UA 访问。解决用 UaExpert 之类的工具先确认节点能读到正确值再对比源码里的节点 ID。PLC 侧要勾选变量的 OPC UA 可见性。5.3 订阅收不到通知现象订阅创建成功但回调一直不触发。原因SamplingInterval 设得比 PLC 扫描周期还短服务端实际按扫描周期采样但 PublishingInterval 太长或者变量变化幅度没超过死区。解决把 PublishingInterval 调到 100 到 500 毫秒检查监控项的死区设置默认死区是 0但有些库会设默认值。5.4 批量读超时或返回部分错误现象一次读 200 个节点超时或部分节点返回 BadNodeIdUnknown。原因单次请求节点数超限或者其中某个节点 ID 无效导致整个请求失败。解决分片到 50 个一批先校验节点 ID 有效性再批量读。无效节点单独处理不要让它拖垮整批。5.5 长时间运行后内存涨现象客户端跑几天后内存持续上涨。原因订阅回调里创建的对象没释放或者断线重连时旧订阅没删。解决回调里避免创建大对象重连逻辑里先delete()旧订阅再建新的。用tracemalloc或objgraph定位泄漏点常见的是 handler 被全局引用。6. 进阶把客户端做成可配置的采集服务6.1 配置文件驱动的节点映射与采集策略硬编码节点 ID 的源码只能跑一个项目。要复用得把节点映射、采集周期、数据类型做成配置。常见做法是 YAML 或 JSON每个变量一条记录包含节点 ID、别名、数据类型、是否订阅、死区。# config.yaml plc: url: opc.tcp://192.168.0.10:4840 security: None variables: - node: ns3;sDB1.Temperature alias: temp_01 type: float subscribe: true deadband: 0.1 - node: ns3;sDB1.Pressure alias: press_01 type: float subscribe: false poll_interval: 1000源码启动时读配置按 subscribe 字段决定走订阅还是轮询。deadband 传给监控项减少无效通知。poll_interval 用于轮询组。这样换项目只改配置不动代码。6.2 数据缓冲与断线续传的简单实现断线期间数据不能丢。简单做法是本地开一个环形缓冲或 SQLite断线时写入本地重连后补传。环形缓冲适合内存够的场景SQLite 适合要持久化的场景。import sqlite3 def buffer_write(alias, value, ts): conn sqlite3.connect(buffer.db) conn.execute(INSERT INTO buffer VALUES (?,?,?), (alias, value, ts)) conn.commit() conn.close() def flush_buffer(send_func): conn sqlite3.connect(buffer.db) rows conn.execute(SELECT * FROM buffer ORDER BY ts).fetchall() for row in rows: send_func(row[0], row[1], row[2]) conn.execute(DELETE FROM buffer WHERE alias? AND ts?, (row[0], row[2])) conn.commit() conn.close()参数上buffer.db 放本地磁盘flush 在重连后触发。注意并发写要加锁或者用 WAL 模式。补传时按时间排序避免乱序。6.3 用日志和指标验证采集质量采集服务不能只看“有没有数据”要看“数据对不对、全不全”。源码里加几个关键指标连接状态、订阅通知数、读写失败数、队列积压数。用 logging 模块输出到文件或者暴露 Prometheus 指标。import logging from prometheus_client import Counter, Gauge notify_count Counter(opcua_notify_total, 订阅通知总数) fail_count Counter(opcua_fail_total, 读写失败总数) queue_size Gauge(opcua_queue_size, 当前队列积压) logging.basicConfig( filenameopcua_client.log, levellogging.INFO, format%(asctime)s %(levelname)s %(message)s )验证时先看 notify_count 是否随产线运行增长再看 fail_count 是否为零或偶发。queue_size 持续上涨说明消费者跟不上要调批量大小或落库频率。日志里记录每次重连和订阅重建方便回溯。我自己的习惯是任何采集服务上线前先让它空跑 24 小时只看日志和指标不接业务。这 24 小时能暴露 80% 的稳定性问题。希望帮到你。本文还有配套的精品资源点击获取
