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

临时提交

Browse Source
This commit is contained in:
David P.G
2025-07-24 22:18:22 +08:00
parent 58cb05645e
commit 3a3ed7a264
6 changed files with 5 additions and 1363 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
DMS.Core/Enums/ProtocolType.cs
View File

@@ -16,5 +16,6 @@ public enum ProtocolType
/// <summary>
/// Modbus TCP 协议。
/// </summary>
ModbusTcp
ModbusTcp,
OpcUA
}

1
DMS.Core/Models/Variable.cs
View File

@@ -143,4 +143,5 @@ public class Variable
public PollLevelType PollLevelType { get; set; }
public DateTime UpdateTime { get; set; }
public OpcUaUpdateType OpcUaUpdateType { get; set; }
}

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

@@ -117,10 +117,8 @@ public partial class App : System.Windows.Application
services.AddSingleton<GrowlNotificationService>();
services.AddHostedService<S7BackgroundService>();
services.AddHostedService<OpcUaBackgroundService>();
services.AddSingleton<MqttBackgroundService>();
services.AddSingleton<ChannelBusService>();
services.AddHostedService(provider => provider.GetRequiredService<MqttBackgroundService>());
services.AddSingleton<OpcUaBackgroundService>();
services.AddHostedService<DMS.Infrastructure.Services.MqttBackgroundService>();
// 注册 AutoMapper
services.AddAutoMapper(typeof(App).Assembly);

345
DMS.WPF/Services/MqttBackgroundService.cs
View File

@@ -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
{
/// <summary>
/// MQTT后台服务，继承自BackgroundService，用于在后台管理MQTT连接和数据发布。
/// </summary>
public class MqttBackgroundService : BackgroundService
{
private readonly DataServices _dataServices;
private readonly MqttRepository _mqttRepository;
private readonly ConcurrentDictionary<int, IMqttClient> _mqttClients;
private readonly ConcurrentDictionary<int, Mqtt> _mqttConfigDic;
private readonly ConcurrentDictionary<int, int> _reconnectAttempts;
private readonly SemaphoreSlim _reloadSemaphore = new(0);
private readonly Channel<VariableMqtt> _messageChannel;
/// <summary>
/// 构造函数，注入DataServices。
/// </summary>
public MqttBackgroundService(DataServices dataServices, MqttRepository mqttRepository)
{
_dataServices = dataServices;
_mqttRepository = mqttRepository;
_mqttClients = new ConcurrentDictionary<int, IMqttClient>();
_mqttConfigDic = new ConcurrentDictionary<int, Mqtt>();
_reconnectAttempts = new ConcurrentDictionary<int, int>();
_messageChannel = Channel.CreateUnbounded<VariableMqtt>();
_dataServices.OnMqttListChanged += HandleMqttListChanged;
}
/// <summary>
/// 将待发送的变量数据异步推入队列。
/// </summary>
/// <param name="data">包含MQTT别名和变量数据的对象。</param>
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<VariableMqtt>();
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<VariableMqtt> 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<Mqtt> mqtts)
{
NlogHelper.Info("Mqtt列表发生了变化，正在重新加载数据...");
_reloadSemaphore.Release();
}
}
}

565
DMS.WPF/Services/OpcUaBackgroundService.cs
View File

@@ -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<int, Device> _opcUaDevices;
// 存储 OPC UA 会话，键为终结点 URL，值为会话对象。
private readonly ConcurrentDictionary<string, Session> _opcUaSessions;
// 存储 OPC UA 订阅，键为终结点 URL，值为订阅对象。
private readonly ConcurrentDictionary<string, Subscription> _opcUaSubscriptions;
// 存储活动的 OPC UA 变量，键为变量的OpcNodeId
private readonly ConcurrentDictionary<string, Variable> _opcUaPollVariablesByNodeId;
// 储存所有要轮询更新的变量，键是Device.Id,值是这个设备所有要轮询的变量
private readonly ConcurrentDictionary<int, List<Variable>> _opcUaPollVariablesByDeviceId;
// 储存所有要订阅更新的变量，键是Device.Id,值是这个设备所有要轮询的变量
private readonly ConcurrentDictionary<int, List<Variable>> _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<int, Device>();
_opcUaSessions = new ConcurrentDictionary<string, Session>();
_opcUaSubscriptions = new ConcurrentDictionary<string, Subscription>();
_opcUaPollVariablesByNodeId = new ConcurrentDictionary<string, Variable>();
_opcUaPollVariablesByDeviceId = new ConcurrentDictionary<int, List<Variable>>();
_opcUaSubVariablesByDeviceId = new ConcurrentDictionary<int, List<Variable>>();
_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<Device> 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();
}
/// <summary>
/// 从数据库加载所有活动的 OPC UA 变量，并进行相应的连接和订阅管理。
/// </summary>
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;
}
}
/// <summary>
/// 连接到 OPC UA 服务器并订阅或轮询指定的变量。
/// </summary>
private async Task ConnectOpcUaServiceAsync(CancellationToken stoppingToken)
{
if (stoppingToken.IsCancellationRequested)
{
return;
}
var connectTasks = new List<Task>();
// 遍历_opcUaDevices中的所有设备，尝试连接
foreach (var device in _opcUaDevices.Values.ToList())
{
connectTasks.Add(ConnectSingleOpcUaDeviceAsync(device, stoppingToken));
}
await Task.WhenAll(connectTasks);
}
/// <summary>
/// 连接单个OPC UA设备。
/// </summary>
/// <param name="device">要连接的设备。</param>
/// <param name="stoppingToken">取消令牌。</param>
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);
}
}
/// <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.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);
}
}
/// <summary>
/// 读取单个 OPC UA 变量并处理其数据。
/// </summary>
/// <param name="session">OPC UA 会话。</param>
/// <param name="variable">要读取的变量。</param>
/// <param name="stoppingToken">取消令牌。</param>
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);
}
}
/// <summary>
/// 更新变量数据，并将其推送到数据处理队列。
/// </summary>
/// <param name="variable">要更新的变量。</param>
/// <param name="value">读取到的数据值。</param>
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);
}
}
/// <summary>
/// 设置 OPC UA 订阅并添加监控项。
/// </summary>
/// <param name="stoppingToken">取消令牌。</param>
private async Task SetupOpcUaSubscriptionAsync(CancellationToken stoppingToken)
{
if (stoppingToken.IsCancellationRequested)
{
return;
}
var setupSubscriptionTasks = new List<Task>();
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);
}
/// <summary>
/// 订阅变量变化的通知
/// </summary>
/// <param name="variable">发生变化的变量。</param>
/// <param name="monitoredItem"></param>
/// <param name="e"></param>
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);
}
}
}
/// <summary>
/// 断开所有 OPC UA 会话。
/// </summary>
private async Task DisconnectAllOpcUaSessionsAsync()
{
if (_opcUaSessions.IsEmpty)
return;
NlogHelper.Info("正在断开所有 OPC UA 会话...");
var closeTasks = new List<Task>();
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);
}
}
}

