using System; using System.Collections.Generic; using System.Collections.ObjectModel; using System.Linq; using System.Threading; using System.Threading.Tasks; using Microsoft.Extensions.Hosting; using Opc.Ua; using Opc.Ua.Client; using Opc.Ua.Configuration; using PMSWPF.Enums; using PMSWPF.Helper; using PMSWPF.Models; namespace PMSWPF.Services { public class OpcUaBackgroundService { private readonly DataServices _dataServices; // 存储 OPC UA 设备,键为设备Id,值为会话对象。 private readonly Dictionary _deviceDic; // 存储 OPC UA 会话,键为终结点 URL,值为会话对象。 private readonly Dictionary _sessionsDic; // 存储 OPC UA 订阅,键为终结点 URL,值为订阅对象。 private readonly Dictionary _subscriptionsDic; // 存储活动的 OPC UA 变量,键为变量的OpcNodeId private readonly Dictionary _opcUaNodeIdVariableDic; // Key: VariableData.Id // 储存所有要轮询更新的变量,键是Device.Id,值是这个设备所有要轮询的变量 private readonly Dictionary> _pollVariableDic; // Key: VariableData.Id // 储存所有要订阅更新的变量,键是Device.Id,值是这个设备所有要轮询的变量 private readonly Dictionary> _subVariableDic; // 后台服务的主线程,负责连接服务器,加载变量,订阅变量 private Thread _serviceMainThread; // 轮询线程 private Thread _pollThread; // 重新加载事件 private readonly ManualResetEvent _reloadEvent = new ManualResetEvent(false); // 停止事件,触发后会停止整个Opc后台服务 private readonly ManualResetEvent _stopdEvent = new ManualResetEvent(false); /// /// OpcUaBackgroundService 的构造函数。 /// /// 数据服务,用于访问数据库中的变量信息。 public OpcUaBackgroundService(DataServices dataServices) { _dataServices = dataServices; _sessionsDic = new Dictionary(); _subscriptionsDic = new Dictionary(); _opcUaNodeIdVariableDic = new(); _pollVariableDic = new(); _subVariableDic = new(); _deviceDic = new(); } /// /// 后台服务的主执行方法。 /// /// 用于通知服务停止的取消令牌。 /// 表示异步操作的任务。 public void StartService() { NlogHelper.Info("OPC UA 服务正在启动..."); _reloadEvent.Set(); _serviceMainThread = new Thread(Execute); _serviceMainThread.IsBackground = true; _serviceMainThread.Name = "OpcUaServiceThread"; _serviceMainThread.Start(); } public void StopService() { NlogHelper.Info("OPC UA 服务正在停止..."); _stopdEvent.Set(); DisconnectAllOpcUaSessions(); _reloadEvent.Close(); _stopdEvent.Close(); NlogHelper.Info("OPC UA 服务已经停止。"); } private void Execute() { // 订阅变量数据变化事件,以便在变量配置发生变化时重新加载。 _dataServices.OnDeviceListChanged += HandleDeviceListChanged; while (!_stopdEvent.WaitOne(0)) { _reloadEvent.WaitOne(); if (_dataServices.Devices == null || _dataServices.Devices.Count == 0) { _reloadEvent.Reset(); continue; } NlogHelper.Info("OpcUa后台服务开始加载变量..."); // 初始化时加载所有活动的 OPC UA 变量。 LoadOpcUaVariables(); //连接服务器 ConnectOpcUaService(); // // 添加订阅变量 SetupOpcUaSubscription(); if (_pollThread == null) { _pollThread = new Thread(PollOpcUaVariable); _pollThread.IsBackground = true; _pollThread.Name = "OpcUaPollThread"; _pollThread.Start(); } NlogHelper.Info("OpcUa后台服务已启动。"); _reloadEvent.Reset(); } // 循环运行,直到接收到停止信号。 // while (!stoppingToken.IsCancellationRequested) // { // // 可以在这里添加周期性任务,例如检查和重新连接断开的会话。 // await Task.Delay(TimeSpan.FromSeconds(10), stoppingToken); // } NlogHelper.Info("OpcUa后台服务正在停止。"); // 服务停止时,取消订阅事件并断开所有 OPC UA 连接。 _dataServices.OnDeviceListChanged -= HandleDeviceListChanged; } private void HandleDeviceListChanged(List devices) { NlogHelper.Info("变量数据已更改。正在重新加载 OPC UA 变量。"); _reloadEvent.Set(); } /// /// 从数据库加载所有活动的 OPC UA 变量,并进行相应的连接和订阅管理。 /// private void LoadOpcUaVariables() { try { var _opcUaDevices = _dataServices .Devices.Where(d => d.ProtocolType == ProtocolType.OpcUA && d.IsActive == true) .ToList(); if (_opcUaDevices.Count == 0) return; _deviceDic.Clear(); _pollVariableDic.Clear(); _subVariableDic.Clear(); _opcUaNodeIdVariableDic.Clear(); foreach (var opcUaDevice in _opcUaDevices) { // 将设备保存到字典中,方便之后查找 _deviceDic.Add(opcUaDevice.Id, opcUaDevice); //查找设备中所有要轮询的变量 var dPollList = opcUaDevice.VariableTables?.SelectMany(vt => vt.DataVariables) .Where(vd => vd.IsActive == true && vd.ProtocolType == ProtocolType.OpcUA && vd.OpcUaUpdateType == OpcUaUpdateType.OpcUaPoll) .ToList(); // 将变量保存到字典中,方便Read后还原 foreach (var variableData in dPollList) { _opcUaNodeIdVariableDic.Add(variableData.OpcUaNodeId, variableData); } NlogHelper.Info($"加载OpcUa轮询变量:{dPollList.Count}个"); _pollVariableDic.Add(opcUaDevice.Id, dPollList); //查找设备中所有要订阅的变量 var dSubList = opcUaDevice.VariableTables?.SelectMany(vt => vt.DataVariables) .Where(vd => vd.IsActive == true && vd.ProtocolType == ProtocolType.OpcUA && vd.OpcUaUpdateType == OpcUaUpdateType.OpcUaSubscription) .ToList(); _subVariableDic.Add(opcUaDevice.Id, dSubList); NlogHelper.Info($"加载OpcUa订阅变量:{dSubList.Count}个"); } } catch (Exception e) { NotificationHelper.ShowError($"加载OpcUa变量的过程中发生了错误:{e.Message}"); } } /// /// 连接到 OPC UA 服务器并订阅或轮询指定的变量。 /// private void ConnectOpcUaService() { foreach (Device device in _deviceDic.Values) { Session session = null; // 检查是否已存在到该终结点的活动会话。 if (_sessionsDic.TryGetValue(device.OpcUaEndpointUrl, out session) && session.Connected) { NlogHelper.Info($"已连接到 OPC UA 服务器: {device.OpcUaEndpointUrl}"); continue; } session = ServiceHelper.CreateOpcUaSession(device.OpcUaEndpointUrl); if (session == null) return; // 连接失败,直接返回 _sessionsDic[device.OpcUaEndpointUrl] = session; } } private void PollOpcUaVariable() { NlogHelper.Info("OpcUa轮询变量线程已启动,开始轮询变量...."); while (!_stopdEvent.WaitOne(0)) { try { foreach (var deviceId in _pollVariableDic.Keys.ToList()) { Thread.Sleep(100); if (!_deviceDic.TryGetValue(deviceId, out var device) || device.OpcUaEndpointUrl == null) { NlogHelper.Warn( $"OpcUa轮询变量时,在deviceDic中未找到ID为 {deviceId} 的设备,或其服务器地址为空,请检查!"); continue; } _sessionsDic.TryGetValue(device.OpcUaEndpointUrl, out Session session); if (session == null || !session.Connected) { if (!_stopdEvent.WaitOne(0)) { NlogHelper.Warn( $"用于 {device.OpcUaEndpointUrl} 的 OPC UA 会话未连接。正在尝试重新连接..."); // 尝试重新连接会话 ConnectOpcUaService(); continue; } } var nodesToRead = new ReadValueIdCollection(); if (!_pollVariableDic.TryGetValue(deviceId, out var variableList)) { continue; } foreach (var variable in variableList) { // 获取变量的轮询间隔。 if (!ServiceHelper.PollingIntervals.TryGetValue(variable.PollLevelType, out var interval)) { NlogHelper.Info($"未知的轮询级别 {variable.PollLevelType},跳过变量 {variable.Name}。"); continue; } // 检查是否达到轮询时间。 if ((DateTime.Now - variable.UpdateTime) < interval) continue; // 未到轮询时间,跳过。 nodesToRead.Add(new ReadValueId { NodeId = new NodeId(variable.OpcUaNodeId), AttributeId = Attributes.Value }); } // 如果没有要读取的变量则跳过 if (nodesToRead.Count == 0) continue; session.Read( null, 0, TimestampsToReturn.Both, nodesToRead, out DataValueCollection results, out DiagnosticInfoCollection diagnosticInfos); if (results == null || results.Count == 0) continue; for (int i = 0; i < results.Count; i++) { var value = results[i]; var nodeId = nodesToRead[i] .NodeId.ToString(); if (!_opcUaNodeIdVariableDic.TryGetValue(nodeId, out var variable)) { NlogHelper.Warn( $"在字典中未找到OpcUaNodeId为 {nodeId} 的变量对象!"); continue; } if (!StatusCode.IsGood(value.StatusCode)) { NlogHelper.Warn( $"读取 OPC UA 变量 {variable.Name} ({variable.OpcUaNodeId}) 失败: {value.StatusCode}"); continue; } // 更新变量数据 variable.DataValue = value.Value.ToString(); variable.DisplayValue = value.Value.ToString(); // 或者根据需要进行格式化 variable.UpdateTime = DateTime.Now; NlogHelper.Info($"轮询变量:{variable.Name},值:{variable.DataValue}"); // Console.WriteLine($"结果变量跟更新时间:{variable.UpdateTime}"); // await _dataServices.UpdateVariableDataAsync(variable); } } } catch (Exception ex) { NotificationHelper.ShowError($"OPC UA 轮询期间发生错误: {ex.Message}", ex); } } NlogHelper.Info("OpcUa轮询变量线程已停止。"); } /// /// 订阅变量变化的通知 /// /// /// private void OnSubNotification(MonitoredItem monitoredItem, MonitoredItemNotificationEventArgs e) { foreach (var value in monitoredItem.DequeueValues()) { NlogHelper.Info( $"[OPC UA 通知] {monitoredItem.DisplayName}: {value.Value} | 时间戳: {value.SourceTimestamp.ToLocalTime()} | 状态: {value.StatusCode}"); Console.WriteLine( $"[通知] {monitoredItem.DisplayName}: {value.Value} | 时间戳: {value.SourceTimestamp.ToLocalTime()} | 状态: {value.StatusCode}"); } } /// /// 断开所有 OPC UA 会话。 /// private void DisconnectAllOpcUaSessions() { NlogHelper.Info("正在断开所有 OPC UA 会话..."); foreach (var endpointUrl in _sessionsDic.Keys.ToList()) { NlogHelper.Info($"正在断开 OPC UA 会话: {endpointUrl}"); if (_sessionsDic.TryGetValue(endpointUrl, out var session)) { if (_subscriptionsDic.TryGetValue(endpointUrl, out var subscription)) { // 删除订阅。 subscription.Delete(true); _subscriptionsDic.Remove(endpointUrl); } // 关闭会话。 session.Close(); _sessionsDic.Remove(endpointUrl); NotificationHelper.ShowInfo($"已从 OPC UA 服务器断开连接: {endpointUrl}"); } } } /// /// 设置 OPC UA 订阅并添加监控项。 /// /// OPC UA 会话。 /// 要订阅的变量信息。 /// OPC UA 服务器的终结点 URL。 private void SetupOpcUaSubscription() { foreach (var deviceId in _subVariableDic.Keys) { var device = _dataServices.Devices.FirstOrDefault(d => d.Id == deviceId); Subscription subscription = null; // 得到session if (!_sessionsDic.TryGetValue(device.OpcUaEndpointUrl, out var session)) { NlogHelper.Info($"从OpcUa会话字典中获取会话失败: {device.OpcUaEndpointUrl} "); continue; } // 判断设备是否已经添加了订阅 if (_subscriptionsDic.TryGetValue(device.OpcUaEndpointUrl, out subscription)) { NlogHelper.Info($"OPC UA 终结点 {device.OpcUaEndpointUrl} 已存在订阅。"); } else { subscription = new Subscription(session.DefaultSubscription); subscription.PublishingInterval = 1000; // 发布间隔(毫秒) session.AddSubscription(subscription); subscription.Create(); _subscriptionsDic[device.OpcUaEndpointUrl] = subscription; } // 将变量添加到订阅 foreach (VariableData variable in _subVariableDic[deviceId]) { // 7. 创建监控项并添加到订阅中。 MonitoredItem monitoredItem = new MonitoredItem(subscription.DefaultItem); monitoredItem.DisplayName = variable.Name; monitoredItem.StartNodeId = new NodeId(variable.OpcUaNodeId); // 设置要监控的节点 ID monitoredItem.AttributeId = Attributes.Value; // 监控节点的值属性 monitoredItem.SamplingInterval = 1000; // 采样间隔(毫秒) monitoredItem.QueueSize = 1; // 队列大小 monitoredItem.DiscardOldest = true; // 丢弃最旧的数据 // 注册数据变化通知事件。 monitoredItem.Notification += OnSubNotification; subscription.AddItem(monitoredItem); subscription.ApplyChanges(); // 应用更改 } NlogHelper.Info($"设备: {device.Name}, 添加了 {(_subVariableDic[deviceId]?.Count ?? 0)} 个订阅变量。"); } } } }