用C#实现OPC UA客户端并存入SQL Server的完整指南

简介：OPC UA客户端与SQL Server数据落地的C#实践项目，面向工业自动化、物联网数据采集及企业级数据集成的开发者，也可作为高校自动化专业项目的参考。项目借助OpcUaHelper开源库简化OPC UA协议交互，完整演示从服务器读取数据、以“_”分隔符转换字符串格式，再到SQL Server建表、插入落库的实现链路，对理解设备通信与服务端存储对接很有参考价值。资源共144个文件，以DLL依赖库、C#源码、配置与XML文档为主，整体约5.51MB，典型的Visual Studio工程结构便于按目录检索和二次开发。已有3239人学习浏览，适合正在搭建OPC UA采集服务或研究C#与SQL Server集成方案的开发者对照借鉴。

1. 用C#做OPC UA客户端并把数据存入SQL Server：为什么这是产线数字化的第一块基石

很多工厂的PLC、仪表、传感器都在跑OPC UA协议，但数据只停在HMI画面上，车间主任想看到的趋势曲线、OEE报表和能耗分析却拿不到数。我接过不少这类项目，最常见的痛不是设备不支持OPC UA，而是缺一个可靠的采集服务：用C#写OPC UA客户端连上服务器、订阅变量变化、再把带时间戳的数据写进SQL Server。这件事听起来简单，真正落地时会遇到证书校验、订阅不触发、时间戳偏移、批量写入性能这些坑。这篇笔记就是把这套链路完整拆开，从选型到踩坑，给准备自己动手的开发者一条能走通的路。

2. 选型与工程脚手架：官方SDK、.NET版本和UA安全策略先定下来

2.1 几类C# OPC UA客户端方案，区别在哪

接手一个OPC UA采集项目，第一个要回答的问题是“用什么库”。C#生态里大致有三条路：官方SDK、商业封装组件、自研协议栈。我偏向用官方SDK，原因是它在协议覆盖面上最全，UA的发现、订阅、方法调用、历史读取都支持，而且跨平台，能跑在Windows服务里，也能部署到Linux工控机。

商业组件上手快，曲线和变量管理都是现成的，但遇到非标行为时黑匣子很难排查，授权费用也不低。自研协议栈则不现实，OPC UA光安全策略、二进制编码和证书体系就够写几个月，只适合有极致裁剪需求的项目。所以我的默认方案是官方NuGet包，也就是包名OPCFoundation.NetStandard.Opc.Ua那个，随.NET Standard 2.0走，.NET 6/8的工控程序都能用。

选型还有一个隐形成本：团队里有没有人读过UA规范。官方SDK的资料散，但代码注释和示例工程其实很完整，关键是第一批连接要有人带。如果你之前只写过Modbus TCP采集，建议先在测试环境跑通一次匿名连接，再上生产。

2.2 最小工程：引用官方SDK并完成匿名连接

我先建一个控制台工程作为采集服务的主干，后面加Windows服务或容器化都方便。命令如下：

dotnet new console -n OpcUaToSql cd OpcUaToSql dotnet add package OpCFoundation.NetStandard.Opc.Ua

dotnet add package不带版本号时拉取当前最新稳定版，实际项目里建议锁定一个版本，避免同事机器上解析出不同版本导致行为差异。工程创建后，先写连接代码，核心是构造ApplicationConfiguration：

var appConfig = new ApplicationConfiguration { ApplicationName = "OpcUaCollector", ApplicationUri = "urn:collector:opcua", ApplicationType = ApplicationType.Client, SecurityConfiguration = new SecurityConfiguration { ApplicationCertificate = new CertificateIdentifier { StoreType = CertificateStoreType.X509Store, StorePath = "CurrentUser\\My", SubjectName = "CN=OpcUaCollector" }, TrustedPeerCertificates = new CertificateTrustList { StoreType = CertificateStoreType.Directory, StorePath = "Certificates/TrustedPeers" }, TrustedIssuerCertificates = new CertificateTrustList { StoreType = CertificateStoreType.Directory, StorePath = "Certificates/TrustedIssuers" }, AutoAcceptUntrustedCertificates = false }, TransportQuotas = new TransportQuotas { OperationTimeout = 15000 }, ClientConfiguration = new ClientConfiguration { DefaultSessionTimeout = 60000 } }; await appConfig.Validate(ApplicationType.Client);

这段配置里有几个关键参数。StorePath决定了客户端证书放在哪里，测试环境用CurrentUser\\My最省事，生产环境建议换成文件目录并在部署脚本里统一生成证书。AutoAcceptUntrustedCertificates默认必须设为false，否则就是给中间人攻击开门，测试时临时改true跑通后要改回来。OperationTimeout是单次请求的超时，工控网正常在10到30秒之间，设太短会误伤慢设备。