448
DMS.WPF/Services/S7BackgroundService.cs
View File

@@ -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
{
/// <summary>
/// S7后台服务，继承自BackgroundService，用于在后台周期性地轮询S7 PLC设备数据。
/// </summary>
public class S7BackgroundService : BackgroundService
{
// 数据服务实例，用于访问和操作应用程序数据，如设备配置。
private readonly DataServices _dataServices;
// 数据处理服务实例，用于将读取到的数据推入处理队列。
private readonly IDataProcessingService _dataProcessingService;
// 存储 S7设备，键为设备Id，值为会话对象。
private readonly ConcurrentDictionary<int, Device> _s7Devices;
// 储存所有要轮询更新的变量，键是Device.Id,值是这个设备所有要轮询的变量
private readonly ConcurrentDictionary<int, List<Variable>> _s7PollVariablesByDeviceId; // Key: Variable.Id
// 存储S7 PLC客户端实例的字典，键为设备ID，值为Plc对象。
private readonly ConcurrentDictionary<string, Plc> _s7PlcClientsByIp;
// 储存所有变量的字典，方便通过id获取变量对象
private readonly Dictionary<int, Variable> _s7VariablesById;
// S7轮询一次读取的变量数，不得大于15
private readonly int _s7PollOnceReadMultipleVars = 9;
// S7轮询一遍后的等待时间
private readonly int _s7PollOnceSleepTimeMs = 100;
private readonly SemaphoreSlim _reloadSemaphore = new SemaphoreSlim(0);
/// <summary>
/// 构造函数，注入数据服务和数据处理服务。
/// </summary>
/// <param name="dataServices">数据服务实例。</param>
/// <param name="dataProcessingService">数据处理服务实例。</param>
public S7BackgroundService(DataServices dataServices, IDataProcessingService dataProcessingService)
{
_dataServices = dataServices;
_dataProcessingService = dataProcessingService;
_s7Devices = new ConcurrentDictionary<int, Device>();
_s7PollVariablesByDeviceId = new ConcurrentDictionary<int, List<Variable>>();
_s7PlcClientsByIp = new ConcurrentDictionary<string, Plc>();
_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();
}
}
/// <summary>
/// 处理设备列表变更事件的回调方法。
/// </summary>
/// <param name="sender">事件发送者。</param>
/// <param name="devices">更新后的设备列表。</param>
private async void HandleDeviceListChanged(List<Device> devices)
{
NlogHelper.Info("设备列表已更改。S7客户端可能需要重新初始化。");
_reloadSemaphore.Release(); // 触发ExecuteAsync中的全面重新加载
}
/// <summary>
/// 处理单个设备IsActive状态变更事件。
/// </summary>
/// <param name="device">发生状态变化的设备。</param>
/// <param name="isActive">设备新的IsActive状态。</param>
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<int, DataItem>(); // Key: Variable.Id, Value: DataItem
var variablesToProcess = new List<Variable>(); // 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);
}
}
/// <summary>
/// 更新变量数据，并将其推送到数据处理队列。
/// </summary>
/// <param name="variable">要更新的变量。</param>
/// <param name="dataItem">包含读取到的数据项。</param>
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<Task>();
// 遍历_s7Devices中的所有设备，尝试连接
foreach (var device in _s7Devices.Values.ToList())
{
connectTasks.Add(ConnectSingleDeviceAsync(device, stoppingToken));
}
await Task.WhenAll(connectTasks);
}
/// <summary>
/// 连接单个S7 PLC设备。
/// </summary>
/// <param name="device">要连接的设备。</param>
/// <param name="stoppingToken">取消令牌。</param>
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);
}
}
/// <summary>
/// 加载变量
/// </summary>
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;
}
}
/// <summary>
/// 关闭所有PLC的连接
/// </summary>
private async Task DisconnectAllPlc()
{
if (_s7PlcClientsByIp.IsEmpty)
return;
// 创建一个任务列表，用于并发关闭所有PLC连接
var closeTasks = new List<Task>();
// 关闭所有活跃的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
}
}
}