xiaowei/DMS
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

修改OpcUa后台服务(未完成)

Browse Source
This commit is contained in:
David P.G
2025-09-03 19:58:02 +08:00
parent 8122ffc6b7
commit bdae3355aa
6 changed files with 266 additions and 428 deletions
Download Patch File Download Diff File Expand all files Collapse all files

13
DMS.Application/Interfaces/IDeviceDataService.cs
View File

@@ -1,13 +0,0 @@
using DMS.Core.Models;
using System.Collections.Generic;
using System.Threading.Tasks;
namespace DMS.Application.Interfaces;
public interface IDeviceDataService
{
List<Device> Devices { get; }
event Action<List<Device>> OnDeviceListChanged;
event Action<Device, bool> OnDeviceIsActiveChanged;
Task InitializeAsync();
}

56
DMS.Infrastructure/Services/DeviceDataService.cs
View File

@@ -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<Device> _devices;
public List<Device> Devices => _devices;
public event Action<List<Device>> OnDeviceListChanged;
public event Action<Device, bool> OnDeviceIsActiveChanged;
public DeviceDataService(IRepositoryManager repositoryManager)
{
_repositoryManager = repositoryManager;
_devices = new List<Device>();
}
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();
}
}

4
DMS.Infrastructure/Services/MqttBackgroundService.cs
View File

