From 3a3ed7a2640c89a67f730966f640614df8b7bff6 Mon Sep 17 00:00:00 2001 From: "David P.G" Date: Thu, 24 Jul 2025 22:18:22 +0800 Subject: [PATCH] =?UTF-8?q?=E4=B8=B4=E6=97=B6=E6=8F=90=E4=BA=A4?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- DMS.Core/Enums/ProtocolType.cs | 3 +- DMS.Core/Models/Variable.cs | 1 + DMS.WPF/App.xaml.cs | 6 +- DMS.WPF/Services/MqttBackgroundService.cs | 345 ------------- DMS.WPF/Services/OpcUaBackgroundService.cs | 565 --------------------- DMS.WPF/Services/S7BackgroundService.cs | 448 ---------------- 6 files changed, 5 insertions(+), 1363 deletions(-) delete mode 100644 DMS.WPF/Services/MqttBackgroundService.cs delete mode 100644 DMS.WPF/Services/OpcUaBackgroundService.cs delete mode 100644 DMS.WPF/Services/S7BackgroundService.cs diff --git a/DMS.Core/Enums/ProtocolType.cs b/DMS.Core/Enums/ProtocolType.cs index c4ada39..7823873 100644 --- a/DMS.Core/Enums/ProtocolType.cs +++ b/DMS.Core/Enums/ProtocolType.cs @@ -16,5 +16,6 @@ public enum ProtocolType /// /// Modbus TCP 协议。 /// - ModbusTcp + ModbusTcp, + OpcUA } \ No newline at end of file diff --git a/DMS.Core/Models/Variable.cs b/DMS.Core/Models/Variable.cs index 406f951..276fd0b 100644 --- a/DMS.Core/Models/Variable.cs +++ b/DMS.Core/Models/Variable.cs @@ -143,4 +143,5 @@ public class Variable public PollLevelType PollLevelType { get; set; } public DateTime UpdateTime { get; set; } + public OpcUaUpdateType OpcUaUpdateType { get; set; } } \ No newline at end of file diff --git a/DMS.WPF/App.xaml.cs b/DMS.WPF/App.xaml.cs index ee4202f..a806bc5 100644 --- a/DMS.WPF/App.xaml.cs +++ b/DMS.WPF/App.xaml.cs @@ -117,10 +117,8 @@ public partial class App : System.Windows.Application services.AddSingleton(); services.AddHostedService(); services.AddHostedService(); - services.AddSingleton(); - services.AddSingleton(); - services.AddHostedService(provider => provider.GetRequiredService()); - services.AddSingleton(); + services.AddHostedService(); + // 注册 AutoMapper services.AddAutoMapper(typeof(App).Assembly); diff --git a/DMS.WPF/Services/MqttBackgroundService.cs b/DMS.WPF/Services/MqttBackgroundService.cs deleted file mode 100644 index bc1a4cc..0000000 --- a/DMS.WPF/Services/MqttBackgroundService.cs +++ /dev/null @@ -1,345 +0,0 @@ -using System.Collections.Concurrent; -using System.Text; -using System.Threading.Channels; -using DMS.Data.Repositories; -using DMS.Helper; -using DMS.WPF.Models; -using Microsoft.Extensions.Hosting; -using MQTTnet; -using MQTTnet.Client; -using MQTTnet.Client.Connecting; -using MQTTnet.Client.Disconnecting; -using MQTTnet.Client.Options; - -namespace DMS.Services -{ - /// - /// MQTT后台服务，继承自BackgroundService，用于在后台管理MQTT连接和数据发布。 - /// - public class MqttBackgroundService : BackgroundService - { - private readonly DataServices _dataServices; - private readonly MqttRepository _mqttRepository; - - private readonly ConcurrentDictionary _mqttClients; - private readonly ConcurrentDictionary _mqttConfigDic; - private readonly ConcurrentDictionary _reconnectAttempts; - - private readonly SemaphoreSlim _reloadSemaphore = new(0); - private readonly Channel _messageChannel; - - /// - /// 构造函数，注入DataServices。 - /// - public MqttBackgroundService(DataServices dataServices, MqttRepository mqttRepository) - { - _dataServices = dataServices; - _mqttRepository = mqttRepository; - _mqttClients = new ConcurrentDictionary(); - _mqttConfigDic = new ConcurrentDictionary(); - _reconnectAttempts = new ConcurrentDictionary(); - _messageChannel = Channel.CreateUnbounded(); - - _dataServices.OnMqttListChanged += HandleMqttListChanged; - } - - /// - /// 将待发送的变量数据异步推入队列。 - /// - /// 包含MQTT别名和变量数据的对象。 - public async Task SendVariableAsync(VariableMqtt data) - { - await _messageChannel.Writer.WriteAsync(data); - } - - protected override async Task ExecuteAsync(CancellationToken stoppingToken) - { - NlogHelper.Info("Mqtt后台服务正在启动。"); - _reloadSemaphore.Release(); - - var processQueueTask = ProcessMessageQueueAsync(stoppingToken); - - try - { - while (!stoppingToken.IsCancellationRequested) - { - await _reloadSemaphore.WaitAsync(stoppingToken); - - if (stoppingToken.IsCancellationRequested) break; - - if (_dataServices.Mqtts == null || _dataServices.Mqtts.Count == 0) - { - NlogHelper.Info("没有可用的Mqtt配置，等待Mqtt列表更新..."); - continue; - } - - if (!LoadMqttConfigurations()) - { - NlogHelper.Info("加载Mqtt配置过程中发生了错误，停止后面的操作。"); - continue; - } - - await ConnectMqttList(stoppingToken); - NlogHelper.Info("Mqtt后台服务已启动。"); - - while (!stoppingToken.IsCancellationRequested && _reloadSemaphore.CurrentCount == 0) - { - await Task.Delay(1000, stoppingToken); - } - } - } - catch (OperationCanceledException) - { - NlogHelper.Info("Mqtt后台服务正在停止。"); - } - catch (Exception e) - { - NlogHelper.Error($"Mqtt后台服务运行中发生了错误:{e.Message}", e); - } - finally - { - _messageChannel.Writer.Complete(); - await processQueueTask; // 等待消息队列处理完成 - await DisconnectAll(stoppingToken); - _dataServices.OnMqttListChanged -= HandleMqttListChanged; - NlogHelper.Info("Mqtt后台服务已停止。"); - } - } - - private async Task ProcessMessageQueueAsync(CancellationToken stoppingToken) - { - NlogHelper.Info("MQTT消息发送队列处理器已启动。"); - var batch = new List(); - var timer = new PeriodicTimer(TimeSpan.FromSeconds(1)); - - while (!stoppingToken.IsCancellationRequested) - { - try - { - // 等待信号：要么是新消息到达，要么是1秒定时器触发 - await Task.WhenAny( - _messageChannel.Reader.WaitToReadAsync(stoppingToken).AsTask(), - timer.WaitForNextTickAsync(stoppingToken).AsTask() - ); - - // 尽可能多地读取消息，直到达到批次上限 - while (batch.Count < 50 && _messageChannel.Reader.TryRead(out var message)) - { - batch.Add(message); - } - - if (batch.Any()) - { - await SendBatchAsync(batch, stoppingToken); - batch.Clear(); - } - } - catch (OperationCanceledException) - { - NlogHelper.Info("MQTT消息发送队列处理器已停止。"); - break; - } - catch (Exception ex) - { - NlogHelper.Error($"处理MQTT消息队列时发生错误: {ex.Message}", ex); - await Task.Delay(5000, stoppingToken); // 发生未知错误时，延迟一段时间再重试 - } - } - } - - private async Task SendBatchAsync(List batch, CancellationToken stoppingToken) - { - NlogHelper.Info($"准备发送一批 {batch.Count} 条MQTT消息。"); - // 按MQTT服务器ID进行分组 - var groupedByMqtt = batch.GroupBy(vm => vm.MqttId); - - foreach (var group in groupedByMqtt) - { - var mqttId = group.Key; - if (!_mqttClients.TryGetValue(mqttId, out var client) || !client.IsConnected) - { - NlogHelper.Warn($"MQTT客户端 (ID: {mqttId}) 未连接或不存在，跳过 {group.Count()} 条消息。"); - continue; - } - - var messages = group.Select(vm => new MqttApplicationMessageBuilder() - .WithTopic(vm.Mqtt.PublishTopic) - .WithPayload(vm.Variable?.DataValue ?? string.Empty) - .WithQualityOfServiceLevel(MQTTnet.Protocol.MqttQualityOfServiceLevel.AtMostOnce) - .Build()) - .ToList(); - try - { - foreach (var message in messages) - { - await client.PublishAsync(message, stoppingToken); - } - NlogHelper.Info($"成功向MQTT客户端 (ID: {mqttId}) 发送 {messages.Count} 条消息。"); - } - catch (Exception ex) - { - NlogHelper.Error($"向MQTT客户端 (ID: {mqttId}) 批量发送消息时发生错误: {ex.Message}", ex); - } - } - } - - private async Task DisconnectAll(CancellationToken stoppingToken) - { - var disconnectTasks = _mqttClients.Values.Select(client => client.DisconnectAsync(new MqttClientDisconnectOptions(), stoppingToken)); - await Task.WhenAll(disconnectTasks); - _mqttClients.Clear(); - } - - private bool LoadMqttConfigurations() - { - try - { - NlogHelper.Info("开始加载Mqtt配置文件..."); - _mqttConfigDic.Clear(); - var mqttConfigList = _dataServices.Mqtts.Where(m => m.IsActive).ToList(); - - foreach (var mqtt in mqttConfigList) - { - mqtt.OnMqttIsActiveChanged += OnMqttIsActiveChangedHandler; - _mqttConfigDic.TryAdd(mqtt.Id, mqtt); - mqtt.ConnectMessage = "配置加载成功."; - } - - NlogHelper.Info($"Mqtt配置文件加载成功，开启的Mqtt客户端：{mqttConfigList.Count}个。"); - return true; - } - catch (Exception e) - { - NotificationHelper.ShowError($"Mqtt后台服务在加载配置的过程中发生了错误:{e.Message}", e); - return false; - } - } - - private async void OnMqttIsActiveChangedHandler(Mqtt mqtt) - { - try - { - if (mqtt.IsActive) - { - await ConnectMqtt(mqtt, CancellationToken.None); - } - else - { - if (_mqttClients.TryRemove(mqtt.Id, out var client) && client.IsConnected) - { - await client.DisconnectAsync(); - NlogHelper.Info($"{mqtt.Name}的客户端，与服务器断开连接."); - } - mqtt.IsConnected = false; - mqtt.ConnectMessage = "已断开连接."; - } - - await _mqttRepository.UpdateAsync(mqtt); - NotificationHelper.ShowSuccess($"Mqtt客户端：{mqtt.Name},激活状态修改成功。"); - } - catch (Exception e) - { - NotificationHelper.ShowError($"{mqtt.Name}客户端，开启或关闭的过程中发生了错误:{e.Message}", e); - } - } - - private async Task ConnectMqttList(CancellationToken stoppingToken) - { - var connectTasks = _mqttConfigDic.Values.Select(mqtt => ConnectMqtt(mqtt, stoppingToken)); - await Task.WhenAll(connectTasks); - } - - private async Task ConnectMqtt(Mqtt mqtt, CancellationToken stoppingToken) - { - if (_mqttClients.TryGetValue(mqtt.Id, out var existingClient) && existingClient.IsConnected) - { - NlogHelper.Info($"{mqtt.Name}的Mqtt服务器连接已存在。"); - return; - } - - NlogHelper.Info($"开始连接：{mqtt.Name}的服务器..."); - mqtt.ConnectMessage = "开始连接服务器..."; - - var factory = new MqttFactory(); - var client = factory.CreateMqttClient(); - - var options = new MqttClientOptionsBuilder() - .WithClientId(mqtt.ClientID) - .WithTcpServer(mqtt.Host, mqtt.Port) - .WithCredentials(mqtt.UserName, mqtt.PassWord) - .WithCleanSession() - .Build(); - - client.UseConnectedHandler(async e => await HandleConnected(e, client, mqtt)); - client.UseApplicationMessageReceivedHandler(e => HandleMessageReceived(e, mqtt)); - client.UseDisconnectedHandler(async e => await HandleDisconnected(e, options, client, mqtt, stoppingToken)); - - try - { - await client.ConnectAsync(options, stoppingToken); - _mqttClients.AddOrUpdate(mqtt.Id, client, (id, oldClient) => client); - } - catch (Exception ex) - { - mqtt.ConnectMessage = $"连接MQTT服务器失败: {ex.Message}"; - NlogHelper.Error($"连接MQTT服务器失败: {mqtt.Name}", ex); - } - } - - private static void HandleMessageReceived(MqttApplicationMessageReceivedEventArgs e, Mqtt mqtt) - { - var topic = e.ApplicationMessage.Topic; - var payload = Encoding.UTF8.GetString(e.ApplicationMessage.Payload); - NlogHelper.Info($"MQTT客户端 {mqtt.Name} 收到消息: 主题={topic}, 消息={payload}"); - } - - private async Task HandleDisconnected(MqttClientDisconnectedEventArgs args, IMqttClientOptions options, IMqttClient client, Mqtt mqtt, CancellationToken stoppingToken) - { - NotificationHelper.ShowWarn($"与MQTT服务器断开连接: {mqtt.Name}"); - mqtt.ConnectMessage = "断开连接."; - mqtt.IsConnected = false; - - if (stoppingToken.IsCancellationRequested || !mqtt.IsActive) return; - - _reconnectAttempts.AddOrUpdate(mqtt.Id, 1, (id, count) => count + 1); - var attempt = _reconnectAttempts[mqtt.Id]; - - var delay = TimeSpan.FromSeconds(Math.Min(60, Math.Pow(2, attempt))); - NlogHelper.Info($"与MQTT服务器：{mqtt.Name} 的连接已断开。将在 {delay.TotalSeconds} 秒后尝试第 {attempt} 次重新连接..."); - mqtt.ConnectMessage = $"连接已断开，{delay.TotalSeconds}秒后尝试重连..."; - - await Task.Delay(delay, stoppingToken); - - try - { - mqtt.ConnectMessage = "开始重新连接服务器..."; - await client.ConnectAsync(options, stoppingToken); - } - catch (Exception ex) - { - mqtt.ConnectMessage = "重新连接失败."; - NlogHelper.Error($"重新与Mqtt服务器连接失败: {mqtt.Name}", ex); - } - } - - private async Task HandleConnected(MqttClientConnectedEventArgs args, IMqttClient client, Mqtt mqtt) - { - _reconnectAttempts.TryRemove(mqtt.Id, out _); - NotificationHelper.ShowSuccess($"已连接到MQTT服务器: {mqtt.Name}"); - mqtt.IsConnected = true; - mqtt.ConnectMessage = "连接成功."; - - if (!string.IsNullOrEmpty(mqtt.SubTopic)) - { - await client.SubscribeAsync(new MqttTopicFilterBuilder().WithTopic(mqtt.SubTopic).Build()); - NlogHelper.Info($"MQTT客户端 {mqtt.Name} 已订阅主题: {mqtt.SubTopic}"); - } - } - - private void HandleMqttListChanged(List mqtts) - { - NlogHelper.Info("Mqtt列表发生了变化，正在重新加载数据..."); - _reloadSemaphore.Release(); - } - } -} \ No newline at end of file diff --git a/DMS.WPF/Services/OpcUaBackgroundService.cs b/DMS.WPF/Services/OpcUaBackgroundService.cs deleted file mode 100644 index e861434..0000000 --- a/DMS.WPF/Services/OpcUaBackgroundService.cs +++ /dev/null @@ -1,565 +0,0 @@ -using System.Collections.Concurrent; -using DMS.Core.Enums; -using DMS.Helper; -using DMS.WPF.Models; -using Microsoft.Extensions.Hosting; -using Opc.Ua; -using Opc.Ua.Client; - -namespace DMS.Services -{ - public class OpcUaBackgroundService : BackgroundService - { - private readonly DataServices _dataServices; - private readonly IDataProcessingService _dataProcessingService; - - // 存储 OPC UA 设备，键为设备Id，值为会话对象。 - private readonly ConcurrentDictionary _opcUaDevices; - - // 存储 OPC UA 会话，键为终结点 URL，值为会话对象。 - private readonly ConcurrentDictionary _opcUaSessions; - - // 存储 OPC UA 订阅，键为终结点 URL，值为订阅对象。 - private readonly ConcurrentDictionary _opcUaSubscriptions; - - // 存储活动的 OPC UA 变量，键为变量的OpcNodeId - private readonly ConcurrentDictionary _opcUaPollVariablesByNodeId; - - // 储存所有要轮询更新的变量，键是Device.Id,值是这个设备所有要轮询的变量 - private readonly ConcurrentDictionary> _opcUaPollVariablesByDeviceId; - - // 储存所有要订阅更新的变量，键是Device.Id,值是这个设备所有要轮询的变量 - private readonly ConcurrentDictionary> _opcUaSubVariablesByDeviceId; - - private readonly SemaphoreSlim _reloadSemaphore = new SemaphoreSlim(0); - - // OPC UA 轮询间隔（毫秒） - private readonly int _opcUaPollIntervalMs = 100; - - // OPC UA 订阅发布间隔（毫秒） - private readonly int _opcUaSubscriptionPublishingIntervalMs = 1000; - - // OPC UA 订阅采样间隔（毫秒） - private readonly int _opcUaSubscriptionSamplingIntervalMs = 1000; - - public OpcUaBackgroundService(DataServices dataServices, IDataProcessingService dataProcessingService) - { - _dataServices = dataServices; - _dataProcessingService = dataProcessingService; - _opcUaDevices = new ConcurrentDictionary(); - _opcUaSessions = new ConcurrentDictionary(); - _opcUaSubscriptions = new ConcurrentDictionary(); - _opcUaPollVariablesByNodeId = new ConcurrentDictionary(); - _opcUaPollVariablesByDeviceId = new ConcurrentDictionary>(); - _opcUaSubVariablesByDeviceId = new ConcurrentDictionary>(); - - _dataServices.OnDeviceListChanged += HandleDeviceListChanged; - _dataServices.OnDeviceIsActiveChanged += HandleDeviceIsActiveChanged; - } - - protected override async Task ExecuteAsync(CancellationToken stoppingToken) - { - NlogHelper.Info("OPC UA 后台服务正在启动。"); - _reloadSemaphore.Release(); // Initial trigger to load variables and connect - - try - { - while (!stoppingToken.IsCancellationRequested) - { - await _reloadSemaphore.WaitAsync(stoppingToken); // Wait for a reload signal - - if (stoppingToken.IsCancellationRequested) - { - break; - } - - if (_dataServices.Devices == null || _dataServices.Devices.Count == 0) - { - NlogHelper.Info("没有可用的OPC UA设备，等待设备列表更新..."); - continue; - } - - var isLoaded = LoadVariables(); - if (!isLoaded) - { - NlogHelper.Info("加载变量过程中发生了错误，停止后面的操作。"); - continue; - } - - await ConnectOpcUaServiceAsync(stoppingToken); - await SetupOpcUaSubscriptionAsync(stoppingToken); - NlogHelper.Info("OPC UA 后台服务已启动。"); - - // 持续轮询，直到取消请求或需要重新加载 - while (!stoppingToken.IsCancellationRequested && _reloadSemaphore.CurrentCount == 0) - { - await PollOpcUaVariableOnceAsync(stoppingToken); - await Task.Delay(_opcUaPollIntervalMs, stoppingToken); - } - } - } - catch (OperationCanceledException) - { - NlogHelper.Info("OPC UA 后台服务已停止。"); - } - catch (Exception e) - { - NlogHelper.Error($"OPC UA 后台服务运行中发生了错误:{e.Message}", e); - } - finally - { - await DisconnectAllOpcUaSessionsAsync(); - _dataServices.OnDeviceListChanged -= HandleDeviceListChanged; - _dataServices.OnDeviceIsActiveChanged -= HandleDeviceIsActiveChanged; - } - } - - private void HandleDeviceListChanged(List devices) - { - NlogHelper.Info("设备列表已更改。OPC UA 客户端可能需要重新初始化。"); - _reloadSemaphore.Release(); // 触发ExecuteAsync中的全面重新加载 - } - - private async void HandleDeviceIsActiveChanged(Device device, bool isActive) - { - if (device.ProtocolType != ProtocolType.OpcUA) - return; - - NlogHelper.Info($"设备 {device.Name} (ID: {device.Id}) 的IsActive状态改变为 {isActive}。"); - - if (!isActive) - { - // 设备变为非活动状态，断开连接 - if (_opcUaSessions.TryRemove(device.OpcUaEndpointUrl, out var session)) - { - try - { - if (_opcUaSubscriptions.TryRemove(device.OpcUaEndpointUrl, out var subscription)) - { - // 删除订阅。 - await subscription.DeleteAsync(true); - NlogHelper.Info($"已删除设备 {device.Name} ({device.OpcUaEndpointUrl}) 的订阅。"); - } - - if (session.Connected) - { - await session.CloseAsync(); - NotificationHelper.ShowSuccess($"已断开设备 {device.Name} ({device.OpcUaEndpointUrl}) 的连接。"); - } - } - catch (Exception ex) - { - NlogHelper.Error($"断开设备 {device.Name} ({device.OpcUaEndpointUrl}) 连接时发生错误：{ex.Message}", ex); - } - } - } - - // 触发重新加载，让LoadVariables和ConnectOpcUaServiceAsync处理设备列表的更新 - _reloadSemaphore.Release(); - } - - /// - /// 从数据库加载所有活动的 OPC UA 变量，并进行相应的连接和订阅管理。 - /// - private bool LoadVariables() - { - try - { - _opcUaDevices.Clear(); - _opcUaPollVariablesByDeviceId.Clear(); - _opcUaSubVariablesByDeviceId.Clear(); - _opcUaPollVariablesByNodeId.Clear(); - - NlogHelper.Info("开始加载OPC UA变量...."); - var opcUaDevices = _dataServices - .Devices.Where(d => d.ProtocolType == ProtocolType.OpcUA && d.IsActive == true) - .ToList(); - - if (opcUaDevices.Count == 0) - { - NlogHelper.Info("没有找到活动的OPC UA设备。"); - return true; // No active devices, but not an error - } - - int totalPollVariableCount = 0; - int totalSubVariableCount = 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.ProtocolType == 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.ProtocolType == ProtocolType.OpcUA && - vd.OpcUaUpdateType == OpcUaUpdateType.OpcUaSubscription) - .ToList(); - totalSubVariableCount += dSubList.Count; - _opcUaSubVariablesByDeviceId.AddOrUpdate(opcUaDevice.Id, dSubList, (key, oldValue) => dSubList); - } - - NlogHelper.Info( - $"OPC UA 变量加载成功，共加载OPC UA设备：{opcUaDevices.Count}个，轮询变量数：{totalPollVariableCount}，订阅变量数：{totalSubVariableCount}"); - return true; - } - catch (Exception e) - { - NotificationHelper.ShowError($"加载OPC UA变量的过程中发生了错误：{e.Message}", e); - return false; - } - } - - /// - /// 连接到 OPC UA 服务器并订阅或轮询指定的变量。 - /// - private async Task ConnectOpcUaServiceAsync(CancellationToken stoppingToken) - { - if (stoppingToken.IsCancellationRequested) - { - return; - } - - var connectTasks = new List(); - - // 遍历_opcUaDevices中的所有设备，尝试连接 - foreach (var device in _opcUaDevices.Values.ToList()) - { - connectTasks.Add(ConnectSingleOpcUaDeviceAsync(device, stoppingToken)); - } - - await Task.WhenAll(connectTasks); - } - - /// - /// 连接单个OPC UA设备。 - /// - /// 要连接的设备。 - /// 取消令牌。 - private async Task ConnectSingleOpcUaDeviceAsync(Device device, CancellationToken stoppingToken = default) - { - if (stoppingToken.IsCancellationRequested) - { - return; - } - - // Check if already connected - if (_opcUaSessions.TryGetValue(device.OpcUaEndpointUrl, out var existingSession)) - { - if (existingSession.Connected) - { - NlogHelper.Info($"已连接到 OPC UA 服务器: {device.OpcUaEndpointUrl}"); - return; - } - else - { - // Remove disconnected session from dictionary to attempt reconnection - _opcUaSessions.TryRemove(device.OpcUaEndpointUrl, out _); - } - } - - NlogHelper.Info($"开始连接OPC UA服务器: {device.Name} ({device.OpcUaEndpointUrl})"); - try - { - var session = await ServiceHelper.CreateOpcUaSessionAsync(device.OpcUaEndpointUrl, stoppingToken); - if (session == null) - { - NlogHelper.Warn($"创建OPC UA会话失败: {device.OpcUaEndpointUrl}"); - return; // 连接失败，直接返回 - } - - _opcUaSessions.AddOrUpdate(device.OpcUaEndpointUrl, session, (key, oldValue) => session); - NotificationHelper.ShowSuccess($"已连接到OPC UA服务器: {device.Name} ({device.OpcUaEndpointUrl})"); - } - catch (Exception e) - { - NotificationHelper.ShowError( - $"OPC UA服务连接 {device.Name} ({device.OpcUaEndpointUrl}) 过程中发生错误：{e.Message}", e); - } - } - - 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) - { - NlogHelper.Info("OPC UA 后台服务轮询变量被取消。"); - } - catch (Exception ex) - { - NotificationHelper.ShowError($"OPC UA 后台服务在轮询变量过程中发生错误：{ex.Message}", ex); - } - } - - /// - /// 轮询单个设备的所有 OPC UA 变量。 - /// - /// 设备的 ID。 - /// 取消令牌。 - private async Task PollSingleDeviceVariablesAsync(int deviceId, CancellationToken stoppingToken) - { - if (stoppingToken.IsCancellationRequested) return; - - if (!_opcUaDevices.TryGetValue(deviceId, out var device) || device.OpcUaEndpointUrl == null) - { - NlogHelper.Warn($"OpcUa轮询变量时，在deviceDic中未找到ID为 {deviceId} 的设备，或其服务器地址为空，请检查！"); - return; - } - - if (!device.IsActive) return; - - if (!_opcUaSessions.TryGetValue(device.OpcUaEndpointUrl, out var session) || !session.Connected) - { - if (device.IsActive) - { - NlogHelper.Warn($"用于 {device.OpcUaEndpointUrl} 的 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 (!ServiceHelper.PollingIntervals.TryGetValue(variable.PollLevelType, out var interval) || (DateTime.Now - variable.UpdateTime) < interval) - { - continue; - } - - await ReadAndProcessOpcUaVariableAsync(session, variable, stoppingToken); - } - } - - /// - /// 读取单个 OPC UA 变量并处理其数据。 - /// - /// OPC UA 会话。 - /// 要读取的变量。 - /// 取消令牌。 - private async Task ReadAndProcessOpcUaVariableAsync(Session session, Variable variable, CancellationToken stoppingToken) - { - var nodesToRead = new ReadValueIdCollection - { - new ReadValueId - { - NodeId = new NodeId(variable.OpcUaNodeId), - AttributeId = Attributes.Value - } - }; - - try - { - var readResponse = await session.ReadAsync(null, 0, TimestampsToReturn.Both, nodesToRead, stoppingToken); - var result = readResponse.Results?.FirstOrDefault(); - if (result == null) return; - - if (!StatusCode.IsGood(result.StatusCode)) - { - NlogHelper.Warn($"读取 OPC UA 变量 {variable.Name} ({variable.OpcUaNodeId}) 失败: {result.StatusCode}"); - return; - } - - await UpdateAndEnqueueVariable(variable, result.Value); - } - catch (ServiceResultException ex) when (ex.StatusCode == StatusCodes.BadSessionIdInvalid) - { - NlogHelper.Error($"OPC UA会话ID无效，变量: {variable.Name} ({variable.OpcUaNodeId})。正在尝试重新连接...", ex); - // Assuming device can be retrieved from variable or passed as parameter if needed for ConnectSingleOpcUaDeviceAsync - // For now, I'll just log and let the outer loop handle reconnection if the session is truly invalid for the device. - // If a full device object is needed here, it would need to be passed down from PollSingleDeviceVariablesAsync. - // For simplicity, I'll remove the direct reconnection attempt here and rely on the outer loop. - await ConnectSingleOpcUaDeviceAsync(variable.VariableTable.Device, stoppingToken); - } - catch (Exception ex) - { - NlogHelper.Error($"轮询OPC UA变量 {variable.Name} ({variable.OpcUaNodeId}) 时发生未知错误: {ex.Message}", ex); - } - } - - /// - /// 更新变量数据，并将其推送到数据处理队列。 - /// - /// 要更新的变量。 - /// 读取到的数据值。 - private async Task UpdateAndEnqueueVariable(Variable variable, object value) - { - try - { - // 更新变量的原始数据值和显示值。 - variable.DataValue = value.ToString(); - variable.DisplayValue = value.ToString(); // 或者根据需要进行格式化 - variable.UpdateTime = DateTime.Now; - // Console.WriteLine($"OpcUa后台服务轮询变量：{variable.Name},值：{variable.DataValue}"); - // 将更新后的数据推入处理队列。 - await _dataProcessingService.EnqueueAsync(variable); - } - catch (Exception ex) - { - NlogHelper.Error($"更新变量 {variable.Name} 并入队失败:{ex.Message}", ex); - } - } - - /// - /// 设置 OPC UA 订阅并添加监控项。 - /// - /// 取消令牌。 - private async Task SetupOpcUaSubscriptionAsync(CancellationToken stoppingToken) - { - if (stoppingToken.IsCancellationRequested) - { - return; - } - - var setupSubscriptionTasks = new List(); - - foreach (var deviceId in _opcUaSubVariablesByDeviceId.Keys.ToList()) - { - setupSubscriptionTasks.Add(Task.Run(() => - { - if (stoppingToken.IsCancellationRequested) - { - return; // 任务被取消，退出循环 - } - - var device = _dataServices.Devices.FirstOrDefault(d => d.Id == deviceId); - if (device == null) - { - NlogHelper.Warn($"未找到ID为 {deviceId} 的设备，无法设置订阅。"); - return; - } - - Subscription subscription = null; - // 得到session - if (!_opcUaSessions.TryGetValue(device.OpcUaEndpointUrl, out var session)) - { - NlogHelper.Info($"从OpcUa会话字典中获取会话失败： {device.OpcUaEndpointUrl} "); - return; - } - - // 判断设备是否已经添加了订阅 - if (_opcUaSubscriptions.TryGetValue(device.OpcUaEndpointUrl, out subscription)) - { - NlogHelper.Info($"OPC UA 终结点 {device.OpcUaEndpointUrl} 已存在订阅。"); - } - else - { - subscription = new Subscription(session.DefaultSubscription); - subscription.PublishingInterval = _opcUaSubscriptionPublishingIntervalMs; // 发布间隔（毫秒） - session.AddSubscription(subscription); - subscription.Create(); - _opcUaSubscriptions.AddOrUpdate(device.OpcUaEndpointUrl, 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,sender, e); - - subscription.AddItem(monitoredItem); - } - - subscription.ApplyChanges(); // 应用更改 - NlogHelper.Info($"设备: {device.Name}, 添加了 {variablesToSubscribe.Count} 个订阅变量。"); - } - })); - } - - await Task.WhenAll(setupSubscriptionTasks); - } - - /// - /// 订阅变量变化的通知 - /// - /// 发生变化的变量。 - /// - /// - private async void OnSubNotification(Variable variable, MonitoredItem monitoredItem, - MonitoredItemNotificationEventArgs e) - { - - foreach (var value in monitoredItem.DequeueValues()) - { - Console.WriteLine( - $"[OPC UA 通知] {monitoredItem.DisplayName}: {value.Value} | 时间戳: {value.SourceTimestamp.ToLocalTime()} | 状态: {value.StatusCode}"); - if (StatusCode.IsGood(value.StatusCode)) - { - await UpdateAndEnqueueVariable(variable, value.Value); - } - } - } - - /// - /// 断开所有 OPC UA 会话。 - /// - private async Task DisconnectAllOpcUaSessionsAsync() - { - if (_opcUaSessions.IsEmpty) - return; - - NlogHelper.Info("正在断开所有 OPC UA 会话..."); - var closeTasks = new List(); - - foreach (var endpointUrl in _opcUaSessions.Keys.ToList()) - { - closeTasks.Add(Task.Run(async () => - { - NlogHelper.Info($"正在断开 OPC UA 会话: {endpointUrl}"); - if (_opcUaSessions.TryRemove(endpointUrl, out var session)) - { - if (_opcUaSubscriptions.TryRemove(endpointUrl, out var subscription)) - { - // 删除订阅。 - await subscription.DeleteAsync(true); - } - - // 关闭会话。 - await session.CloseAsync(); - NotificationHelper.ShowInfo($"已从 OPC UA 服务器断开连接: {endpointUrl}"); - } - })); - } - - await Task.WhenAll(closeTasks); - } - } -} \ No newline at end of file diff --git a/DMS.WPF/Services/S7BackgroundService.cs b/DMS.WPF/Services/S7BackgroundService.cs deleted file mode 100644 index 533fcf9..0000000 --- a/DMS.WPF/Services/S7BackgroundService.cs +++ /dev/null @@ -1,448 +0,0 @@ -using System.Collections.Concurrent; -using DMS.Core.Enums; -using DMS.Helper; -using DMS.WPF.Models; -using Microsoft.Extensions.Hosting; -using S7.Net; -using S7.Net.Types; -using DateTime = System.DateTime; - -namespace DMS.Services -{ - /// - /// S7后台服务，继承自BackgroundService，用于在后台周期性地轮询S7 PLC设备数据。 - /// - public class S7BackgroundService : BackgroundService - { - // 数据服务实例，用于访问和操作应用程序数据，如设备配置。 - private readonly DataServices _dataServices; - - // 数据处理服务实例，用于将读取到的数据推入处理队列。 - private readonly IDataProcessingService _dataProcessingService; - - // 存储 S7设备，键为设备Id，值为会话对象。 - private readonly ConcurrentDictionary _s7Devices; - - // 储存所有要轮询更新的变量，键是Device.Id,值是这个设备所有要轮询的变量 - private readonly ConcurrentDictionary> _s7PollVariablesByDeviceId; // Key: Variable.Id - - // 存储S7 PLC客户端实例的字典，键为设备ID，值为Plc对象。 - private readonly ConcurrentDictionary _s7PlcClientsByIp; - - // 储存所有变量的字典，方便通过id获取变量对象 - private readonly Dictionary _s7VariablesById; - - // S7轮询一次读取的变量数，不得大于15 - private readonly int _s7PollOnceReadMultipleVars = 9; - - // S7轮询一遍后的等待时间 - private readonly int _s7PollOnceSleepTimeMs = 100; - - - - private readonly SemaphoreSlim _reloadSemaphore = new SemaphoreSlim(0); - - - /// - /// 构造函数，注入数据服务和数据处理服务。 - /// - /// 数据服务实例。 - /// 数据处理服务实例。 - public S7BackgroundService(DataServices dataServices, IDataProcessingService dataProcessingService) - { - _dataServices = dataServices; - _dataProcessingService = dataProcessingService; - _s7Devices = new ConcurrentDictionary(); - _s7PollVariablesByDeviceId = new ConcurrentDictionary>(); - _s7PlcClientsByIp = new ConcurrentDictionary(); - _s7VariablesById = new(); - - // 订阅设备列表变更事件，以便在设备配置更新时重新加载。 - _dataServices.OnDeviceListChanged += HandleDeviceListChanged; - // 订阅单个设备IsActive状态变更事件 - _dataServices.OnDeviceIsActiveChanged += HandleDeviceIsActiveChanged; - } - - - protected override async Task ExecuteAsync(CancellationToken stoppingToken) - { - NlogHelper.Info("S7后台服务正在启动。"); - _reloadSemaphore.Release(); // Initial trigger to load variables and connect - - try - { - while (!stoppingToken.IsCancellationRequested) - { - await _reloadSemaphore.WaitAsync(stoppingToken); // Wait for a reload signal - - if (stoppingToken.IsCancellationRequested) - { - break; - } - - if (_dataServices.Devices == null || _dataServices.Devices.Count == 0) - { - NlogHelper.Info("没有可用的S7设备，等待设备列表更新..."); - continue; - } - - var isLoaded = LoadVariables(); - if (!isLoaded) - { - NlogHelper.Info("加载变量过程中发生了错误，停止后面的操作。"); - continue; - } - - await ConnectS7Service(stoppingToken); - NlogHelper.Info("S7后台服务开始轮询变量...."); - - // 持续轮询，直到取消请求或需要重新加载 - while (!stoppingToken.IsCancellationRequested && _reloadSemaphore.CurrentCount == 0) - { - await PollS7VariableOnce(stoppingToken); - await Task.Delay(_s7PollOnceSleepTimeMs, stoppingToken); - } - } - } - catch (OperationCanceledException) - { - NlogHelper.Info("S7后台服务已停止。"); - } - catch (Exception e) - { - NlogHelper.Error($"S7后台服务运行中发生了错误:{e.Message}", e); - } - finally - { - await DisconnectAllPlc(); - } - } - - /// - /// 处理设备列表变更事件的回调方法。 - /// - /// 事件发送者。 - /// 更新后的设备列表。 - private async void HandleDeviceListChanged(List devices) - { - NlogHelper.Info("设备列表已更改。S7客户端可能需要重新初始化。"); - - - _reloadSemaphore.Release(); // 触发ExecuteAsync中的全面重新加载 - } - - /// - /// 处理单个设备IsActive状态变更事件。 - /// - /// 发生状态变化的设备。 - /// 设备新的IsActive状态。 - private async void HandleDeviceIsActiveChanged(Device device, bool isActive) - { - if (device.ProtocolType != ProtocolType.S7) - return; - - - NlogHelper.Info($"设备 {device.Name} (ID: {device.Id}) 的IsActive状态改变为 {isActive}。"); - if (!isActive) - { - // 设备变为非活动状态，断开连接 - if (_s7PlcClientsByIp.TryRemove(device.Ip, out var plcClient)) - { - try - { - if (plcClient.IsConnected) - { - plcClient.Close(); - NotificationHelper.ShowSuccess($"已断开设备 {device.Name} ({device.Ip}) 的连接。"); - } - } - catch (Exception ex) - { - NlogHelper.Error($"断开设备 {device.Name} ({device.Ip}) 连接时发生错误：{ex.Message}", ex); - } - } - } - - // 触发重新加载，让LoadVariables和ConnectS7Service处理设备列表的更新 - _reloadSemaphore.Release(); - } - - - private async Task PollS7VariableOnce(CancellationToken stoppingToken) - { - try - { - // 获取当前需要轮询的设备ID列表的快照 - var deviceIdsToPoll = _s7PollVariablesByDeviceId.Keys.ToList(); - - // 为每个设备创建并发轮询任务 - var pollingTasks = deviceIdsToPoll.Select(async deviceId => - { - if (!_s7Devices.TryGetValue(deviceId, out var device)) - { - NlogHelper.Warn($"S7服务轮询时在deviceDic中没有找到Id为：{deviceId}的设备"); - return; // 跳过此设备 - } - - if (!_s7PlcClientsByIp.TryGetValue(device.Ip, out var plcClient)) - { - NlogHelper.Warn($"S7服务轮询时没有找到设备I：{deviceId}的初始化好的Plc客户端对象！"); - return; // 跳过此设备 - } - - if (!plcClient.IsConnected) - { - NlogHelper.Warn($"S7服务轮询时设备：{device.Name},没有连接，跳过本次轮询。"); - return; // 跳过此设备，等待ConnectS7Service重新连接 - } - - if (!_s7PollVariablesByDeviceId.TryGetValue(deviceId, out var variableList)) - { - NlogHelper.Warn($"S7服务轮询时没有找到设备I：{deviceId},要轮询的变量列表！"); - return; // 跳过此设备 - } - - // 轮询当前设备的所有变量 - var dataItemsToRead = new Dictionary(); // Key: Variable.Id, Value: DataItem - var variablesToProcess = new List(); // List of variables to process in this batch - - foreach (var variable in variableList) - { - if (stoppingToken.IsCancellationRequested) - { - return; // 任务被取消，退出循环 - } - - // 获取变量的轮询间隔。 - if (!ServiceHelper.PollingIntervals.TryGetValue( - variable.PollLevelType, out var interval)) - { - NlogHelper.Info($"未知轮询级别 {variable.PollLevelType}，跳过变量 {variable.Name}。"); - continue; - } - - // 检查是否达到轮询时间。 - if ((DateTime.Now - variable.UpdateTime) < interval) - continue; // 未到轮询时间，跳过。 - - dataItemsToRead[variable.Id] = DataItem.FromAddress(variable.S7Address); - variablesToProcess.Add(variable); - - // 达到批量读取数量或已是最后一个变量，执行批量读取 - if (dataItemsToRead.Count >= _s7PollOnceReadMultipleVars || variable == variableList.Last()) - { - try - { - // Perform the bulk read - await plcClient.ReadMultipleVarsAsync(dataItemsToRead.Values.ToList(),stoppingToken); - - // Process the results - foreach (var varData in variablesToProcess) - { - if (dataItemsToRead.TryGetValue(varData.Id, out var dataItem)) - { - // Now dataItem has the updated value from the PLC - await UpdateAndEnqueueVariable(varData, dataItem); - } - } - } - catch (Exception ex) - { - NlogHelper.Error($"从设备 {device.Name} 批量读取变量失败:{ex.Message}", ex); - } - finally - { - dataItemsToRead.Clear(); - variablesToProcess.Clear(); - } - } - } - }).ToList(); - - // 等待所有设备的轮询任务完成 - await Task.WhenAll(pollingTasks); - } - catch (OperationCanceledException) - { - NlogHelper.Info("S7后台服务轮询变量被取消。"); - } - catch (Exception ex) - { - NotificationHelper.ShowError($"S7后台服务在轮询变量过程中发生错误：{ex.Message}", ex); - } - } - - /// - /// 更新变量数据，并将其推送到数据处理队列。 - /// - /// 要更新的变量。 - /// 包含读取到的数据项。 - private async Task UpdateAndEnqueueVariable(Variable variable, DataItem dataItem) - { - try - { - // 更新变量的原始数据值和显示值。 - variable.DataValue = dataItem.Value.ToString(); - variable.DisplayValue = dataItem.Value.ToString(); - variable.UpdateTime = DateTime.Now; - // Console.WriteLine($"S7后台服务轮询变量：{variable.Name}，值：{variable.DataValue}"); - // 将更新后的数据推入处理队列。 - await _dataProcessingService.EnqueueAsync(variable); - } - catch (Exception ex) - { - NlogHelper.Error($"更新变量 {variable.Name} 并入队失败:{ex.Message}", ex); - } - } - - private async Task ConnectS7Service(CancellationToken stoppingToken) - { - if (stoppingToken.IsCancellationRequested) - { - return; - } - - var connectTasks = new List(); - - // 遍历_s7Devices中的所有设备，尝试连接 - foreach (var device in _s7Devices.Values.ToList()) - { - connectTasks.Add(ConnectSingleDeviceAsync(device, stoppingToken)); - } - - await Task.WhenAll(connectTasks); - } - - /// - /// 连接单个S7 PLC设备。 - /// - /// 要连接的设备。 - /// 取消令牌。 - private async Task ConnectSingleDeviceAsync(Device device, CancellationToken stoppingToken = default) - { - if (stoppingToken.IsCancellationRequested) - { - return; - } - - // Check if already connected - if (_s7PlcClientsByIp.TryGetValue(device.Ip, out var existingPlc)) - { - if (existingPlc.IsConnected) - { - NlogHelper.Info($"已连接到 S7 服务器: {device.Ip}:{device.Prot}"); - return; - } - else - { - // Remove disconnected PLC from dictionary to attempt reconnection - _s7PlcClientsByIp.TryRemove(device.Ip, out _); - } - } - - NlogHelper.Info($"开始连接S7 PLC: {device.Name} ({device.Ip})"); - try - { - var plcClient = new Plc(device.CpuType, device.Ip, (short)device.Prot, device.Rack, device.Slot); - await plcClient.OpenAsync(stoppingToken); // 尝试打开连接。 - - _s7PlcClientsByIp.AddOrUpdate(device.Ip, plcClient, (key, oldValue) => plcClient); - - NotificationHelper.ShowSuccess($"已连接到S7 PLC: {device.Name} ({device.Ip})"); - } - catch (Exception e) - { - NotificationHelper.ShowError($"S7服务连接PLC {device.Name} ({device.Ip}) 过程中发生错误：{e.Message}", e); - } - } - - /// - /// 加载变量 - /// - private bool LoadVariables() - { - try - { - _s7Devices.Clear(); - _s7PollVariablesByDeviceId.Clear(); - _s7VariablesById.Clear(); // 确保在重新加载变量时清空此字典 - - NlogHelper.Info("开始加载S7变量...."); - var s7Devices = _dataServices - .Devices.Where(d => d.IsActive == true && d.ProtocolType == 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.ProtocolType == 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; - } - - NlogHelper.Info($"S7变量加载成功，共加载S7设备：{s7Devices.Count}个，变量数：{totalVariableCount}"); - return true; - } - catch (Exception e) - { - NotificationHelper.ShowError($"S7后台服务加载变量时发生了错误：{e.Message}", e); - return false; - } - } - - - /// - /// 关闭所有PLC的连接 - /// - private async Task DisconnectAllPlc() - { - if (_s7PlcClientsByIp.IsEmpty) - return; - - // 创建一个任务列表，用于并发关闭所有PLC连接 - var closeTasks = new List(); - - // 关闭所有活跃的PLC连接。 - foreach (var plcClient in _s7PlcClientsByIp.Values) - { - if (plcClient.IsConnected) - { - closeTasks.Add(Task.Run(() => - { - try - { - plcClient.Close(); - NlogHelper.Info($"关闭S7连接: {plcClient.IP}"); - } - catch (Exception e) - { - NlogHelper.Error($"S7后台服务关闭{plcClient.IP},后台连接时发生错误：{e.Message}", e); - } - })); - } - } - - // 等待所有关闭任务完成 - await Task.WhenAll(closeTasks); - _s7PlcClientsByIp.Clear(); // Clear the dictionary after all connections are attempted to be closed - } - } -} \ No newline at end of file