跳到主要内容

大规模监控 (1000+ 节点)

工厂级 SCADA 经常要监控成千上万个 Tag. 关键不是 SDK 性能, 是怎么用 — 多订阅分流 / AddMany 批量 / 异步消费. 用 C# 示例.

配套示例

完整代码​

using DarraOpcUa_Client;
using System;
using System.Collections.Concurrent;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using System.Threading.Channels;

class Program
{
    static async Task Main()
    {
        // 替换成现场 PLC 工控机的 IP
        const string endpoint = "opc.tcp://192.168.1.100:4840";

        // 1500 个 Tag, 按更新频率分三组 (实际项目里 NodeId 格式以 PLC 项目导出的标识符为准)
        var fastTags   = Enumerable.Range(1, 100 ).Select(i => $"ns=2;s=Fast.T{i}").ToList();    // 100ms
        var mediumTags = Enumerable.Range(1, 800 ).Select(i => $"ns=2;s=Medium.T{i}").ToList();  // 1s
        var slowTags   = Enumerable.Range(1, 600 ).Select(i => $"ns=2;s=Slow.T{i}").ToList();    // 30s

        using var ua = new DarraOpcUa(endpoint);
        ua.Connect();
        Console.WriteLine($"[OK] 已连接 {endpoint}");

        // 1. 异步消费队列
        var channel = Channel.CreateUnbounded<DataChangeEventArgs>();
        long counter = 0;
        _ = Task.Run(async () =>
        {
            await foreach (var e in channel.Reader.ReadAllAsync())
            {
                Interlocked.Increment(ref counter);
                // TODO: 在这里写数据库 / 文件 / 转发
            }
        });

        // 2. 三个订阅, 不同 publishingInterval
        using var fast   = ua.CreateSubscription(100);
        using var medium = ua.CreateSubscription(1_000);
        using var slow   = ua.CreateSubscription(30_000);

        EventHandler<DataChangeEventArgs> handler = (s, e) => channel.Writer.TryWrite(e);
        fast.DataChanged   += handler;
        medium.DataChanged += handler;
        slow.DataChanged   += handler;

        // 3. AddMany 一次 RPC 加 N 个监控项
        fast  .AddMany(fastTags);
        medium.AddMany(mediumTags);
        slow  .AddMany(slowTags);
        Console.WriteLine($"已订阅 {fastTags.Count + mediumTags.Count + slowTags.Count} 个节点 (3 个 Subscription)");

        // 4. 每分钟报告吞吐
        var timer = new Timer(_ =>
        {
            var n = Interlocked.Exchange(ref counter, 0);
            Console.WriteLine($"{DateTime.Now:HH:mm:ss}  上一分钟 DataChanges = {n} ({n / 60.0:F0}/s)");
        }, null, 60_000, 60_000);

        Console.WriteLine("\n运行中, Ctrl+C 退出...");
        await Task.Delay(Timeout.Infinite);
    }
}

代码要点​

  • 按更新频率分多个 Subscription: 高频放慢订阅 = 凑批延迟丢实时性, 低频放快订阅 = 服务端 CPU 浪费
  • 经验值: 单 Subscription 不超过 1000 MI, 单 Session 不超过 ~10000 MI, 再多分多个 Session
  • 回调统一 channel.Writer.TryWrite 入队, 任何阻塞 (DB/文件/HTTP) 都拖慢 Publish, NotificationQueue 满会丢
  • Interlocked.Increment 计吞吐比拉 ServerStatus 直接, 长肥管道可调大 KeepAliveCount/LifetimeCount

注意事项​

  • 单 Session 极限: 一般认为 ~10000 MI 是单 Session 上限, 再多建议起多个 Session 分到多个 CPU 核.
  • KeepAliveCount / LifetimeCount 默认能用, 长肥管道 (高带宽高延迟) 可以适当调大避免 Publish 抖动误判断线.
  • DataChange 用 Interlocked.Increment 计数监控吞吐, 比定时拉 ServerStatus 更直接.

相关链接​