@@ -19,7 +19,6 @@ namespace DMS.Infrastructure.Services;
/// </summary>
public class MqttBackgroundService : BackgroundService
{
private readonly IDeviceDataService _deviceDataService;
private readonly IRepositoryManager _repositoryManager;
private readonly ILogger<MqttBackgroundService> _logger;
@@ -33,9 +32,8 @@ public class MqttBackgroundService : BackgroundService
/// <summary>
/// 构造函数，注入DataServices。
/// </summary>
public MqttBackgroundService(IDeviceDataService deviceDataService, IRepositoryManager repositoryManager, ILogger<MqttBackgroundService> logger)
public MqttBackgroundService(IRepositoryManager repositoryManager, ILogger<MqttBackgroundService> logger)
{
_deviceDataService = deviceDataService;
_repositoryManager = repositoryManager;
_logger = logger;
_mqttClients = new ConcurrentDictionary<int, IMqttClient>();

449
DMS.Infrastructure/Services/OpcUaBackgroundService.cs
View File

@@ -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<OpcUaBackgroundService> _logger;
// 存储 OPC UA 设备，键为设备Id，值为会话对象。
private readonly ConcurrentDictionary<int, Device> _opcUaDevices;
private readonly ConcurrentDictionary<int, DeviceDto> _opcUaDevices = new();
// 存储 OPC UA 会话，键为终结点 URL，值为会话对象。
private readonly ConcurrentDictionary<string, Session> _opcUaSessions;
private readonly ConcurrentDictionary<DeviceDto, OpcUaService> _opcUaServices;
// 存储 OPC UA 订阅，键为终结点 URL，值为订阅对象。
private readonly ConcurrentDictionary<string, Subscription> _opcUaSubscriptions;
@@ -33,7 +37,7 @@ public class OpcUaBackgroundService : BackgroundService
private readonly ConcurrentDictionary<int, List<Variable>> _opcUaPollVariablesByDeviceId;
// 储存所有要订阅更新的变量，键是Device.Id,值是这个设备所有要轮询的变量
private readonly ConcurrentDictionary<int, List<Variable>> _opcUaSubVariablesByDeviceId;
private readonly ConcurrentDictionary<int, List<VariableDto>> _opcUaVariablesByDeviceId;
private readonly SemaphoreSlim _reloadSemaphore = new SemaphoreSlim(0);
@@ -47,11 +51,15 @@ public class OpcUaBackgroundService : BackgroundService
private readonly int _opcUaSubscriptionSamplingIntervalMs = 1000;
// 模拟 PollingIntervals，实际应用中可能从配置或数据库加载
private static readonly Dictionary<PollLevelType, TimeSpan> PollingIntervals = new Dictionary<PollLevelType, TimeSpan>
private static readonly Dictionary<PollLevelType, TimeSpan> PollingIntervals
= new Dictionary<PollLevelType, TimeSpan>
{
{ PollLevelType.TenMilliseconds, TimeSpan.FromMilliseconds((int)PollLevelType.TenMilliseconds) },
{ PollLevelType.HundredMilliseconds, TimeSpan.FromMilliseconds((int)PollLevelType.HundredMilliseconds) },
{ PollLevelType.FiveHundredMilliseconds, TimeSpan.FromMilliseconds((int)PollLevelType.FiveHundredMilliseconds) },
{
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) },
@@ -64,27 +72,29 @@ public class OpcUaBackgroundService : BackgroundService
{ PollLevelType.ThirtyMinutes, TimeSpan.FromMilliseconds((int)PollLevelType.ThirtyMinutes) }
};
public OpcUaBackgroundService(IDeviceDataService deviceDataService, IDataCenterService dataCenterService, IDataProcessingService dataProcessingService, ILogger<OpcUaBackgroundService> logger)
public OpcUaBackgroundService(IDataCenterService dataCenterService,
ILogger<OpcUaBackgroundService> logger)
{
_deviceDataService = deviceDataService;
_dataProcessingService = dataProcessingService;
_dataCenterService = dataCenterService;
// _dataProcessingService = dataProcessingService;
_logger = logger;
_opcUaDevices = new ConcurrentDictionary<int, Device>();
_opcUaSessions = new ConcurrentDictionary<string, Session>();
_opcUaServices = new ConcurrentDictionary<DeviceDto, OpcUaService>();
_opcUaSubscriptions = new ConcurrentDictionary<string, Subscription>();
_opcUaPollVariablesByNodeId = new ConcurrentDictionary<string, Variable>();
_opcUaPollVariablesByDeviceId = new ConcurrentDictionary<int, List<Variable>>();
_opcUaSubVariablesByDeviceId = new ConcurrentDictionary<int, List<Variable>>();
_opcUaVariablesByDeviceId = new ConcurrentDictionary<int, List<VariableDto>>();
_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<Device> 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();
// }
/// <summary>
/// 从数据库加载所有活动的 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)
var variableDtos = opcUaDevice.VariableTables?.SelectMany(vt => vt.Variables)
.Where(vd => vd.IsActive == true &&
vd.Protocol == ProtocolType.OpcUa &&
vd.OpcUaUpdateType == OpcUaUpdateType.OpcUaSubscription)
vd.Protocol == ProtocolType.OpcUa)
.ToList();
totalSubVariableCount += dSubList.Count;
_opcUaSubVariablesByDeviceId.AddOrUpdate(opcUaDevice.Id, dSubList, (key, oldValue) => dSubList);
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
/// </summary>
/// <param name="device">要连接的设备。</param>
/// <param name="stoppingToken">取消令牌。</param>
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}");
}
}
/// <summary>
/// 轮询单个设备的所有 OPC UA 变量。
/// </summary>
/// <param name="deviceId">设备的 ID。</param>
/// <param name="stoppingToken">取消令牌。</param>
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);
}
}
// /// <summary>
// /// 轮询单个设备的所有 OPC UA 变量。
// /// </summary>
// /// <param name="deviceId">设备的 ID。</param>
// /// <param name="stoppingToken">取消令牌。</param>
// 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);
// }
// }
/// <summary>
/// 读取单个 OPC UA 变量并处理其数据。
@@ -386,7 +358,8 @@ public class OpcUaBackgroundService : BackgroundService
/// <param name="session">OPC UA 会话。</param>
/// <param name="variable">要读取的变量。</param>
/// <param name="stoppingToken">取消令牌。</param>
private async Task ReadAndProcessOpcUaVariableAsync(Session session, Variable variable, CancellationToken stoppingToken)
private async Task ReadAndProcessOpcUaVariableAsync(Session session, Variable variable,
CancellationToken stoppingToken)
{
var nodesToRead = new ReadValueIdCollection
{
@@ -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
/// <param name="stoppingToken">取消令牌。</param>
private async Task SetupOpcUaSubscriptionAsync(CancellationToken stoppingToken)
{
if (stoppingToken.IsCancellationRequested)
foreach (var opcUaServiceKayValuePair in _opcUaServices)
{
return;
var device = opcUaServiceKayValuePair.Key;
var opcUaService = opcUaServiceKayValuePair.Value;
if (_opcUaVariablesByDeviceId.TryGetValue(device.Id, out var opcUaVariables))
{
var variableGroup = opcUaVariables.GroupBy(variable => variable.PollLevel);
foreach (var vGroup in variableGroup)
{
var pollLevelType = vGroup.Key;
var opcUaNodes
= vGroup.Select(variableDto => new OpcUaNode() { NodeId = variableDto.OpcUaNodeId })
.ToList();
opcUaService.SubscribeToNode(opcUaNodes,HandleDataChanged);
}
}
}
}
var setupSubscriptionTasks = new List<Task>();
private void HandleDataChanged(OpcUaNode opcUaNode)
{
foreach (var deviceId in _opcUaSubVariablesByDeviceId.Keys.ToList())
{
setupSubscriptionTasks.Add(Task.Run(async () =>
{
if (stoppingToken.IsCancellationRequested)
{
return; // 任务被取消，退出循环
}
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);
}
/// <summary>
@@ -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
/// </summary>
private async Task DisconnectAllOpcUaSessionsAsync()
{
if (_opcUaSessions.IsEmpty)
if (_opcUaServices.IsEmpty)
return;
_logger.LogInformation("正在断开所有 OPC UA 会话...");
var closeTasks = new List<Task>();
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}");
}
}));
}

