using System.Collections.Concurrent; using DMS.Core.Enums; using DMS.Core.Models; using Microsoft.Extensions.Hosting; using S7.Net; using S7.Net.Types; using DateTime = System.DateTime; using Microsoft.Extensions.Logging; using DMS.Application.Interfaces; using DMS.Core.Interfaces; namespace DMS.Infrastructure.Services; /// /// S7后台服务,继承自BackgroundService,用于在后台周期性地轮询S7 PLC设备数据。 /// public class S7BackgroundService : BackgroundService { // 数据服务实例,用于访问和操作应用程序数据,如设备配置。 private readonly IDeviceDataService _deviceDataService; // 数据处理服务实例,用于将读取到的数据推入处理队列。 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 ILogger _logger; private readonly SemaphoreSlim _reloadSemaphore = new SemaphoreSlim(0); // 模拟 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) } }; /// /// 构造函数,注入数据服务和数据处理服务。 /// /// 设备数据服务实例。 /// 数据处理服务实例。 /// 日志记录器实例。 public S7BackgroundService(IDeviceDataService deviceDataService, 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) { _logger.LogInformation("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 (_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后台服务开始轮询变量...."); // 持续轮询,直到取消请求或需要重新加载 while (!stoppingToken.IsCancellationRequested && _reloadSemaphore.CurrentCount == 0) { await PollS7VariableOnce(stoppingToken); await Task.Delay(_s7PollOnceSleepTimeMs, stoppingToken); } } } catch (OperationCanceledException) { _logger.LogInformation("S7后台服务已停止。"); } catch (Exception e) { _logger.LogError(e, $"S7后台服务运行中发生了错误:{e.Message}"); } finally { await DisconnectAllPlc(); } } /// /// 处理设备列表变更事件的回调方法。 /// /// 更新后的设备列表。 private void HandleDeviceListChanged(List devices) { _logger.LogInformation("设备列表已更改。S7客户端可能需要重新初始化。"); _reloadSemaphore.Release(); // 触发ExecuteAsync中的全面重新加载 } /// /// 处理单个设备IsActive状态变更事件。 /// /// 发生状态变化的设备。 /// 设备新的IsActive状态。 private async void HandleDeviceIsActiveChanged(Device device, bool isActive) { if (device.Protocol != ProtocolType.S7) return; _logger.LogInformation($"设备 {device.Name} (ID: {device.Id}) 的IsActive状态改变为 {isActive}。"); if (!isActive) { // 设备变为非活动状态,断开连接 if (_s7PlcClientsByIp.TryRemove(device.IpAddress, out var plcClient)) { try { if (plcClient.IsConnected) { plcClient.Close(); _logger.LogInformation($"已断开设备 {device.Name} ({device.IpAddress}) 的连接。"); } } catch (Exception ex) { _logger.LogError(ex, $"断开设备 {device.Name} ({device.IpAddress}) 连接时发生错误:{ex.Message}"); } } } // 触发重新加载,让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)) { _logger.LogWarning($"S7服务轮询时在deviceDic中没有找到Id为:{deviceId}的设备"); return; // 跳过此设备 } if (!_s7PlcClientsByIp.TryGetValue(device.IpAddress, out var plcClient)) { _logger.LogWarning($"S7服务轮询时没有找到设备I:{deviceId}的初始化好的Plc客户端对象!"); return; // 跳过此设备 } if (!plcClient.IsConnected) { _logger.LogWarning($"S7服务轮询时设备:{device.Name},没有连接,跳过本次轮询。"); return; // 跳过此设备,等待ConnectS7Service重新连接 } if (!_s7PollVariablesByDeviceId.TryGetValue(deviceId, out var variableList)) { _logger.LogWarning($"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 (!PollingIntervals.TryGetValue( variable.PollLevelType, out var interval)) { _logger.LogInformation($"未知轮询级别 {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) { _logger.LogError(ex, $"从设备 {device.Name} 批量读取变量失败:{ex.Message}"); } finally { dataItemsToRead.Clear(); variablesToProcess.Clear(); } } } }).ToList(); // 等待所有设备的轮询任务完成 await Task.WhenAll(pollingTasks); } catch (OperationCanceledException) { _logger.LogInformation("S7后台服务轮询变量被取消。"); } catch (Exception ex) { _logger.LogError(ex, $"S7后台服务在轮询变量过程中发生错误:{ex.Message}"); } } /// /// 更新变量数据,并将其推送到数据处理队列。 /// /// 要更新的变量。 /// 包含读取到的数据项。 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) { _logger.LogError(ex, $"更新变量 {variable.Name} 并入队失败:{ex.Message}"); } } 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.IpAddress, out var existingPlc)) { if (existingPlc.IsConnected) { _logger.LogInformation($"已连接到 S7 服务器: {device.IpAddress}:{device.Port}"); return; } else { // Remove disconnected PLC from dictionary to attempt reconnection _s7PlcClientsByIp.TryRemove(device.IpAddress, out _); } } _logger.LogInformation($"开始连接S7 PLC: {device.Name} ({device.IpAddress})"); try { var s7CpuType = ConvertCpuType(device.CpuType); var plcClient = new Plc(s7CpuType, device.IpAddress, (short)device.Port, device.Rack, device.Slot); await plcClient.OpenAsync(stoppingToken); // 尝试打开连接。 _s7PlcClientsByIp.AddOrUpdate(device.IpAddress, plcClient, (key, oldValue) => plcClient); _logger.LogInformation($"已连接到S7 PLC: {device.Name} ({device.IpAddress})"); } catch (Exception e) { _logger.LogError(e, $"S7服务连接PLC {device.Name} ({device.IpAddress}) 过程中发生错误:{e.Message}"); } } /// /// 将字符串形式的CPU类型转换为S7.Net.CpuType枚举。 /// /// CPU类型的字符串表示。 /// 对应的S7.Net.CpuType枚举值。 /// 如果无法解析CPU类型字符串。 private S7.Net.CpuType ConvertCpuType(string cpuTypeString) { if (Enum.TryParse(cpuTypeString, true, out S7.Net.CpuType cpuType)) { return cpuType; } throw new ArgumentException($"无法解析CPU类型: {cpuTypeString}"); } /// /// 加载变量 /// 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的连接 /// 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(); _logger.LogInformation($"关闭S7连接: {plcClient.IP}"); } catch (Exception e) { _logger.LogError(e, $"S7后台服务关闭{plcClient.IP},后台连接时发生错误:{e.Message}"); } })); } } // 等待所有关闭任务完成 await Task.WhenAll(closeTasks); _s7PlcClientsByIp.Clear(); // Clear the dictionary after all connections are attempted to be closed } }