From b26fcafee8d139a7a7b3a296dd912124fff2e72c Mon Sep 17 00:00:00 2001 From: "David P.G" Date: Fri, 18 Jul 2025 19:15:29 +0800 Subject: [PATCH] =?UTF-8?q?=E6=B7=BB=E5=8A=A0MQTT=E5=8F=91=E9=80=81?= =?UTF-8?q?=E6=B6=88=E6=81=AF=E7=9A=84=E5=8A=9F=E8=83=BD?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- App.xaml.cs | 5 +- Data/Repositories/DeviceRepository.cs | 1 + Models/VariableMqtt.cs | 16 +- Services/MqttBackgroundService.cs | 312 ++++++++++---------- Services/Processors/MqttPublishProcessor.cs | 39 +++ 5 files changed, 215 insertions(+), 158 deletions(-) create mode 100644 Services/Processors/MqttPublishProcessor.cs diff --git a/App.xaml.cs b/App.xaml.cs index fb73336..ae17662 100644 --- a/App.xaml.cs +++ b/App.xaml.cs @@ -65,6 +65,7 @@ public partial class App : Application var dataProcessingService = Host.Services.GetRequiredService(); dataProcessingService.AddProcessor(Host.Services.GetRequiredService()); dataProcessingService.AddProcessor(Host.Services.GetRequiredService()); + dataProcessingService.AddProcessor(Host.Services.GetRequiredService()); dataProcessingService.AddProcessor(Host.Services.GetRequiredService()); dataProcessingService.AddProcessor(Host.Services.GetRequiredService()); } @@ -105,7 +106,8 @@ public partial class App : Application services.AddSingleton(); services.AddHostedService(); services.AddHostedService(); - services.AddHostedService(); + services.AddSingleton(); + services.AddHostedService(provider => provider.GetRequiredService()); services.AddSingleton(); // 注册 AutoMapper @@ -118,6 +120,7 @@ public partial class App : Application services.AddSingleton(); services.AddSingleton(); services.AddSingleton(); + services.AddSingleton(); // 注册数据仓库 services.AddSingleton(); diff --git a/Data/Repositories/DeviceRepository.cs b/Data/Repositories/DeviceRepository.cs index 18ee837..43ad920 100644 --- a/Data/Repositories/DeviceRepository.cs +++ b/Data/Repositories/DeviceRepository.cs @@ -73,6 +73,7 @@ public class DeviceRepository var dlist = await db.Queryable() .Includes(d => d.VariableTables, dv => dv.Device) .Includes(d => d.VariableTables, dvd => dvd.Variables, data => data.VariableTable) + .Includes(d=>d.VariableTables,vt=>vt.Variables,v=>v.VariableMqtts ) .ToListAsync(); diff --git a/Models/VariableMqtt.cs b/Models/VariableMqtt.cs index d5b4a36..066f1ab 100644 --- a/Models/VariableMqtt.cs +++ b/Models/VariableMqtt.cs @@ -16,9 +16,13 @@ public partial class VariableMqtt : ObservableObject public VariableMqtt(Variable variable, Mqtt mqtt) { - Variable = variable; - Mqtt = mqtt; - MqttAlias = MqttAlias != String.Empty ? MqttAlias : variable.Name; + if (mqtt != null && variable != null) + { + Variable = variable; + Mqtt = mqtt; + MqttAlias = MqttAlias != String.Empty ? MqttAlias : variable.Name; + } + } /// @@ -49,6 +53,10 @@ public partial class VariableMqtt : ObservableObject { get { + if (Variable!=null) + { + + if (Variable.ProtocolType == ProtocolType.S7) { return Variable.S7Address; @@ -57,7 +65,7 @@ public partial class VariableMqtt : ObservableObject { return Variable.OpcUaNodeId; } - + } return string.Empty; } } diff --git a/Services/MqttBackgroundService.cs b/Services/MqttBackgroundService.cs index 8103948..dd5e041 100644 --- a/Services/MqttBackgroundService.cs +++ b/Services/MqttBackgroundService.cs @@ -1,5 +1,6 @@ using System.Collections.Concurrent; using System.Text; +using System.Threading.Channels; using Microsoft.Extensions.Hosting; using MQTTnet; using MQTTnet.Client; @@ -17,26 +18,19 @@ namespace PMSWPF.Services /// public class MqttBackgroundService : BackgroundService { - // 数据服务实例，用于访问和操作应用程序数据，如MQTT配置和变量数据。 private readonly DataServices _dataServices; private readonly MqttRepository _mqttRepository; - // 存储MQTT客户端实例的字典，键为MQTT配置ID，值为IMqttClient对象。 private readonly ConcurrentDictionary _mqttClients; - - // 存储MQTT配置的字典，键为MQTT配置ID，值为Mqtt模型对象。 private readonly ConcurrentDictionary _mqttConfigDic; - - // 存储每个客户端重连尝试次数的字典 private readonly ConcurrentDictionary _reconnectAttempts; - private readonly SemaphoreSlim _reloadSemaphore = new SemaphoreSlim(0); - + private readonly SemaphoreSlim _reloadSemaphore = new(0); + private readonly Channel _messageChannel; /// /// 构造函数，注入DataServices。 /// - /// 数据服务实例。 public MqttBackgroundService(DataServices dataServices, MqttRepository mqttRepository) { _dataServices = dataServices; @@ -44,25 +38,34 @@ namespace PMSWPF.Services _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(); // Initial trigger to load variables and connect + _reloadSemaphore.Release(); + + var processQueueTask = ProcessMessageQueueAsync(stoppingToken); try { while (!stoppingToken.IsCancellationRequested) { - await _reloadSemaphore.WaitAsync(stoppingToken); // Wait for a reload signal + await _reloadSemaphore.WaitAsync(stoppingToken); - if (stoppingToken.IsCancellationRequested) - { - break; - } + if (stoppingToken.IsCancellationRequested) break; if (_dataServices.Mqtts == null || _dataServices.Mqtts.Count == 0) { @@ -70,8 +73,7 @@ namespace PMSWPF.Services continue; } - var isLoaded = LoadMqttConfigurations(); - if (!isLoaded) + if (!LoadMqttConfigurations()) { NlogHelper.Info("加载Mqtt配置过程中发生了错误，停止后面的操作。"); continue; @@ -80,15 +82,15 @@ namespace PMSWPF.Services await ConnectMqttList(stoppingToken); NlogHelper.Info("Mqtt后台服务已启动。"); - // while (!stoppingToken.IsCancellationRequested && _reloadSemaphore.CurrentCount == 0) - // { - // await Task.Delay(1000, stoppingToken); - // } + while (!stoppingToken.IsCancellationRequested && _reloadSemaphore.CurrentCount == 0) + { + await Task.Delay(1000, stoppingToken); + } } } catch (OperationCanceledException) { - NlogHelper.Info("Mqtt后台服务已停止。"); + NlogHelper.Info("Mqtt后台服务正在停止。"); } catch (Exception e) { @@ -96,60 +98,119 @@ namespace PMSWPF.Services } 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) { - // 断开所有已连接的MQTT客户端。 - foreach (var mqttId in _mqttClients.Keys.ToList()) - { - try - { - var client = _mqttClients[mqttId]; - var mqtt = _mqttConfigDic[mqttId]; - mqtt.IsConnected = false; - if (client.IsConnected) - { - await client.DisconnectAsync(new MqttClientDisconnectOptions(), stoppingToken); - } - } - catch (Exception e) - { - NlogHelper.Error($"MqttID:{mqttId},断开连接的过程中发生了错误:{e.Message}", e); - } - } + var disconnectTasks = _mqttClients.Values.Select(client => client.DisconnectAsync(new MqttClientDisconnectOptions(), stoppingToken)); + await Task.WhenAll(disconnectTasks); + _mqttClients.Clear(); } - - /// - /// 加载并连接MQTT配置。 - /// - /// 表示异步操作的任务。 private bool LoadMqttConfigurations() { try { NlogHelper.Info("开始加载Mqtt配置文件..."); _mqttConfigDic.Clear(); - // 从数据服务获取所有MQTT配置。 - var _mqttConfigList = _dataServices.Mqtts.Where(m => m.IsActive) - .ToList(); - foreach (var mqtt in _mqttConfigList) + var mqttConfigList = _dataServices.Mqtts.Where(m => m.IsActive).ToList(); + + foreach (var mqtt in mqttConfigList) { mqtt.OnMqttIsActiveChanged += OnMqttIsActiveChangedHandler; - _mqttConfigDic[mqtt.Id] = mqtt; + _mqttConfigDic.TryAdd(mqtt.Id, mqtt); mqtt.ConnectMessage = "配置加载成功."; } - NlogHelper.Info($"Mqtt配置文件加载成功，开启的Mqtt客户端：{_mqttConfigList.Count}个。"); + NlogHelper.Info($"Mqtt配置文件加载成功，开启的Mqtt客户端：{mqttConfigList.Count}个。"); return true; } catch (Exception e) { - NotificationHelper.ShowError($"Mqtt后台服务在加载变量的过程中发生了错误:{e.Message}"); + NotificationHelper.ShowError($"Mqtt后台服务在加载配置的过程中发生了错误:{e.Message}", e); return false; } } @@ -160,26 +221,17 @@ namespace PMSWPF.Services { if (mqtt.IsActive) { - _reloadSemaphore.Release(); + await ConnectMqtt(mqtt, CancellationToken.None); } else { - if (!_mqttClients.TryGetValue(mqtt.Id, out var client)) - { - NlogHelper.Warn($"没有在Mqtt连接字典中找到名字为：{mqtt.Name}的连接。"); - return; - } - - if (client.IsConnected) + if (_mqttClients.TryRemove(mqtt.Id, out var client) && client.IsConnected) { await client.DisconnectAsync(); + NlogHelper.Info($"{mqtt.Name}的客户端，与服务器断开连接."); } - mqtt.IsConnected = false; - mqtt.ConnectMessage = "断开连接."; - - _mqttClients.TryRemove(mqtt.Id, out _); - NlogHelper.Info($"{mqtt.Name}的客户端，与服务器断开连接."); + mqtt.ConnectMessage = "已断开连接."; } await _mqttRepository.UpdateAsync(mqtt); @@ -191,63 +243,47 @@ namespace PMSWPF.Services } } - /// - /// 连接到指定的MQTT代理。 - /// - /// MQTT配置对象。 - /// 表示异步操作的任务。 private async Task ConnectMqttList(CancellationToken stoppingToken) { - foreach (Mqtt mqtt in _mqttConfigDic.Values.ToList()) - { - try - { - if (_mqttClients.TryGetValue(mqtt.Id, out var mclient) && mclient.IsConnected) - { - NlogHelper.Info($"{mqtt.Name}的Mqtt服务器连接已存在。"); - mqtt.ConnectMessage = "连接成功."; - mqtt.IsConnected = true; - continue; - } - - await ConnectMqtt(mqtt, stoppingToken); - } - catch (Exception ex) - { - NotificationHelper.ShowError($"连接MQTT服务器失败: {mqtt.Name} - {ex.Message}", ex); - } - } + 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 = "开始连接服务器..."; - // 创建MQTT客户端工厂和客户端实例。 + var factory = new MqttFactory(); var client = factory.CreateMqttClient(); - // 构建MQTT客户端连接选项。 + var options = new MqttClientOptionsBuilder() .WithClientId(mqtt.ClientID) .WithTcpServer(mqtt.Host, mqtt.Port) .WithCredentials(mqtt.UserName, mqtt.PassWord) - .WithCleanSession() // 清理会话，每次连接都是新会话 + .WithCleanSession() .Build(); - // 设置连接成功事件处理程序。 - client.UseConnectedHandler(async (e) => { await HandleConnected(e, client, mqtt); }); + 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)); - // 设置接收消息处理程序 - client.UseApplicationMessageReceivedHandler(e => { HandleMessageReceived(e, mqtt); }); - - // 设置断开连接事件处理程序。 - client.UseDisconnectedHandler(async (e) => await HandleDisconnected(e, options, client, mqtt, stoppingToken)); - - // 尝试连接到MQTT代理。 - await client.ConnectAsync(options, stoppingToken); - - // 将连接成功的客户端添加到字典。 - _mqttClients[mqtt.Id] = client; + 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) @@ -255,85 +291,55 @@ namespace PMSWPF.Services 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) + 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; - // 增加重连尝试次数 - if (!_reconnectAttempts.ContainsKey(mqtt.Id)) - { - _reconnectAttempts[mqtt.Id] = 0; - } + if (stoppingToken.IsCancellationRequested || !mqtt.IsActive) return; - _reconnectAttempts[mqtt.Id]++; + _reconnectAttempts.AddOrUpdate(mqtt.Id, 1, (id, count) => count + 1); + var attempt = _reconnectAttempts[mqtt.Id]; - // 指数退避策略 - var maxDelay = TimeSpan.FromMinutes(5); // 最大延迟5分钟 - var baseDelay = TimeSpan.FromSeconds(5); // 基础延迟5秒 - var delay = TimeSpan.FromSeconds(baseDelay.TotalSeconds * - Math.Pow(2, _reconnectAttempts[mqtt.Id] - 1)); - if (delay > maxDelay) - { - delay = maxDelay; - } + var delay = TimeSpan.FromSeconds(Math.Min(60, Math.Pow(2, attempt))); + NlogHelper.Info($"与MQTT服务器：{mqtt.Name} 的连接已断开。将在 {delay.TotalSeconds} 秒后尝试第 {attempt} 次重新连接..."); + mqtt.ConnectMessage = $"连接已断开，{delay.TotalSeconds}秒后尝试重连..."; - NlogHelper.Info( - $"与MQTT服务器：{mqtt.Name} 的连接已断开。将在 {delay.TotalSeconds} 秒后尝试第 {_reconnectAttempts[mqtt.Id]} 次重新连接..."); - - mqtt.ConnectMessage = $"连接已断开。将在 {delay.TotalSeconds} 秒后尝试第 {_reconnectAttempts[mqtt.Id]} 次重新连接..."; await Task.Delay(delay, stoppingToken); + try { - mqtt.ConnectMessage = $"开始重新连接服务器..."; + mqtt.ConnectMessage = "开始重新连接服务器..."; await client.ConnectAsync(options, stoppingToken); } catch (Exception ex) { - mqtt.ConnectMessage = $"重新与Mqtt服务器连接失败."; + mqtt.ConnectMessage = "重新连接失败."; NlogHelper.Error($"重新与Mqtt服务器连接失败: {mqtt.Name}", ex); } } private async Task HandleConnected(MqttClientConnectedEventArgs args, IMqttClient client, Mqtt mqtt) { - // 重置重连尝试次数 - if (_reconnectAttempts.ContainsKey(mqtt.Id)) - { - _reconnectAttempts[mqtt.Id] = 0; - } - + _reconnectAttempts.TryRemove(mqtt.Id, out _); NotificationHelper.ShowSuccess($"已连接到MQTT服务器: {mqtt.Name}"); mqtt.IsConnected = true; mqtt.ConnectMessage = "连接成功."; - // 订阅主题 - await client.SubscribeAsync(new MqttTopicFilterBuilder().WithTopic(mqtt.SubTopic) - .Build()); - NlogHelper.Info($"MQTT客户端 {mqtt.Name} 已订阅主题: {mqtt.SubTopic}"); + if (!string.IsNullOrEmpty(mqtt.SubTopic)) + { + await client.SubscribeAsync(new MqttTopicFilterBuilder().WithTopic(mqtt.SubTopic).Build()); + NlogHelper.Info($"MQTT客户端 {mqtt.Name} 已订阅主题: {mqtt.SubTopic}"); + } } - - /// - /// 处理MQTT列表变化事件的回调方法。 - /// - /// 事件发送者。 - /// 更新后的MQTT配置列表。 private void HandleMqttListChanged(List mqtts) { - NlogHelper.Info("Mqtt列表发生了变化，正在重新加载数据..."); // 记录MQTT列表变化信息 - // 重新加载MQTT配置和变量数据。 + NlogHelper.Info("Mqtt列表发生了变化，正在重新加载数据..."); _reloadSemaphore.Release(); } } -} +} \ No newline at end of file diff --git a/Services/Processors/MqttPublishProcessor.cs b/Services/Processors/MqttPublishProcessor.cs new file mode 100644 index 0000000..01c22cc --- /dev/null +++ b/Services/Processors/MqttPublishProcessor.cs @@ -0,0 +1,39 @@ +using System.Threading.Tasks; +using PMSWPF.Models; + +namespace PMSWPF.Services.Processors +{ + /// + /// 负责将变量数据发布到MQTT的处理器。 + /// + public class MqttPublishProcessor : IVariableProcessor + { + private readonly MqttBackgroundService _mqttBackgroundService; + + public MqttPublishProcessor(MqttBackgroundService mqttBackgroundService) + { + _mqttBackgroundService = mqttBackgroundService; + } + + /// + /// 处理单个变量上下文，如果有关联的MQTT配置，则将其推送到发送队列。 + /// + /// 包含变量及其元数据的上下文对象。 + public async Task ProcessAsync(VariableContext context) + { + var variable = context.Data; + if (variable?.VariableMqtts == null || variable.VariableMqtts.Count == 0) + { + return; // 没有关联的MQTT配置，直接返回 + } + + // 遍历所有关联的MQTT配置，并将其推入发送队列 + foreach (var variableMqtt in variable.VariableMqtts) + { + // 确保VariableMqtt对象中包含了最新的Variable数据 + variableMqtt.Variable = variable; + await _mqttBackgroundService.SendVariableAsync(variableMqtt); + } + } + } +}