130
DMS.Infrastructure/Services/S7BackgroundService.cs
View File

@@ -17,8 +17,6 @@ namespace DMS.Infrastructure.Services;
/// </summary>
public class S7BackgroundService : BackgroundService
{
// 数据服务实例，用于访问和操作应用程序数据，如设备配置。
private readonly IDeviceDataService _deviceDataService;
// 数据处理服务实例，用于将读取到的数据推入处理队列。
private readonly IDataProcessingService _dataProcessingService;
@@ -69,20 +67,14 @@ public class S7BackgroundService : BackgroundService
/// <param name="deviceDataService">设备数据服务实例。</param>
/// <param name="dataProcessingService">数据处理服务实例。</param>
/// <param name="logger">日志记录器实例。</param>
public S7BackgroundService(IDeviceDataService deviceDataService, IDataProcessingService dataProcessingService, ILogger<S7BackgroundService> logger)
public S7BackgroundService( IDataProcessingService dataProcessingService, ILogger<S7BackgroundService> logger)
{
_deviceDataService = deviceDataService;
_dataProcessingService = dataProcessingService;
_logger = logger;
_s7Devices = new ConcurrentDictionary<int, Device>();
_s7PollVariablesByDeviceId = new ConcurrentDictionary<int, List<Variable>>();
_s7PlcClientsByIp = new ConcurrentDictionary<string, Plc>();
_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
/// <summary>
/// 加载变量
/// </summary>
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;
// // }
// }
/// <summary>
/// 关闭所有PLC的连接

2
DMS.WPF/App.xaml.cs
View File

@@ -138,6 +138,8 @@ public partial class App : System.Windows.Application
});
// 注册数据处理服务和处理器
services.AddHostedService<OpcUaBackgroundService>();
services.AddSingleton<IDataProcessingService, DataProcessingService>();
services.AddHostedService(provider => (DataProcessingService)provider.GetRequiredService<IDataProcessingService>());
services.AddSingleton<CheckValueChangedProcessor>();