配置完成后，用发现端点的方式拿到服务器的Endpoint描述，再创建会话：

// 通过发现端点获取服务器支持的安全策略 var endpoints = await DiscoveryClient.GetEndpointsAsync("opc.tcp://127.0.0.1:4840"); var selected = endpoints.FirstOrDefault(e => e.SecurityPolicyUri == SecurityPolicies.None); var configured = new ConfiguredEndpoint(null, selected, EndpointConfiguration.Create(appConfig)); // 匿名身份建立会话，sessionName用于服务器端审计日志区分客户端 var session = await Session.Create( appConfig, configured, false, "CSharpDataCollector", 60000, new UserIdentity(), // 匿名身份 null);

先做GetEndpointsAsync是为了拿到服务器实际支持的SecurityPolicyUri列表，有些设备只开Basic256Sha256，你硬选None就会被拒。Session.Create里的updateBeforeConnect参数为false表示不再重新发现端点，直连更快。匿名身份适合内网且服务器允许匿名的情况，UserIdentity()换成new UserIdentity("user", "password")就是用户名密码模式。

2.3 用户名密码与证书校验：把安全策略配到生产可用

内网部署时很多人图省事全用匿名加None策略，但设备和服务器都在增加，安全策略还是要配起来。生产环境我一般至少做到：客户端证书由统一脚本生成，服务器证书加入客户端信任目录，用户名密码走独立账号不共用管理员。

证书信任这部分最容易翻车。把TrustedPeerCertificates指向一个共享目录，部署多台采集机器时把服务器证书拷贝进去即可。服务器那边也要把每台采集机的客户端证书加入信任列表，单向信任会导致服务器主动断开会话。如果你第一次连某个新服务器，看Certificates/Rejected目录，那里躺着被拒的证书，把它挪到Trusted目录再重连是测试环境最常用的手段。

用户名密码模式下，UA服务器的用户管理通常独立于操作系统账号，配置时注意密码策略和权限范围，给采集账号只读权限即可，不要给能写点的管理员账号。UA还会在会话建立时协商传输安全，如果选了Basic256Sha256但证书过期，连接不会马上报错，而是表现为握手后立即断开，日志里只有BadCertificateUntrusted，这类问题排查起来最耗时间。

3. 读数据的两条路线：订阅推送与定时轮询怎么选

3.1 订阅模式：数据变化主动推送的完整实现

OPC UA最实用的能力是订阅，服务器在有数据变化时主动推给客户端，不占请求资源。创建订阅和监控项的代码如下：

