From bdae3355aa5ddc78a655135e2cbd6d815fb78f41 Mon Sep 17 00:00:00 2001 From: "David P.G" Date: Wed, 3 Sep 2025 19:58:02 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BF=AE=E6=94=B9OpcUa=E5=90=8E=E5=8F=B0?= =?UTF-8?q?=E6=9C=8D=E5=8A=A1(=E6=9C=AA=E5=AE=8C=E6=88=90)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../Interfaces/IDeviceDataService.cs | 13 - .../Services/DeviceDataService.cs | 56 -- .../Services/MqttBackgroundService.cs | 4 +- .../Services/OpcUaBackgroundService.cs | 489 ++++++++---------- .../Services/S7BackgroundService.cs | 130 +++-- DMS.WPF/App.xaml.cs | 2 + 6 files changed, 266 insertions(+), 428 deletions(-) delete mode 100644 DMS.Application/Interfaces/IDeviceDataService.cs delete mode 100644 DMS.Infrastructure/Services/DeviceDataService.cs diff --git a/DMS.Application/Interfaces/IDeviceDataService.cs b/DMS.Application/Interfaces/IDeviceDataService.cs deleted file mode 100644 index 2a9cf14..0000000 --- a/DMS.Application/Interfaces/IDeviceDataService.cs +++ /dev/null @@ -1,13 +0,0 @@ -using DMS.Core.Models; -using System.Collections.Generic; -using System.Threading.Tasks; - -namespace DMS.Application.Interfaces; - -public interface IDeviceDataService -{ - List Devices { get; } - event Action> OnDeviceListChanged; - event Action OnDeviceIsActiveChanged; - Task InitializeAsync(); -} \ No newline at end of file diff --git a/DMS.Infrastructure/Services/DeviceDataService.cs b/DMS.Infrastructure/Services/DeviceDataService.cs deleted file mode 100644 index 0309316..0000000 --- a/DMS.Infrastructure/Services/DeviceDataService.cs +++ /dev/null @@ -1,56 +0,0 @@ -using DMS.Application.Interfaces; -using DMS.Core.Interfaces; -using DMS.Core.Models; -using System; -using System.Collections.Generic; -using System.Linq; -using System.Threading.Tasks; - -namespace DMS.Infrastructure.Services; - -public class DeviceDataService : IDeviceDataService -{ - private readonly IRepositoryManager _repositoryManager; - private List _devices; - - public List Devices => _devices; - - public event Action> OnDeviceListChanged; - public event Action OnDeviceIsActiveChanged; - - public DeviceDataService(IRepositoryManager repositoryManager) - { - _repositoryManager = repositoryManager; - _devices = new List(); - } - - public async Task InitializeAsync() - { - await LoadDevicesAsync(); - } - - private async Task LoadDevicesAsync() - { - _devices = (await _repositoryManager.Devices.GetAllAsync()).ToList(); - OnDeviceListChanged?.Invoke(_devices); - } - - // 模拟设备激活状态变更，实际应用中可能由其他服务触发 - public async Task SetDeviceIsActiveAsync(int deviceId, bool isActive) - { - var device = _devices.FirstOrDefault(d => d.Id == deviceId); - if (device != null) - { - device.IsActive = isActive; - OnDeviceIsActiveChanged?.Invoke(device, isActive); - // 实际应用中，这里可能还需要更新数据库 - await _repositoryManager.Devices.UpdateAsync(device); - } - } - - // 模拟设备列表变更，实际应用中可能由其他服务触发 - public async Task RefreshDeviceListAsync() - { - await LoadDevicesAsync(); - } -} \ No newline at end of file diff --git a/DMS.Infrastructure/Services/MqttBackgroundService.cs b/DMS.Infrastructure/Services/MqttBackgroundService.cs index 6c36991..1f78947 100644 --- a/DMS.Infrastructure/Services/MqttBackgroundService.cs +++ b/DMS.Infrastructure/Services/MqttBackgroundService.cs @@ -19,7 +19,6 @@ namespace DMS.Infrastructure.Services; /// public class MqttBackgroundService : BackgroundService { - private readonly IDeviceDataService _deviceDataService; private readonly IRepositoryManager _repositoryManager; private readonly ILogger _logger; @@ -33,9 +32,8 @@ public class MqttBackgroundService : BackgroundService /// /// 构造函数，注入DataServices。 /// - public MqttBackgroundService(IDeviceDataService deviceDataService, IRepositoryManager repositoryManager, ILogger logger) + public MqttBackgroundService(IRepositoryManager repositoryManager, ILogger logger) { - _deviceDataService = deviceDataService; _repositoryManager = repositoryManager; _logger = logger; _mqttClients = new ConcurrentDictionary(); diff --git a/DMS.Infrastructure/Services/OpcUaBackgroundService.cs b/DMS.Infrastructure/Services/OpcUaBackgroundService.cs index ae60d6f..3a7b173 100644 --- a/DMS.Infrastructure/Services/OpcUaBackgroundService.cs +++ b/DMS.Infrastructure/Services/OpcUaBackgroundService.cs @@ -1,4 +1,6 @@ using System.Collections.Concurrent; +using DMS.Application.DTOs; +using DMS.Application.DTOs.Events; using DMS.Core.Enums; using DMS.Core.Models; using Microsoft.Extensions.Hosting; @@ -8,20 +10,22 @@ using Microsoft.Extensions.Logging; using DMS.Application.Interfaces; using DMS.Core.Interfaces; using DMS.Infrastructure.Helper; +using DMS.Infrastructure.Models; namespace DMS.Infrastructure.Services; public class OpcUaBackgroundService : BackgroundService { - private readonly IDeviceDataService _deviceDataService; - private readonly IDataProcessingService _dataProcessingService; + private readonly IDataCenterService _dataCenterService; + + // private readonly IDataProcessingService _dataProcessingService; private readonly ILogger _logger; // 存储 OPC UA 设备，键为设备Id，值为会话对象。 - private readonly ConcurrentDictionary _opcUaDevices; + private readonly ConcurrentDictionary _opcUaDevices = new(); // 存储 OPC UA 会话，键为终结点 URL，值为会话对象。 - private readonly ConcurrentDictionary _opcUaSessions; + private readonly ConcurrentDictionary _opcUaServices; // 存储 OPC UA 订阅，键为终结点 URL，值为订阅对象。 private readonly ConcurrentDictionary _opcUaSubscriptions; @@ -33,7 +37,7 @@ public class OpcUaBackgroundService : BackgroundService private readonly ConcurrentDictionary> _opcUaPollVariablesByDeviceId; // 储存所有要订阅更新的变量，键是Device.Id,值是这个设备所有要轮询的变量 - private readonly ConcurrentDictionary> _opcUaSubVariablesByDeviceId; + private readonly ConcurrentDictionary> _opcUaVariablesByDeviceId; private readonly SemaphoreSlim _reloadSemaphore = new SemaphoreSlim(0); @@ -47,44 +51,50 @@ public class OpcUaBackgroundService : BackgroundService private readonly int _opcUaSubscriptionSamplingIntervalMs = 1000; // 模拟 PollingIntervals，实际应用中可能从配置或数据库加载 - private static readonly Dictionary PollingIntervals = new Dictionary - { - { PollLevelType.TenMilliseconds, TimeSpan.FromMilliseconds((int)PollLevelType.TenMilliseconds) }, - { PollLevelType.HundredMilliseconds, TimeSpan.FromMilliseconds((int)PollLevelType.HundredMilliseconds) }, - { PollLevelType.FiveHundredMilliseconds, TimeSpan.FromMilliseconds((int)PollLevelType.FiveHundredMilliseconds) }, - { PollLevelType.OneSecond, TimeSpan.FromMilliseconds((int)PollLevelType.OneSecond) }, - { PollLevelType.FiveSeconds, TimeSpan.FromMilliseconds((int)PollLevelType.FiveSeconds) }, - { PollLevelType.TenSeconds, TimeSpan.FromMilliseconds((int)PollLevelType.TenSeconds) }, - { PollLevelType.TwentySeconds, TimeSpan.FromMilliseconds((int)PollLevelType.TwentySeconds) }, - { PollLevelType.ThirtySeconds, TimeSpan.FromMilliseconds((int)PollLevelType.ThirtySeconds) }, - { PollLevelType.OneMinute, TimeSpan.FromMilliseconds((int)PollLevelType.OneMinute) }, - { PollLevelType.ThreeMinutes, TimeSpan.FromMilliseconds((int)PollLevelType.ThreeMinutes) }, - { PollLevelType.FiveMinutes, TimeSpan.FromMilliseconds((int)PollLevelType.FiveMinutes) }, - { PollLevelType.TenMinutes, TimeSpan.FromMilliseconds((int)PollLevelType.TenMinutes) }, - { PollLevelType.ThirtyMinutes, TimeSpan.FromMilliseconds((int)PollLevelType.ThirtyMinutes) } - }; + private static readonly Dictionary PollingIntervals + = new Dictionary + { + { PollLevelType.TenMilliseconds, TimeSpan.FromMilliseconds((int)PollLevelType.TenMilliseconds) }, + { PollLevelType.HundredMilliseconds, TimeSpan.FromMilliseconds((int)PollLevelType.HundredMilliseconds) }, + { + PollLevelType.FiveHundredMilliseconds, + TimeSpan.FromMilliseconds((int)PollLevelType.FiveHundredMilliseconds) + }, + { PollLevelType.OneSecond, TimeSpan.FromMilliseconds((int)PollLevelType.OneSecond) }, + { PollLevelType.FiveSeconds, TimeSpan.FromMilliseconds((int)PollLevelType.FiveSeconds) }, + { PollLevelType.TenSeconds, TimeSpan.FromMilliseconds((int)PollLevelType.TenSeconds) }, + { PollLevelType.TwentySeconds, TimeSpan.FromMilliseconds((int)PollLevelType.TwentySeconds) }, + { PollLevelType.ThirtySeconds, TimeSpan.FromMilliseconds((int)PollLevelType.ThirtySeconds) }, + { PollLevelType.OneMinute, TimeSpan.FromMilliseconds((int)PollLevelType.OneMinute) }, + { PollLevelType.ThreeMinutes, TimeSpan.FromMilliseconds((int)PollLevelType.ThreeMinutes) }, + { PollLevelType.FiveMinutes, TimeSpan.FromMilliseconds((int)PollLevelType.FiveMinutes) }, + { PollLevelType.TenMinutes, TimeSpan.FromMilliseconds((int)PollLevelType.TenMinutes) }, + { PollLevelType.ThirtyMinutes, TimeSpan.FromMilliseconds((int)PollLevelType.ThirtyMinutes) } + }; - public OpcUaBackgroundService(IDeviceDataService deviceDataService, IDataCenterService dataCenterService, IDataProcessingService dataProcessingService, ILogger logger) + public OpcUaBackgroundService(IDataCenterService dataCenterService, + ILogger logger) { - _deviceDataService = deviceDataService; - _dataProcessingService = dataProcessingService; + _dataCenterService = dataCenterService; + // _dataProcessingService = dataProcessingService; _logger = logger; - _opcUaDevices = new ConcurrentDictionary(); - _opcUaSessions = new ConcurrentDictionary(); + _opcUaServices = new ConcurrentDictionary(); _opcUaSubscriptions = new ConcurrentDictionary(); _opcUaPollVariablesByNodeId = new ConcurrentDictionary(); _opcUaPollVariablesByDeviceId = new ConcurrentDictionary>(); - _opcUaSubVariablesByDeviceId = new ConcurrentDictionary>(); + _opcUaVariablesByDeviceId = new ConcurrentDictionary>(); - _deviceDataService.OnDeviceListChanged += HandleDeviceListChanged; - _deviceDataService.OnDeviceIsActiveChanged += HandleDeviceIsActiveChanged; + _dataCenterService.DataLoadCompleted += DataLoadCompleted; + } + + private void DataLoadCompleted(object? sender, DataLoadCompletedEventArgs e) + { + _reloadSemaphore.Release(); } protected override async Task ExecuteAsync(CancellationToken stoppingToken) { _logger.LogInformation("OPC UA 后台服务正在启动。"); - _reloadSemaphore.Release(); // Initial trigger to load variables and connect - try { while (!stoppingToken.IsCancellationRequested) @@ -96,7 +106,7 @@ public class OpcUaBackgroundService : BackgroundService break; } - if (_deviceDataService.Devices == null || _deviceDataService.Devices.Count == 0) + if (_dataCenterService.Devices.IsEmpty) { _logger.LogInformation("没有可用的OPC UA设备，等待设备列表更新..."); continue; @@ -113,12 +123,12 @@ public class OpcUaBackgroundService : BackgroundService await SetupOpcUaSubscriptionAsync(stoppingToken); _logger.LogInformation("OPC UA 后台服务已启动。"); - // 持续轮询，直到取消请求或需要重新加载 - while (!stoppingToken.IsCancellationRequested && _reloadSemaphore.CurrentCount == 0) - { - await PollOpcUaVariableOnceAsync(stoppingToken); - await Task.Delay(_opcUaPollIntervalMs, stoppingToken); - } + // // 持续轮询，直到取消请求或需要重新加载 + // while (!stoppingToken.IsCancellationRequested && _reloadSemaphore.CurrentCount == 0) + // { + // await PollOpcUaVariableOnceAsync(stoppingToken); + // await Task.Delay(_opcUaPollIntervalMs, stoppingToken); + // } } } catch (OperationCanceledException) @@ -132,54 +142,47 @@ public class OpcUaBackgroundService : BackgroundService finally { await DisconnectAllOpcUaSessionsAsync(); - _deviceDataService.OnDeviceListChanged -= HandleDeviceListChanged; - _deviceDataService.OnDeviceIsActiveChanged -= HandleDeviceIsActiveChanged; } } - private void HandleDeviceListChanged(List devices) - { - _logger.LogInformation("设备列表已更改。OPC UA 客户端可能需要重新初始化。"); - _reloadSemaphore.Release(); // 触发ExecuteAsync中的全面重新加载 - } - private async void HandleDeviceIsActiveChanged(Device device, bool isActive) - { - if (device.Protocol != ProtocolType.OpcUa) - return; - - _logger.LogInformation($"设备 {device.Name} (ID: {device.Id}) 的IsActive状态改变为 {isActive}。"); - - if (!isActive) - { - // 设备变为非活动状态，断开连接 - if (_opcUaSessions.TryRemove(device.OpcUaServerUrl, out var session)) - { - try - { - if (_opcUaSubscriptions.TryRemove(device.OpcUaServerUrl, out var subscription)) - { - // 删除订阅。 - await subscription.DeleteAsync(true); - _logger.LogInformation($"已删除设备 {device.Name} ({device.OpcUaServerUrl}) 的订阅。"); - } - - if (session.Connected) - { - await session.CloseAsync(); - _logger.LogInformation($"已断开设备 {device.Name} ({device.OpcUaServerUrl}) 的连接。"); - } - } - catch (Exception ex) - { - _logger.LogError(ex, $"断开设备 {device.Name} ({device.OpcUaServerUrl}) 连接时发生错误：{ex.Message}"); - } - } - } - - // 触发重新加载，让LoadVariables和ConnectOpcUaServiceAsync处理设备列表的更新 - _reloadSemaphore.Release(); - } + // private async void HandleDeviceIsActiveChanged(Device device, bool isActive) + // { + // if (device.Protocol != ProtocolType.OpcUa) + // return; + // + // _logger.LogInformation($"设备 {device.Name} (ID: {device.Id}) 的IsActive状态改变为 {isActive}。"); + // + // if (!isActive) + // { + // // 设备变为非活动状态，断开连接 + // if (_opcUaServices.TryRemove(device.OpcUaServerUrl, out var session)) + // { + // try + // { + // if (_opcUaSubscriptions.TryRemove(device.OpcUaServerUrl, out var subscription)) + // { + // // 删除订阅。 + // await subscription.DeleteAsync(true); + // _logger.LogInformation($"已删除设备 {device.Name} ({device.OpcUaServerUrl}) 的订阅。"); + // } + // + // if (session.Connected) + // { + // await session.CloseAsync(); + // _logger.LogInformation($"已断开设备 {device.Name} ({device.OpcUaServerUrl}) 的连接。"); + // } + // } + // catch (Exception ex) + // { + // _logger.LogError(ex, $"断开设备 {device.Name} ({device.OpcUaServerUrl}) 连接时发生错误：{ex.Message}"); + // } + // } + // } + // + // // 触发重新加载，让LoadVariables和ConnectOpcUaServiceAsync处理设备列表的更新 + // _reloadSemaphore.Release(); + // } /// /// 从数据库加载所有活动的 OPC UA 变量，并进行相应的连接和订阅管理。 @@ -190,55 +193,29 @@ public class OpcUaBackgroundService : BackgroundService { _opcUaDevices.Clear(); _opcUaPollVariablesByDeviceId.Clear(); - _opcUaSubVariablesByDeviceId.Clear(); + _opcUaVariablesByDeviceId.Clear(); _opcUaPollVariablesByNodeId.Clear(); _logger.LogInformation("开始加载OPC UA变量...."); - var opcUaDevices = _deviceDataService - .Devices.Where(d => d.Protocol == ProtocolType.OpcUa && d.IsActive == true) + var opcUaDevices = _dataCenterService + .Devices.Values.Where(d => d.Protocol == ProtocolType.OpcUa && d.IsActive == true) .ToList(); - - if (opcUaDevices.Count == 0) - { - _logger.LogInformation("没有找到活动的OPC UA设备。"); - return true; // No active devices, but not an error - } - - int totalPollVariableCount = 0; - int totalSubVariableCount = 0; - + int totalVariableCount = 0; foreach (var opcUaDevice in opcUaDevices) { _opcUaDevices.AddOrUpdate(opcUaDevice.Id, opcUaDevice, (key, oldValue) => opcUaDevice); - //查找设备中所有要轮询的变量 - var dPollList = opcUaDevice.VariableTables?.SelectMany(vt => vt.Variables) - .Where(vd => vd.IsActive == true && - vd.Protocol == ProtocolType.OpcUa && - vd.OpcUaUpdateType == OpcUaUpdateType.OpcUaPoll) - .ToList(); - // 将变量保存到字典中，方便Read后还原 - foreach (var variable in dPollList) - { - _opcUaPollVariablesByNodeId.AddOrUpdate(variable.OpcUaNodeId, variable, - (key, oldValue) => variable); - } - - totalPollVariableCount += dPollList.Count; - _opcUaPollVariablesByDeviceId.AddOrUpdate(opcUaDevice.Id, dPollList, (key, oldValue) => dPollList); - //查找设备中所有要订阅的变量 - var dSubList = opcUaDevice.VariableTables?.SelectMany(vt => vt.Variables) - .Where(vd => vd.IsActive == true && - vd.Protocol == ProtocolType.OpcUa && - vd.OpcUaUpdateType == OpcUaUpdateType.OpcUaSubscription) - .ToList(); - totalSubVariableCount += dSubList.Count; - _opcUaSubVariablesByDeviceId.AddOrUpdate(opcUaDevice.Id, dSubList, (key, oldValue) => dSubList); + var variableDtos = opcUaDevice.VariableTables?.SelectMany(vt => vt.Variables) + .Where(vd => vd.IsActive == true && + vd.Protocol == ProtocolType.OpcUa) + .ToList(); + totalVariableCount += variableDtos.Count; + _opcUaVariablesByDeviceId.AddOrUpdate(opcUaDevice.Id, variableDtos, (key, oldValue) => variableDtos); } _logger.LogInformation( - $"OPC UA 变量加载成功，共加载OPC UA设备：{opcUaDevices.Count}个，轮询变量数：{totalPollVariableCount}，订阅变量数：{totalSubVariableCount}"); + $"OPC UA 变量加载成功，共加载OPC UA设备：{opcUaDevices.Count}个，变量数：{totalVariableCount}"); return true; } catch (Exception e) @@ -274,39 +251,30 @@ public class OpcUaBackgroundService : BackgroundService /// /// 要连接的设备。 /// 取消令牌。 - private async Task ConnectSingleOpcUaDeviceAsync(Device device, CancellationToken stoppingToken = default) + private async Task ConnectSingleOpcUaDeviceAsync(DeviceDto device, CancellationToken stoppingToken = default) { - if (stoppingToken.IsCancellationRequested) - { - return; - } - // Check if already connected - if (_opcUaSessions.TryGetValue(device.OpcUaServerUrl, out var existingSession)) + if (_opcUaServices.TryGetValue(device, out var existOpcUaService)) { - if (existingSession.Connected) + if (existOpcUaService.IsConnected) { _logger.LogInformation($"已连接到 OPC UA 服务器: {device.OpcUaServerUrl}"); return; } - else - { - // Remove disconnected session from dictionary to attempt reconnection - _opcUaSessions.TryRemove(device.OpcUaServerUrl, out _); - } } _logger.LogInformation($"开始连接OPC UA服务器: {device.Name} ({device.OpcUaServerUrl})"); try { - var session = await OpcUaHelper.CreateOpcUaSessionAsync(device.OpcUaServerUrl, stoppingToken); - if (session == null) + OpcUaService opcUaService = new OpcUaService(); + await opcUaService.ConnectAsync(device.OpcUaServerUrl); + if (!opcUaService.IsConnected) { _logger.LogWarning($"创建OPC UA会话失败: {device.OpcUaServerUrl}"); return; // 连接失败，直接返回 } - _opcUaSessions.AddOrUpdate(device.OpcUaServerUrl, session, (key, oldValue) => session); + _opcUaServices.AddOrUpdate(device, opcUaService, (key, oldValue) => opcUaService); _logger.LogInformation($"已连接到OPC UA服务器: {device.Name} ({device.OpcUaServerUrl})"); } catch (Exception e) @@ -315,70 +283,74 @@ public class OpcUaBackgroundService : BackgroundService } } - private async Task PollOpcUaVariableOnceAsync(CancellationToken stoppingToken) - { - try - { - var deviceIdsToPoll = _opcUaPollVariablesByDeviceId.Keys.ToList(); + // private async Task PollOpcUaVariableOnceAsync(CancellationToken stoppingToken) + // { + // try + // { + // var deviceIdsToPoll = _opcUaPollVariablesByDeviceId.Keys.ToList(); + // + // var pollingTasks = deviceIdsToPoll + // .Select(deviceId => PollSingleDeviceVariablesAsync(deviceId, stoppingToken)) + // .ToList(); + // + // await Task.WhenAll(pollingTasks); + // } + // catch (OperationCanceledException) + // { + // _logger.LogInformation("OPC UA 后台服务轮询变量被取消。"); + // } + // catch (Exception ex) + // { + // _logger.LogError(ex, $"OPC UA 后台服务在轮询变量过程中发生错误：{ex.Message}"); + // } + // } - var pollingTasks = deviceIdsToPoll.Select(deviceId => PollSingleDeviceVariablesAsync(deviceId, stoppingToken)).ToList(); - - await Task.WhenAll(pollingTasks); - } - catch (OperationCanceledException) - { - _logger.LogInformation("OPC UA 后台服务轮询变量被取消。"); - } - catch (Exception ex) - { - _logger.LogError(ex, $"OPC UA 后台服务在轮询变量过程中发生错误：{ex.Message}"); - } - } - - /// - /// 轮询单个设备的所有 OPC UA 变量。 - /// - /// 设备的 ID。 - /// 取消令牌。 - private async Task PollSingleDeviceVariablesAsync(int deviceId, CancellationToken stoppingToken) - { - if (stoppingToken.IsCancellationRequested) return; - - if (!_opcUaDevices.TryGetValue(deviceId, out var device) || device.OpcUaServerUrl == null) - { - _logger.LogWarning($"OpcUa轮询变量时，在deviceDic中未找到ID为 {deviceId} 的设备，或其服务器地址为空，请检查！"); - return; - } - - if (!device.IsActive) return; - - if (!_opcUaSessions.TryGetValue(device.OpcUaServerUrl, out var session) || !session.Connected) - { - if (device.IsActive) - { - _logger.LogWarning($"用于 {device.OpcUaServerUrl} 的 OPC UA 会话未连接。正在尝试重新连接..."); - await ConnectSingleOpcUaDeviceAsync(device, stoppingToken); - } - return; - } - - if (!_opcUaPollVariablesByDeviceId.TryGetValue(deviceId, out var variableList) || variableList.Count == 0) - { - return; - } - - foreach (var variable in variableList) - { - if (stoppingToken.IsCancellationRequested) return; - - if (!PollingIntervals.TryGetValue(variable.PollLevel, out var interval) || (DateTime.Now - variable.UpdatedAt) < interval) - { - continue; - } - - await ReadAndProcessOpcUaVariableAsync(session, variable, stoppingToken); - } - } + // /// + // /// 轮询单个设备的所有 OPC UA 变量。 + // /// + // /// 设备的 ID。 + // /// 取消令牌。 + // private async Task PollSingleDeviceVariablesAsync(int deviceId, CancellationToken stoppingToken) + // { + // if (stoppingToken.IsCancellationRequested) return; + // + // if (!_opcUaDevices.TryGetValue(deviceId, out var device) || device.OpcUaServerUrl == null) + // { + // _logger.LogWarning($"OpcUa轮询变量时，在deviceDic中未找到ID为 {deviceId} 的设备，或其服务器地址为空，请检查！"); + // return; + // } + // + // if (!device.IsActive) return; + // + // if (!_opcUaServices.TryGetValue(device.OpcUaServerUrl, out var session) || !session.Connected) + // { + // if (device.IsActive) + // { + // _logger.LogWarning($"用于 {device.OpcUaServerUrl} 的 OPC UA 会话未连接。正在尝试重新连接..."); + // // await ConnectSingleOpcUaDeviceAsync(device, stoppingToken); + // } + // + // return; + // } + // + // if (!_opcUaPollVariablesByDeviceId.TryGetValue(deviceId, out var variableList) || variableList.Count == 0) + // { + // return; + // } + // + // foreach (var variable in variableList) + // { + // if (stoppingToken.IsCancellationRequested) return; + // + // if (!PollingIntervals.TryGetValue(variable.PollLevel, out var interval) || + // (DateTime.Now - variable.UpdatedAt) < interval) + // { + // continue; + // } + // + // await ReadAndProcessOpcUaVariableAsync(session, variable, stoppingToken); + // } + // } /// /// 读取单个 OPC UA 变量并处理其数据。 @@ -386,16 +358,17 @@ public class OpcUaBackgroundService : BackgroundService /// OPC UA 会话。 /// 要读取的变量。 /// 取消令牌。 - private async Task ReadAndProcessOpcUaVariableAsync(Session session, Variable variable, CancellationToken stoppingToken) + private async Task ReadAndProcessOpcUaVariableAsync(Session session, Variable variable, + CancellationToken stoppingToken) { var nodesToRead = new ReadValueIdCollection - { - new ReadValueId - { - NodeId = new NodeId(variable.OpcUaNodeId), - AttributeId = Attributes.Value - } - }; + { + new ReadValueId + { + NodeId = new NodeId(variable.OpcUaNodeId), + AttributeId = Attributes.Value + } + }; try { @@ -441,7 +414,7 @@ public class OpcUaBackgroundService : BackgroundService variable.UpdatedAt = DateTime.Now; // Console.WriteLine($"OpcUa后台服务轮询变量：{variable.Name},值：{variable.DataValue}"); // 将更新后的数据推入处理队列。 - await _dataProcessingService.EnqueueAsync(variable); + // await _dataProcessingService.EnqueueAsync(variable); } catch (Exception ex) { @@ -455,78 +428,29 @@ public class OpcUaBackgroundService : BackgroundService /// 取消令牌。 private async Task SetupOpcUaSubscriptionAsync(CancellationToken stoppingToken) { - if (stoppingToken.IsCancellationRequested) + foreach (var opcUaServiceKayValuePair in _opcUaServices) { - return; - } + var device = opcUaServiceKayValuePair.Key; + var opcUaService = opcUaServiceKayValuePair.Value; - var setupSubscriptionTasks = new List(); - - foreach (var deviceId in _opcUaSubVariablesByDeviceId.Keys.ToList()) - { - setupSubscriptionTasks.Add(Task.Run(async () => + if (_opcUaVariablesByDeviceId.TryGetValue(device.Id, out var opcUaVariables)) { - if (stoppingToken.IsCancellationRequested) + var variableGroup = opcUaVariables.GroupBy(variable => variable.PollLevel); + foreach (var vGroup in variableGroup) { - return; // 任务被取消，退出循环 + var pollLevelType = vGroup.Key; + var opcUaNodes + = vGroup.Select(variableDto => new OpcUaNode() { NodeId = variableDto.OpcUaNodeId }) + .ToList(); + opcUaService.SubscribeToNode(opcUaNodes,HandleDataChanged); } - - var device = _deviceDataService.Devices.FirstOrDefault(d => d.Id == deviceId); - if (device == null) - { - _logger.LogWarning($"未找到ID为 {deviceId} 的设备，无法设置订阅。"); - return; - } - - Subscription subscription = null; - // 得到session - if (!_opcUaSessions.TryGetValue(device.OpcUaServerUrl, out var session)) - { - _logger.LogInformation($"从OpcUa会话字典中获取会话失败： {device.OpcUaServerUrl} "); - return; - } - - // 判断设备是否已经添加了订阅 - if (_opcUaSubscriptions.TryGetValue(device.OpcUaServerUrl, out subscription)) - { - _logger.LogInformation($"OPC UA 终结点 {device.OpcUaServerUrl} 已存在订阅。"); - } - else - { - subscription = new Subscription(session.DefaultSubscription); - subscription.PublishingInterval = _opcUaSubscriptionPublishingIntervalMs; // 发布间隔（毫秒） - session.AddSubscription(subscription); - subscription.Create(); - _opcUaSubscriptions.AddOrUpdate(device.OpcUaServerUrl, subscription, - (key, oldValue) => subscription); - } - - // 将变量添加到订阅 - if (_opcUaSubVariablesByDeviceId.TryGetValue(deviceId, out var variablesToSubscribe)) - { - foreach (Variable variable in variablesToSubscribe) - { - // 7. 创建监控项并添加到订阅中。 - MonitoredItem monitoredItem = new MonitoredItem(subscription.DefaultItem); - monitoredItem.DisplayName = variable.Name; - monitoredItem.StartNodeId = new NodeId(variable.OpcUaNodeId); // 设置要监控的节点 ID - monitoredItem.AttributeId = Attributes.Value; // 监控节点的值属性 - monitoredItem.SamplingInterval = _opcUaSubscriptionSamplingIntervalMs; // 采样间隔（毫秒） - monitoredItem.QueueSize = 1; // 队列大小 - monitoredItem.DiscardOldest = true; // 丢弃最旧的数据 - // 注册数据变化通知事件。 - monitoredItem.Notification += (sender, e) => OnSubNotification(variable, monitoredItem, e); - - subscription.AddItem(monitoredItem); - } - - subscription.ApplyChanges(); // 应用更改 - _logger.LogInformation($"设备: {device.Name}, 添加了 {variablesToSubscribe.Count} 个订阅变量。"); - } - })); + } } + } - await Task.WhenAll(setupSubscriptionTasks); + private void HandleDataChanged(OpcUaNode opcUaNode) + { + } /// @@ -538,7 +462,6 @@ public class OpcUaBackgroundService : BackgroundService private async void OnSubNotification(Variable variable, MonitoredItem monitoredItem, MonitoredItemNotificationEventArgs e) { - foreach (var value in monitoredItem.DequeueValues()) { _logger.LogInformation( @@ -555,28 +478,20 @@ public class OpcUaBackgroundService : BackgroundService /// private async Task DisconnectAllOpcUaSessionsAsync() { - if (_opcUaSessions.IsEmpty) + if (_opcUaServices.IsEmpty) return; - _logger.LogInformation("正在断开所有 OPC UA 会话..."); var closeTasks = new List(); - foreach (var endpointUrl in _opcUaSessions.Keys.ToList()) + foreach (var device in _opcUaServices.Keys.ToList()) { closeTasks.Add(Task.Run(async () => { - _logger.LogInformation($"正在断开 OPC UA 会话: {endpointUrl}"); - if (_opcUaSessions.TryRemove(endpointUrl, out var session)) + _logger.LogInformation($"正在断开 OPC UA 会话: {device.Name}"); + if (_opcUaServices.TryRemove(device, out var opcUaService)) { - if (_opcUaSubscriptions.TryRemove(endpointUrl, out var subscription)) - { - // 删除订阅。 - await subscription.DeleteAsync(true); - } - - // 关闭会话。 - await session.CloseAsync(); - _logger.LogInformation($"已从 OPC UA 服务器断开连接: {endpointUrl}"); + await opcUaService.DisconnectAsync(); + _logger.LogInformation($"已从 OPC UA 服务器断开连接: {device.Name}"); } })); } diff --git a/DMS.Infrastructure/Services/S7BackgroundService.cs b/DMS.Infrastructure/Services/S7BackgroundService.cs index ffe54c2..f193557 100644 --- a/DMS.Infrastructure/Services/S7BackgroundService.cs +++ b/DMS.Infrastructure/Services/S7BackgroundService.cs @@ -17,8 +17,6 @@ namespace DMS.Infrastructure.Services; /// public class S7BackgroundService : BackgroundService { - // 数据服务实例，用于访问和操作应用程序数据，如设备配置。 - private readonly IDeviceDataService _deviceDataService; // 数据处理服务实例，用于将读取到的数据推入处理队列。 private readonly IDataProcessingService _dataProcessingService; @@ -69,20 +67,14 @@ public class S7BackgroundService : BackgroundService /// 设备数据服务实例。 /// 数据处理服务实例。 /// 日志记录器实例。 - public S7BackgroundService(IDeviceDataService deviceDataService, IDataProcessingService dataProcessingService, ILogger logger) + public S7BackgroundService( IDataProcessingService dataProcessingService, ILogger logger) { - _deviceDataService = deviceDataService; _dataProcessingService = dataProcessingService; _logger = logger; _s7Devices = new ConcurrentDictionary(); _s7PollVariablesByDeviceId = new ConcurrentDictionary>(); _s7PlcClientsByIp = new ConcurrentDictionary(); _s7VariablesById = new(); - - // 订阅设备列表变更事件，以便在设备配置更新时重新加载。 - _deviceDataService.OnDeviceListChanged += HandleDeviceListChanged; - // 订阅单个设备IsActive状态变更事件 - _deviceDataService.OnDeviceIsActiveChanged += HandleDeviceIsActiveChanged; } protected override async Task ExecuteAsync(CancellationToken stoppingToken) @@ -101,18 +93,18 @@ public class S7BackgroundService : BackgroundService break; } - if (_deviceDataService.Devices == null || _deviceDataService.Devices.Count == 0) - { - _logger.LogInformation("没有可用的S7设备，等待设备列表更新..."); - continue; - } - - var isLoaded = LoadVariables(); - if (!isLoaded) - { - _logger.LogInformation("加载变量过程中发生了错误，停止后面的操作。"); - continue; - } + // if (_deviceDataService.Devices == null || _deviceDataService.Devices.Count == 0) + // { + // _logger.LogInformation("没有可用的S7设备，等待设备列表更新..."); + // continue; + // } + // + // var isLoaded = LoadVariables(); + // if (!isLoaded) + // { + // _logger.LogInformation("加载变量过程中发生了错误，停止后面的操作。"); + // continue; + // } await ConnectS7Service(stoppingToken); _logger.LogInformation("S7后台服务开始轮询变量...."); @@ -391,54 +383,54 @@ public class S7BackgroundService : BackgroundService /// /// 加载变量 /// - private bool LoadVariables() - { - try - { - _s7Devices.Clear(); - _s7PollVariablesByDeviceId.Clear(); - _s7VariablesById.Clear(); // 确保在重新加载变量时清空此字典 - - _logger.LogInformation("开始加载S7变量...."); - var s7Devices = _deviceDataService - .Devices.Where(d => d.IsActive == true && d.Protocol == ProtocolType.S7) - .ToList(); // 转换为列表，避免多次枚举 - - int totalVariableCount = 0; - foreach (var device in s7Devices) - { - // device.IsRuning = true; - _s7Devices.AddOrUpdate(device.Id, device, (key, oldValue) => device); - - // 过滤出当前设备和S7协议相关的变量。 - var deviceS7Variables = device.VariableTables - .Where(vt => vt.Protocol == ProtocolType.S7 && vt.IsActive && vt.Variables != null) - .SelectMany(vt => vt.Variables) - .Where(vd => vd.IsActive == true) - .ToList(); // 转换为列表，避免多次枚举 - - // 将变量存储到字典中，方便以后通过ID快速查找 - foreach (var s7Variable in deviceS7Variables) - _s7VariablesById[s7Variable.Id] = s7Variable; - - totalVariableCount += deviceS7Variables.Count; // 使用 Count 属性 - _s7PollVariablesByDeviceId.AddOrUpdate(device.Id, deviceS7Variables, (key, oldValue) => deviceS7Variables); - } - - if (totalVariableCount == 0) - { - return false; - } - - _logger.LogInformation($"S7变量加载成功，共加载S7设备：{s7Devices.Count}个，变量数：{totalVariableCount}"); - return true; - } - catch (Exception e) - { - _logger.LogError(e, $"S7后台服务加载变量时发生了错误:{e.Message}"); - return false; - } - } + // private bool LoadVariables() + // { + // // try + // // { + // // _s7Devices.Clear(); + // // _s7PollVariablesByDeviceId.Clear(); + // // _s7VariablesById.Clear(); // 确保在重新加载变量时清空此字典 + // // + // // _logger.LogInformation("开始加载S7变量...."); + // // var s7Devices = _deviceDataService + // // .Devices.Where(d => d.IsActive == true && d.Protocol == ProtocolType.S7) + // // .ToList(); // 转换为列表，避免多次枚举 + // // + // // int totalVariableCount = 0; + // // foreach (var device in s7Devices) + // // { + // // // device.IsRuning = true; + // // _s7Devices.AddOrUpdate(device.Id, device, (key, oldValue) => device); + // // + // // // 过滤出当前设备和S7协议相关的变量。 + // // var deviceS7Variables = device.VariableTables + // // .Where(vt => vt.Protocol == ProtocolType.S7 && vt.IsActive && vt.Variables != null) + // // .SelectMany(vt => vt.Variables) + // // .Where(vd => vd.IsActive == true) + // // .ToList(); // 转换为列表，避免多次枚举 + // // + // // // 将变量存储到字典中，方便以后通过ID快速查找 + // // foreach (var s7Variable in deviceS7Variables) + // // _s7VariablesById[s7Variable.Id] = s7Variable; + // // + // // totalVariableCount += deviceS7Variables.Count; // 使用 Count 属性 + // // _s7PollVariablesByDeviceId.AddOrUpdate(device.Id, deviceS7Variables, (key, oldValue) => deviceS7Variables); + // // } + // // + // // if (totalVariableCount == 0) + // // { + // // return false; + // // } + // // + // // _logger.LogInformation($"S7变量加载成功，共加载S7设备：{s7Devices.Count}个，变量数：{totalVariableCount}"); + // // return true; + // // } + // // catch (Exception e) + // // { + // // _logger.LogError(e, $"S7后台服务加载变量时发生了错误:{e.Message}"); + // // return false; + // // } + // } /// /// 关闭所有PLC的连接 diff --git a/DMS.WPF/App.xaml.cs b/DMS.WPF/App.xaml.cs index dd71d0b..ffee7b8 100644 --- a/DMS.WPF/App.xaml.cs +++ b/DMS.WPF/App.xaml.cs @@ -138,6 +138,8 @@ public partial class App : System.Windows.Application }); // 注册数据处理服务和处理器 + + services.AddHostedService(); services.AddSingleton(); services.AddHostedService(provider => (DataProcessingService)provider.GetRequiredService()); services.AddSingleton();