简介这份资源面向工业自动化、物联网及企业级数据集成方向的C#开发者提供一套可直接参考的OPC UA客户端实现方案解决客户端连接服务器、读写数据并落地存储到SQL Server的完整链路问题。项目基于OpcUaHelper开源库简化协议调用通过NuGet即可引入相关类与方法数据经下划线分隔的字符串格式解析后写入SQL Server涵盖建表、插入及查询更新等操作适合具备一定C#基础、希望切入工业通信场景的中级开发者。资源包共144个文件以56个dll依赖库、54个xml配置文档、9个cs源码文件为主另含config、resx、exe等运行与资源文件整体约5.51MB源码与依赖结构完整。目前已有3240人学习下载可帮助读者快速理解OPC UA客户端与数据库对接的工程组织方式并作为二次开发的起点。1. 从零手搓 OPC UA 客户端为什么我劝你先搞懂会话与订阅再碰 SQL Server工业现场的数据采集十有八九绕不开 OPC UA。但很多人第一次上手 C# 写客户端时习惯性地把它当成一个“高级版 WebClient”——连上、读一把、写库完事。结果到了产线上要么是断线重连后订阅失效要么是写入 SQL Server 时被高频数据冲垮连接池。我见过一个模拟项目采集 200 个点位每秒变化一次开发同学直接用SqlConnection在回调里同步插入跑了不到十分钟数据库就开始报超时。问题不在 SQL Server也不在 OPC UA而在于没搞清楚这个协议是“会话制 订阅制”的。这份资源要解决的正是从 C# 客户端建立安全会话、创建订阅、到把变化数据可靠落库的完整链路。适合已经能写 C# 控制台程序、但没怎么碰过工业协议和时序写入的开发者。下面按我实际拆包的顺序把选型、代码和踩过的坑一次讲透。2. 环境准备与 NuGet 选型别在包版本上浪费一整天2.1 为什么是 OPCFoundation.NetStandard.Opc.UaC# 生态里能用的 OPC UA 库不多常见做法是直接用 OPC Foundation 官方那套OPCFoundation.NetStandard.Opc.Ua。它分了好几个包别一股脑全装。核心是这三个包名作用是否必须OPCFoundation.NetStandard.Opc.Ua.Core基础类型、编码、传输是OPCFoundation.NetStandard.Opc.Ua.ClientSession、Subscription、MonitoredItem是OPCFoundation.NetStandard.Opc.Ua.Configuration应用配置、证书管理是OPCFoundation.NetStandard.Opc.Ua.Server自己搭服务端才用否版本上我一般锁定在 1.4.372.106 附近。太新的版本有时会改ApplicationConfiguration的加载方式网上搜到的示例对不上容易翻车。用dotnet add package时显式带版本号别让 NuGet 自己解析到最新。dotnet add package OPCFoundation.NetStandard.Opc.Ua.Core --version 1.4.372.106 dotnet add package OPCFoundation.NetStandard.Opc.Ua.Client --version 1.4.372.106 dotnet add package OPCFoundation.NetStandard.Opc.Ua.Configuration --version 1.4.372.106这三行命令分别拉取核心库、客户端库和配置库。版本号必须一致否则运行时会报程序集加载失败。如果你用的是 .NET Framework 而不是 .NET Core包名会变成Opc.Ua.Core等别混用。2.2 应用证书与安全策略开发阶段可以偷懒但别偷错OPC UA 强制要求应用实例证书。很多新手在这一步卡住是因为服务端要求SignAndEncrypt而客户端用了None连上去直接被拒。开发阶段我一般先用None策略跑通逻辑但正式环境必须换成SignAndEncrypt并信任服务端证书。// 构建 ApplicationConfiguration 时安全策略先设为 None var config new ApplicationConfiguration() { ApplicationName MyOpcClient, ApplicationType ApplicationType.Client, SecurityConfiguration new SecurityConfiguration { ApplicationCertificate new CertificateIdentifier { StoreType Directory, StorePath OPC Foundation/CertificateStores/MachineDefault, SubjectName MyOpcClient }, TrustedPeerCertificates new CertificateTrustList { StoreType Directory, StorePath OPC Foundation/CertificateStores/TrustedPeers }, AutoAcceptUntrustedCertificates true // 仅开发环境 }, ClientConfiguration new ClientConfiguration { DefaultSessionTimeout 60000 } }; await config.Validate(ApplicationType.Client);这段代码做了三件事指定证书存储位置、设置信任列表、把AutoAcceptUntrustedCertificates打开。最后这个参数是开发阶段的后悔药正式环境一定要关掉否则中间人攻击风险很高。DefaultSessionTimeout设 60 秒意思是会话超过 60 秒没活动就超时后面做重连时要参考这个值。3. 建立会话与订阅把数据变化变成事件3.1 创建 Session 并绑定订阅会话是 OPC UA 的“连接”订阅是“数据通道”。很多人只建会话不建订阅然后轮询读取这在点位少的时候没问题点位一多就是灾难。正确做法是建好会话后立刻创建订阅把要采集的节点加进去。// 1. 用配置创建端点 var endpointDescription CoreClientUtils.SelectEndpoint(opc.tcp://127.0.0.1:4840, useSecurity: false); var endpointConfiguration EndpointConfiguration.Create(config); var endpoint new ConfiguredEndpoint(null, endpointDescription, endpointConfiguration); // 2. 建立会话 var session await Session.Create( config, endpoint, false, MyOpcClient, 60000, new UserIdentity(new AnonymousIdentityToken()), null ); // 3. 创建订阅发布间隔 1000ms var subscription new Subscription(session.DefaultSubscription) { PublishingInterval 1000, KeepAliveCount 10, LifetimeCount 20, MaxNotificationsPerPublish 1000, PublishingEnabled true }; session.AddSubscription(subscription); await subscription.Create();PublishingInterval是服务端向客户端推送的间隔单位毫秒。KeepAliveCount是没数据变化时发心跳的周期数10 表示 10 个发布周期没变化就发一次心跳。LifetimeCount是订阅存活周期超过这个数没收到发布就认为订阅失效。这三个参数要配合好否则会出现订阅莫名消失的情况。3.2 添加 MonitoredItem 并处理通知MonitoredItem 就是你要采集的点位。每个点位可以单独设采样间隔和死区别用默认值一把梭。// 创建监控项指定 NodeId 和采样间隔 var item new MonitoredItem(subscription.DefaultItem) { DisplayName Temperature1, StartNodeId new NodeId(ns2;sMachine.Temperature1), SamplingInterval 500, QueueSize 10, DiscardOldest true }; // 绑定通知回调 item.Notification (monitoredItem, args) { foreach (var value in args.Notification.Value) { Console.WriteLine(${monitoredItem.DisplayName}: {value.Value} {value.SourceTimestamp}); // 这里把数据推入队列不要直接写库 _dataQueue.Enqueue(new TagData { TagName monitoredItem.DisplayName, Value Convert.ToDouble(value.Value), Timestamp value.SourceTimestamp }); } }; subscription.AddItem(item); await subscription.ApplyChanges();SamplingInterval是服务端采集该节点的间隔QueueSize是客户端缓存多少个未处理通知。DiscardOldest true表示队列满时丢最旧的数据这在数据突发时很有用。回调里我特意没有直接写 SQL Server而是塞进ConcurrentQueue原因下一章讲。4. 数据落库 SQL Server批量写入与连接池调优4.1 为什么不能直接在回调里写库OPC UA 的通知回调运行在客户端内部线程上如果在这里做同步数据库操作一旦 SQL Server 响应慢回调线程就被阻塞进而导致订阅队列积压最终触发BadNoCommunication或订阅超时。我见过最典型的翻车现场回调里new SqlConnection然后ExecuteNonQuery采集 50 个点数据库每秒写入 50 次连接池瞬间打满程序卡死。正确做法是回调只入队另起一个消费者线程批量写入。4.2 批量写入的两种方案SqlBulkCopy 与表值参数小批量每秒几十条可以用表值参数大批量每秒上千条必须上SqlBulkCopy。下面给一个SqlBulkCopy的封装配合DataTable使用。// 消费者线程从队列取数据攒够 500 条或超时 1 秒就批量写入 private async Task ConsumeAndWriteAsync(CancellationToken token) { var buffer new ListTagData(500); while (!token.IsCancellationRequested) { while (buffer.Count 500 _dataQueue.TryDequeue(out var data)) { buffer.Add(data); } if (buffer.Count 0) { await Task.Delay(100, token); continue; } var table new DataTable(); table.Columns.Add(TagName, typeof(string)); table.Columns.Add(Value, typeof(double)); table.Columns.Add(Timestamp, typeof(DateTime)); foreach (var d in buffer) table.Rows.Add(d.TagName, d.Value, d.Timestamp); using (var bulk new SqlBulkCopy(_connectionString)) { bulk.DestinationTableName dbo.TagHistory; bulk.BatchSize 500; bulk.BulkCopyTimeout 30; bulk.ColumnMappings.Add(TagName, TagName); bulk.ColumnMappings.Add(Value, Value); bulk.ColumnMappings.Add(Timestamp, Timestamp); await bulk.WriteToServerAsync(table); } buffer.Clear(); } }BatchSize设 500 和缓冲区大小一致BulkCopyTimeout给 30 秒避免大事务锁表太久。ColumnMappings必须显式写否则列顺序对不上会插错。目标表TagHistory建议加聚集索引在Timestamp上否则查询会越来越慢。4.3 连接字符串与连接池参数SQL Server 连接字符串里Max Pool Size默认是 100对于批量写入的单一消费者来说够用但如果你开了多个消费者线程要适当调大。Connect Timeout设 15 秒别用默认的 30 秒否则网络抖动时线程挂太久。Server127.0.0.1;DatabaseOpcData;User Idsa;PasswordYourPassword; Max Pool Size200;Connect Timeout15;MultipleActiveResultSetsfalse;MultipleActiveResultSets关掉批量写入用不到开着反而增加开销。5. 避坑与排查这五个问题我几乎每个项目都会遇到5.1 订阅创建成功但收不到数据现象subscription.Create()返回成功MonitoredItem也加进去了但回调一次都不触发。原因StartNodeId的命名空间索引或标识符写错服务端找不到节点但 OPC UA 不会主动报错只是静默不推数据。解决先用session.ReadValue(nodeId)单独读一次确认节点存在且能读到值再挂订阅。另外检查SamplingInterval是否小于服务端的最小采样间隔太小会被服务端拒绝。5.2 运行一段时间后报 BadSessionClosed现象程序跑几小时后所有订阅停止日志里出现BadSessionClosed。原因会话超时时间到了而客户端没有发心跳或重连。DefaultSessionTimeout设了 60 秒但KeepAliveCount和PublishingInterval配合不当导致服务端认为客户端已死。解决把KeepAliveCount设为DefaultSessionTimeout / PublishingInterval的 1.5 倍左右。比如超时 60 秒发布间隔 1 秒KeepAliveCount 至少 90。同时监听session.SessionClosing事件触发重连逻辑。5.3 SqlBulkCopy 报“无法访问目标表”现象WriteToServerAsync抛出SqlException提示目标表不存在或权限不足。原因连接字符串里的数据库名写错或者当前用户没有dbo.TagHistory的插入权限。解决先用 SSMS 用同一账号登录手动INSERT一条测试。如果手动能插检查DestinationTableName是否带了dbo.前缀。批量写入时表名必须写全dbo.TagHistory不能只写TagHistory。5.4 时间戳时区错乱现象写入 SQL Server 的Timestamp比实际时间少 8 小时。原因OPC UA 的SourceTimestamp是 UTC 时间直接存入DateTime字段时没有转换。解决在入队前统一转成本地时间value.SourceTimestamp.ToLocalTime()或者数据库字段用datetimeoffset类型存 UTC。我一般选前者查询时不用再换算。5.5 高频写入导致数据库日志暴涨现象采集频率 100ms 一次跑了半天数据库日志文件涨到几十 GB。原因SqlBulkCopy默认每批都是一个独立事务如果数据库是完整恢复模式日志会一直累积。解决把数据库恢复模式改成简单模式或者定期做日志备份。另外把BatchSize调大减少事务数量。如果数据允许丢可以在连接字符串里加SqlBulkCopyOptions.TableLock减少锁开销。6. 进阶技巧用通道背压把采集和写入彻底解耦上面用的ConcurrentQueue在数据量不大时没问题但如果你要采集上千个点位队列会无限增长最终内存爆掉。我后来改成System.Threading.Channels配合背压策略效果稳定很多。// 创建有界通道容量 10000满时等待而不是丢弃 var channel Channel.CreateBoundedTagData(new BoundedChannelOptions(10000) { FullMode BoundedChannelFullMode.Wait, SingleReader true, SingleWriter false }); // 回调里写入通道 item.Notification async (monitoredItem, args) { foreach (var value in args.Notification.Value) { await channel.Writer.WriteAsync(new TagData { TagName monitoredItem.DisplayName, Value Convert.ToDouble(value.Value), Timestamp value.SourceTimestamp.ToLocalTime() }); } }; // 消费者批量读取 var reader channel.Reader; var buffer new ListTagData(500); while (await reader.WaitToReadAsync()) { while (buffer.Count 500 reader.TryRead(out var data)) buffer.Add(data); if (buffer.Count 0) { await WriteBatchAsync(buffer); buffer.Clear(); } }BoundedChannelFullMode.Wait是关键通道满时写入方会异步等待自动形成背压不会丢数据也不会撑爆内存。SingleReader true告诉运行时只有一个消费者内部可以优化锁。WriteBatchAsync就是上一章的SqlBulkCopy封装。验证方法很简单在消费者里加一个计数器每秒打印一次处理条数。如果处理速度持续低于采集速度通道会逐渐填满此时WriteAsync的等待时间会变长你就能直观看到瓶颈在哪。我一般会把通道容量和处理速率做成可配置项现场调试时直接改配置不用重新编译。从那以后我每次做 OPC UA 采集都强制先跑一遍“只入队不写库”的空载测试确认订阅稳定、通道不积压再接数据库。这个习惯帮我省掉了至少三次半夜被叫起来查连接池的麻烦。希望帮到你。本文还有配套的精品资源点击获取