var subscription = new Subscription(session.DefaultSubscriptionState) { PublishingInterval = 1000, // 发布间隔，单位毫秒 PublishingEnabled = true, // 必须显式打开 LifetimeCount = 100, KeepAliveCount = 10 }; session.AddSubscription(subscription); subscription.Create(); var item = new MonitoredItem(session.DefaultSubscriptionState, new NodeId("ns=2;i=1001"), Attributes.Value) { SamplingInterval = 500, // 采样间隔，单位毫秒 QueueSize = 20, // 队列深度，防止积压丢数据 DiscardOldest = true }; item.Notification += OnNotification; subscription.AddItem(item); subscription.ApplyChanges();

PublishingInterval是客户端希望服务器多久发布一次通知，SamplingInterval是服务器多久采样一次变量，两者配合决定数据延迟。比如采样500ms、发布1s，那1秒内最多推送2个变化值，够大多数产线场景。QueueSize=20加DiscardOldest=true表示队列满了丢最旧的数据，对实时采集合理；如果要做历史追溯，应该把这个队列调大或改成丢弃最新值，避免重要数据被冲掉。

回调里拿到的数据要第一时间放进内存队列，不要在回调里做数据库写入或日志等耗时操作。数据到达回调后，取值、质量、时间戳都是DataValue的字段：

private static void OnNotification(MonitoredItem item, MonitoredItemNotificationEventArgs e) { if (e.NotificationValue is not MonitoredItemNotification notification) return; var value = notification.Value; if (value.StatusCode.Code != StatusCodes.Good) return; // 非Good数据不进缓存，避免污染统计 _queue.Add(new TagRecord { TagName = item.StartNodeId.ToString(), Value = value.Value as double?, Quality = value.StatusCode.Code, SourceTimestamp = value.SourceTimestamp }); }

item.StartNodeId.ToString()得到的是类似ns=2;i=1001的字符串，建议在建监控项时自己维护一份点位名到NodeId的映射，把TagName存成Pressure_1这种业务名，报表时不用再解析节点字符串。SourceTimestamp是服务器或设备标记的原始时间，比客户端收到通知的时间更接近真实发生时刻，入库优先用它。

3.2 定时轮询：读快照的兜底方案

订阅虽好，但有些老设备或网关实现的OPC UA服务器并不支持订阅，或者点位刷新本身就慢，这时候轮询反而更可靠。轮询的基础就是隔一段时间主动读一次：

using var cts = new CancellationTokenSource(); while (!cts.IsCancellationRequested) { try { var value = await session.ReadValueAsync( new NodeId("ns=2;i=1001"), cts.Token); if (value.StatusCode.Code == StatusCodes.Good) { _queue.Add(new TagRecord { TagName = "Pressure_1", Value = value.Value as double?, Quality = value.StatusCode.Code, SourceTimestamp = value.SourceTimestamp }); } } catch (ServiceResultException ex) { // BadSessionId表示会话失效，置标记让重连逻辑处理 if (ex.StatusCode.Code == StatusCodes.BadSessionId) _sessionNeedsReconnect = true; } await Task.Delay(1000, cts.Token); // 轮询间隔，按点位变化频率调整 }

轮询间隔不要小于100ms，除非点位极少且服务器性能足够，否则容易把服务器CPU打满。读取采用ReadValueAsync是轻量操作，多个点位串行读太慢，可以并发读，但并发数不要超过10，工控设备的服务线程通常有限。

轮询一个必须处理的异常是BadSessionId，这表示会话已经失效，还在循环里继续读只会空转。我一般会在外层套一个连接管理器，检测到这个状态就整体重建Session。

3.3 两种读法对比：点位规模、实时性和服务器负担

选择订阅还是轮询，核心看三点：点位规模、实时性要求、服务器能力。

我常用的组合是：优先订阅，对关键点位额外每分钟轮询一次做对账，两路数据在入库前去重。这个对账能有效发现订阅静默失效的情况，算是给采集链路上了一道保险。

4. 写入SQL Server：表结构、队列和批量入库

4.1 历史数据表怎么建才不后悔

点位数据要写成历史表，第一原则是“只存事实，不做加工”。一张表同时存数值、质量、时间戳，别把阈值判断、单位换算的结果也塞进来，那是报表层的职责。我常用的建表语句如下：

CREATE TABLE dbo.TagHistory ( Id BIGINT IDENTITY(1,1) NOT NULL PRIMARY KEY, TagName NVARCHAR(128) NOT NULL, ValueFloat FLOAT NULL, ValueText NVARCHAR(512) NULL, QualityCode INT NOT NULL, SourceTimestamp DATETIME2(7) NOT NULL, ServerTimestamp DATETIME2(7) NULL ); CREATE INDEX IX_TagHistory_TagName_Time ON dbo.TagHistory (TagName, SourceTimestamp DESC);

SourceTimestamp用DATETIME2(7)而不是DATETIME，原因是DATETIME精度只有约3.33毫秒且不支持UTC语义，DATETIME2(7)精度100纳秒，OPC UA的时间戳精度再高也不会被截断。ValueFloat和ValueText并存，是为了兼容模拟量和字符串型点位，一个点位用哪个字段由写入端约定，别让报表去猜。

索引只建一个TagName + SourceTimestamp的组合索引，历史表写多读少，索引太多会拖慢写入。超过千万行后按月份做分区或归档到历史库，这个可以在表设计阶段预留分区间隔，不必一开始就做分区。

4.2 从内存队列到SqlBulkCopy：批量入库的核心实现

逐条INSERT在点位超过一百个时就成了性能瓶颈，每行都走一次日志和锁等待。正确做法是采集线程只往BlockingCollection里放，入库线程攒批后SqlBulkCopy一次写入。

private static readonly BlockingCollection<TagRecord> _queue = new( new ConcurrentQueue<TagRecord>(), 20000); // 采集回调里调用 _queue.Add(record); // 入库线程： private static async Task ConsumerAsync( string connectionString, CancellationToken ct) { var table = new DataTable(); table.Columns.Add("TagName", typeof(string)); table.Columns.Add("ValueFloat", typeof(double)); table.Columns.Add("QualityCode", typeof(uint)); table.Columns.Add("SourceTimestamp", typeof(DateTime)); foreach (var record in _queue.GetConsumingEnumerable(ct)) { table.Rows.Add( record.TagName, record.Value, record.Quality, record.SourceTimestamp); if (table.Rows.Count < 1000) continue; await FlushAsync(table, connectionString, ct); table.Clear(); } } private static async Task FlushAsync( DataTable table, string connectionString, CancellationToken ct) { using var bulk = new SqlBulkCopy( connectionString, SqlBulkCopyOptions.UseInternalTransaction) { DestinationTableName = "dbo.TagHistory", BatchSize = 1000, BulkCopyTimeout = 30 }; bulk.ColumnMappings.Add("TagName", "TagName"); bulk.ColumnMappings.Add("ValueFloat", "ValueFloat"); bulk.ColumnMappings.Add("QualityCode", "QualityCode"); bulk.ColumnMappings.Add("SourceTimestamp", "SourceTimestamp"); await bulk.WriteToServerAsync(table, ct); }

BlockingCollection的容量设为20000是缓冲区上限，超过后Add会阻塞，反过来通过背压让采集端放慢速度，避免内存无限增长。BatchSize=1000是经验值，太小体现不出批量优势，太大单次事务时间过长，出问题回滚代价高。SqlBulkCopyOptions.UseInternalTransaction让每一批写入自带事务，这一批失败只回滚本批，不影响前面已提交的数据。

批量入库的时机除了“攒够1000行”，还应该加一个时间维度：当点位少、数据量半天凑不够批时，数据会一直憋在内存里。我在消费循环里加一个定时器，超过2秒就强制刷一次，保证时效性。

4.3 时间戳与Quality：两个最容易存错的东西

采集程序存时间戳时，最隐蔽的错误是混用多种时间来源。SourceTimestamp、ServerTimestamp、客户端本地时间三者含义完全不同：SourceTimestamp是设备或服务器标记的发生时间，ServerTimestamp是UA服务器收到数据的时间，客户端本地时间是通知到达的时间。报表做趋势分析应该用SourceTimestamp，但很多设备时钟并不同步，所以表里我再存一个ServerTimestamp用来排查偏差。客户端本地时间只在没有前两个字段时才用，并且统一转成UTC再入库。

Quality字段直接存StatusCode.Code的原始uint值，不要存StatusCode.ToString()的结果。原因有两个：字符串在统计时没法过滤坏值；StatusCode里除了质量还包含IDataChange等位信息，存原始值后随时可以解析。

还有一个SQL Server特有的坑：DateTime类型会在内部做舍入，SourceTimestamp如果是毫秒级时间戳，插入后秒的毫秒部分可能被篡改。使用DATETIME2(7)可以避免这种情况，但要注意从C#传进去的DateTime如果Kind是Local，写入后会和UTC时间混在一起。我习惯在组装DataTable前统一转成DateTime.SpecifyKind(record.SourceTimestamp, DateTimeKind.Utc)。

5. 避坑排查：客户端连不上、订阅不触发和死锁实录

5.1 连接总报“证书不受信任”

现象：第一次连某台新服务器，程序抛ServiceResultException，状态码是BadCertificateUntrusted，看日志里写的是证书链验证失败。

原因：OPC UA客户端首次运行会生成自签名证书，这台服务器不认识它；同时服务器自己的证书也不在客户端的TrustedPeerCertificates目录里。两边互不信任，握手就直接失败。不少人图快把AutoAcceptUntrustedCertificates设成true，结果测试环境越跑越顺，上线换域名后又开始报错，因为证书里的URI和实际地址对不上。

解决：在Certificates/Rejected目录下找到被拒绝的服务器证书，复制到TrustedPeers目录，重启程序。服务器那边同理，把客户端证书导入它的信任列表。生产环境的客户端证书要通过统一脚本生成，Subject里的CN和域名保持一致，这样即使采集机换了IP，证书依然有效。最省事的测试方法就是把AutoAcceptUntrustedCertificates临时开一次，连上后立刻关掉再验证。

5.2 订阅建立了却一直不回调

现象：会话建好了，监控项也AddItem了，但回调函数一次都没进，程序也不报错。断点打在OnNotification里，等了十分钟毫无动静。

原因：最常见的是PublishingEnabled没设true，或者PublishingInterval设得非常大。另一个原因藏在SamplingInterval里，设成-1表示用服务器默认值，有些服务器的默认采样间隔是几秒甚至禁用状态，表面看订阅成功实际没数据。还有一种情况是点位本身是常量或变化极慢，订阅模式只在值变化时推送，不回调用属正常。

解决：把PublishingInterval显式设为1000，SamplingInterval设为500并确保节点在服务器端确实在变化。排查时先用ReadValueAsync读一次，确认能读到值；再用服务器的诊断信息看监控项状态，如果节点ID写错，状态里会有BadNodeIdInvalid。全部参数都对还不行，就检查会话的KeepAlive是否已经超时，服务器可能已经悄悄断开了会话。

5.3 存进库的时间戳和服务端差好几个小时

现象：报表里看曲线，发现设备和服务器时间对不上，同样一个数据点在数据库里比现场慢8小时，而且不同点位偏差还不一样。

原因：OPC UA规范规定SourceTimestamp应该是UTC时间，但很多国产设备的UA服务器直接写本地时间，时区偏移就直接带进了数据。有些采集程序又在入库时用DateTime.Now覆盖了原始时间戳，等于把“现象时间”换成了“采集时间”，偏差就更大。

解决：采集端不管设备传什么，一律把SourceTimestamp当UTC处理并转成DateTimeKind.Utc，然后额外存一个ServerTimestamp用于对账。如果服务器确实传的是本地时间，在客户端做一次统一偏移修正，写成配置项，不要硬编码时区。Oracle和MySQL处理时区的方式不同，但SQL Server没有时区类型，所以约定好“表内全存UTC，展示层转本地”是最省心的。

5.4 逐条INSERT把数据库写成了死锁现场

现象：点位加到几百个后，数据库出现大量PRIMARY KEY冲突或死锁错误1205，CPU不高但插入速度就是上不去，日志文件疯涨。

原因：采集回调里直接ExecuteNonQuery逐条插入，每条都会启动隐式事务、写日志、参与锁竞争。多个点位并发插入时，索引页和分配页争用加剧，死锁概率直接上升。这不是SQL Server不行，而是写入模式本身不适合高频小事务。

解决：写入链路改成“内存队列 + 攒批 + SqlBulkCopy”。BatchSize控制在500到2000之间，表上只保留一个必要的组合索引。如果业务允许，把数据库的READ_COMMITTED_SNAPSHOT打开，能明显减少读写互相阻塞，但需要DBA评估后由运维执行，采集程序本身不用改代码。

6. 进阶技巧：断线重连与坏值过滤，让采集程序敢挂生产

6.1 断线自动重连的最小实现

生产环境的网络不可能永远稳定，交换机重启、服务器维护都可能让会话断开。OPC UA的KeepAlive事件会在超过阈值后触发，我在这里只置一个标志，不直接做重连：

private static bool _sessionNeedsReconnect; session.KeepAlive += (s, e) => { if (e.ServiceResult.Code != StatusCodes.Good) _sessionNeedsReconnect = true; };

主循环里每隔几秒检查这个标志，发现需要重连时释放旧会话，重新走发现端点和Session.Create流程。不要直接在KeepAlive回调里做重连，事件线程会被漫长的握手阻塞，影响其他监控项的响应。

6.2 只存“好值”：Quality过滤的两种做法

历史库里混入坏值是最容易污染报表的。质量码Bad、Uncertain、BadNoData这些数据点，有的表示设备断线，有的表示传感器超量程，有的表示数值不可信。我见过某项目把停机期间的坏值当正常产量存进去，月度报表直接失真。

过滤策略跟用途绑定：做KPI和趋势分析，只存StatusCode.Code == StatusCodes.Good；做设备OEE分析，坏值也要记录，但单独建表或加QualityCode标记，统计时显式排除。项目上我通常把坏值单独落一张TagHistoryBad表，保留全部原始状态码，主表保持干净，既不影响统计也不会丢诊断线索。

6.3 上线前怎么验证这套链路

新项目我习惯先用计数器模拟几百个点位跑24小时，然后跑一遍对账查询确认没有丢数。最直接的验证方法是比对订阅回调次数和入库条数：

SELECT TagName, COUNT(*) AS RowCount FROM dbo.TagHistory WHERE SourceTimestamp >= DATEADD(hour, -1, GETUTCDATE()) GROUP BY TagName ORDER BY TagName;

每个点位的行数应该和模拟器产生的变化次数一致，如果有点位明显少，优先检查是否被QueueSize丢弃或坏值过滤掉了。之后再故意断掉服务器网络，确认重连标志和会话重建生效，恢复后数据能继续写入。记得在验证时把KeepAliveCount和PublishingInterval调小，让断线检测在30秒内触发，不然测试要等很久。

我第一次做这个链路时，图省事没用批量入库，结果一百个点位就把数据库写成了死锁现场，后来才老老实实改成队列加SqlBulkCopy。做采集服务，“先可靠再高效”是铁律，坏值宁可丢掉也不要存进主表。这条链路跑稳之后，后面加点位、加报表都只是改配置的事，希望这篇笔记能帮你少走几趟弯路。

本文还有配套的精品资源，点击获取