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

添加MQTT发送消息的功能

Browse Source
This commit is contained in:
David P.G
2025-07-18 19:15:29 +08:00
parent 1c14fbd3d6
commit b26fcafee8
5 changed files with 215 additions and 158 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
App.xaml.cs
View File

@@ -65,6 +65,7 @@ public partial class App : Application
var dataProcessingService = Host.Services.GetRequiredService<IDataProcessingService>();
dataProcessingService.AddProcessor(Host.Services.GetRequiredService<CheckValueChangedProcessor>());
dataProcessingService.AddProcessor(Host.Services.GetRequiredService<LoggingProcessor>());
dataProcessingService.AddProcessor(Host.Services.GetRequiredService<MqttPublishProcessor>());
dataProcessingService.AddProcessor(Host.Services.GetRequiredService<UpdateDbVariableProcessor>());
dataProcessingService.AddProcessor(Host.Services.GetRequiredService<HistoryProcessor>());
}
@@ -105,7 +106,8 @@ public partial class App : Application
services.AddSingleton<GrowlNotificationService>();
services.AddHostedService<S7BackgroundService>();
services.AddHostedService<OpcUaBackgroundService>();
services.AddHostedService<MqttBackgroundService>();
services.AddSingleton<MqttBackgroundService>();
services.AddHostedService(provider => provider.GetRequiredService<MqttBackgroundService>());
services.AddSingleton<OpcUaBackgroundService>();
// 注册 AutoMapper
@@ -118,6 +120,7 @@ public partial class App : Application
services.AddSingleton<LoggingProcessor>();
services.AddSingleton<UpdateDbVariableProcessor>();
services.AddSingleton<HistoryProcessor>();
services.AddSingleton<MqttPublishProcessor>();
// 注册数据仓库
services.AddSingleton<DeviceRepository>();

1
Data/Repositories/DeviceRepository.cs
View File

@@ -73,6 +73,7 @@ public class DeviceRepository
var dlist = await db.Queryable<DbDevice>()
.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();

16
Models/VariableMqtt.cs
View File

@@ -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;
}
}
/// <summary>
@@ -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;
}
}

312
Services/MqttBackgroundService.cs
View File

@@ -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
/// </summary>
public class MqttBackgroundService : BackgroundService
{
// 数据服务实例，用于访问和操作应用程序数据，如MQTT配置和变量数据。
private readonly DataServices _dataServices;
private readonly MqttRepository _mqttRepository;
// 存储MQTT客户端实例的字典，键为MQTT配置ID，值为IMqttClient对象。
private readonly ConcurrentDictionary<int, IMqttClient> _mqttClients;
// 存储MQTT配置的字典，键为MQTT配置ID，值为Mqtt模型对象。
private readonly ConcurrentDictionary<int, Mqtt> _mqttConfigDic;
// 存储每个客户端重连尝试次数的字典
private readonly ConcurrentDictionary<int, int> _reconnectAttempts;
private readonly SemaphoreSlim _reloadSemaphore = new SemaphoreSlim(0);
private readonly SemaphoreSlim _reloadSemaphore = new(0);
private readonly Channel<VariableMqtt> _messageChannel;
/// <summary>
/// 构造函数，注入DataServices。
/// </summary>
/// <param name="dataServices">数据服务实例。</param>
public MqttBackgroundService(DataServices dataServices, MqttRepository mqttRepository)
{
_dataServices = dataServices;
@@ -44,25 +38,34 @@ namespace PMSWPF.Services
_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(); // 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<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)
{
// 断开所有已连接的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();
}
/// <summary>
/// 加载并连接MQTT配置。
/// </summary>
/// <returns>表示异步操作的任务。</returns>
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
}
}
/// <summary>
/// 连接到指定的MQTT代理。
/// </summary>
/// <param name="mqtt">MQTT配置对象。</param>
/// <returns>表示异步操作的任务。</returns>
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}");
}
}
/// <summary>
/// 处理MQTT列表变化事件的回调方法。
/// </summary>
/// <param name="sender">事件发送者。</param>
/// <param name="mqtts">更新后的MQTT配置列表。</param>
private void HandleMqttListChanged(List<Mqtt> mqtts)
{
NlogHelper.Info("Mqtt列表发生了变化，正在重新加载数据..."); // 记录MQTT列表变化信息
// 重新加载MQTT配置和变量数据。
NlogHelper.Info("Mqtt列表发生了变化，正在重新加载数据...");
_reloadSemaphore.Release();
}
}
}
}

39
Services/Processors/MqttPublishProcessor.cs Normal file
View File

@@ -0,0 +1,39 @@
using System.Threading.Tasks;
using PMSWPF.Models;
namespace PMSWPF.Services.Processors
{
/// <summary>
/// 负责将变量数据发布到MQTT的处理器。
/// </summary>
public class MqttPublishProcessor : IVariableProcessor
{
private readonly MqttBackgroundService _mqttBackgroundService;
public MqttPublishProcessor(MqttBackgroundService mqttBackgroundService)
{
_mqttBackgroundService = mqttBackgroundService;
}
/// <summary>
/// 处理单个变量上下文，如果有关联的MQTT配置，则将其推送到发送队列。
/// </summary>
/// <param name="context">包含变量及其元数据的上下文对象。</param>
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);
}
}
}
}