MQTT全局服务订阅，基本功能创建，标签打印等功能基本实现
This commit is contained in:
424
ZR.Common/MqttHelper/MqttService.cs
Normal file
424
ZR.Common/MqttHelper/MqttService.cs
Normal file
@@ -0,0 +1,424 @@
|
||||
using System;
|
||||
using System.Linq;
|
||||
using System.Text;
|
||||
using System.Threading;
|
||||
using System.Threading.Tasks;
|
||||
using Infrastructure.Attribute;
|
||||
using Microsoft.Extensions.Configuration;
|
||||
using Microsoft.Extensions.Hosting;
|
||||
using Microsoft.Extensions.Logging;
|
||||
using MQTTnet;
|
||||
using MQTTnet.Client;
|
||||
using MQTTnet.Protocol;
|
||||
|
||||
namespace ZR.Common.MqttHelper
|
||||
{
|
||||
public class MqttService : IHostedService, IDisposable
|
||||
{
|
||||
private readonly ILogger<MqttService> _logger;
|
||||
private readonly IConfiguration _configuration;
|
||||
private readonly IMqttClient _mqttClient;
|
||||
private readonly MyMqttConfig _mqttConfig; // 注入配置类
|
||||
private readonly SemaphoreSlim _connectionLock = new SemaphoreSlim(1, 1);
|
||||
private readonly CancellationTokenSource _disposeCts = new CancellationTokenSource();
|
||||
private Timer _reconnectTimer;
|
||||
private bool _disposed = false;
|
||||
private int _reconnectAttempts = 0;
|
||||
|
||||
// 配置常量
|
||||
private const string MqttConfigSection = "MqttConfig";
|
||||
private const string TopicsConfigKey = "Topics";
|
||||
private const int DefaultReconnectDelaySeconds = 5;
|
||||
private const int MaxReconnectAttempts = 10;
|
||||
private const int MaxReconnectDelaySeconds = 60;
|
||||
|
||||
public MqttService(
|
||||
ILogger<MqttService> logger,
|
||||
IConfiguration configuration,
|
||||
MyMqttConfig mqttConfig
|
||||
)
|
||||
{
|
||||
_logger = logger ?? throw new ArgumentNullException(nameof(logger));
|
||||
_configuration =
|
||||
configuration ?? throw new ArgumentNullException(nameof(configuration));
|
||||
_mqttConfig = mqttConfig ?? throw new ArgumentNullException(nameof(mqttConfig)); // 注入配置类
|
||||
// 创建MQTT客户端
|
||||
var factory = new MqttFactory();
|
||||
_mqttClient = factory.CreateMqttClient();
|
||||
|
||||
// 注册事件处理程序
|
||||
_mqttClient.ConnectedAsync += OnConnectedAsync;
|
||||
_mqttClient.DisconnectedAsync += OnDisconnectedAsync;
|
||||
_mqttClient.ApplicationMessageReceivedAsync += OnMessageReceivedAsync;
|
||||
_logger.LogInformation($"MqttService 实例已创建，哈希值: {GetHashCode()}");
|
||||
}
|
||||
|
||||
public async Task StartAsync(CancellationToken cancellationToken)
|
||||
{
|
||||
_logger.LogInformation("MQTT服务正在启动...");
|
||||
|
||||
try
|
||||
{
|
||||
await ConnectAsync(cancellationToken);
|
||||
_logger.LogInformation("MQTT服务已成功启动");
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger.LogCritical(ex, "MQTT服务启动失败，将在后台尝试重连");
|
||||
ScheduleReconnect(); // 即使启动失败，也应该尝试重连
|
||||
}
|
||||
}
|
||||
|
||||
public async Task StopAsync(CancellationToken cancellationToken)
|
||||
{
|
||||
_logger.LogInformation("MQTT服务正在停止...");
|
||||
|
||||
// 取消所有计划的重连
|
||||
_disposeCts.Cancel();
|
||||
_reconnectTimer?.Dispose();
|
||||
|
||||
await DisconnectAsync(cancellationToken);
|
||||
_logger.LogInformation("MQTT服务已停止");
|
||||
}
|
||||
|
||||
private async Task ConnectAsync(CancellationToken cancellationToken)
|
||||
{
|
||||
await _connectionLock.WaitAsync(cancellationToken);
|
||||
try
|
||||
{
|
||||
if (_mqttClient.IsConnected)
|
||||
{
|
||||
_logger.LogDebug("MQTT客户端已连接，跳过连接操作");
|
||||
return;
|
||||
}
|
||||
// 直接使用MyMqttConfig获取客户端选项
|
||||
var options = _mqttConfig.BuildMqttClientOptions();
|
||||
_logger.LogInformation($"正在连接到MQTT代理服务器 {options.ChannelOptions}...");
|
||||
try
|
||||
{
|
||||
await _mqttClient.ConnectAsync(options, cancellationToken);
|
||||
_reconnectAttempts = 0; // 重置重连计数
|
||||
_logger.LogInformation("已成功连接到MQTT代理服务器");
|
||||
}
|
||||
catch (OperationCanceledException)
|
||||
{
|
||||
_logger.LogWarning("MQTT连接操作被取消");
|
||||
throw;
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_reconnectAttempts++;
|
||||
_logger.LogError(
|
||||
ex,
|
||||
$"连接MQTT代理服务器失败 (尝试次数: {_reconnectAttempts}/{MaxReconnectAttempts})"
|
||||
);
|
||||
|
||||
if (_reconnectAttempts >= MaxReconnectAttempts)
|
||||
{
|
||||
_logger.LogCritical("达到最大重连次数，停止尝试");
|
||||
return;
|
||||
}
|
||||
|
||||
ScheduleReconnect();
|
||||
throw;
|
||||
}
|
||||
}
|
||||
finally
|
||||
{
|
||||
_connectionLock.Release();
|
||||
}
|
||||
}
|
||||
|
||||
private MqttClientOptions BuildMqttClientOptions(IConfigurationSection config)
|
||||
{
|
||||
var builder = new MqttClientOptionsBuilder()
|
||||
.WithClientId(config["ClientId"] ?? Guid.NewGuid().ToString())
|
||||
.WithTcpServer(config["Server"], config.GetValue<int>("Port", 1883))
|
||||
.WithCleanSession();
|
||||
return builder.Build();
|
||||
}
|
||||
|
||||
private async Task DisconnectAsync(CancellationToken cancellationToken)
|
||||
{
|
||||
if (!_mqttClient.IsConnected)
|
||||
{
|
||||
_logger.LogDebug("MQTT客户端未连接，跳过断开操作");
|
||||
return;
|
||||
}
|
||||
|
||||
try
|
||||
{
|
||||
await _mqttClient.DisconnectAsync(cancellationToken: cancellationToken);
|
||||
_logger.LogInformation("已成功断开与MQTT代理服务器的连接");
|
||||
}
|
||||
catch (OperationCanceledException)
|
||||
{
|
||||
_logger.LogWarning("MQTT断开操作被取消");
|
||||
throw;
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger.LogError(ex, "断开MQTT连接时出错");
|
||||
// 即使断开连接失败，也应该释放资源
|
||||
CleanupClientResources();
|
||||
}
|
||||
}
|
||||
|
||||
private Task OnConnectedAsync(MqttClientConnectedEventArgs e)
|
||||
{
|
||||
_logger.LogInformation("MQTT连接已建立");
|
||||
return SubscribeToTopicsAsync();
|
||||
}
|
||||
|
||||
private async Task OnDisconnectedAsync(MqttClientDisconnectedEventArgs e)
|
||||
{
|
||||
_logger.LogWarning($"MQTT连接已断开 - 客户端是否之前已连接: {e.ClientWasConnected}");
|
||||
|
||||
if (_disposed || _disposeCts.IsCancellationRequested)
|
||||
{
|
||||
_logger.LogDebug("MQTT断开连接是预期行为，正在关闭服务");
|
||||
return;
|
||||
}
|
||||
|
||||
if (e.ClientWasConnected)
|
||||
{
|
||||
ScheduleReconnect();
|
||||
}
|
||||
}
|
||||
|
||||
private async Task OnMessageReceivedAsync(MqttApplicationMessageReceivedEventArgs e)
|
||||
{
|
||||
try
|
||||
{
|
||||
var payload = Encoding.UTF8.GetString(e.ApplicationMessage.PayloadSegment);
|
||||
_logger.LogInformation(
|
||||
$"收到MQTT消息 - 主题: {e.ApplicationMessage.Topic}, QoS: {e.ApplicationMessage.QualityOfServiceLevel}"
|
||||
);
|
||||
|
||||
// 消息处理委托给专用的处理器
|
||||
await ProcessMessageAsync(e.ApplicationMessage.Topic, payload);
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger.LogError(ex, "处理MQTT消息时出错");
|
||||
}
|
||||
}
|
||||
|
||||
private async Task ProcessMessageAsync(string topic, string payload)
|
||||
{
|
||||
try
|
||||
{
|
||||
// 这里可以根据主题路由消息到不同的处理程序
|
||||
switch (topic)
|
||||
{
|
||||
case string t when t.StartsWith("devices/"):
|
||||
await HandleDeviceMessage(topic, payload);
|
||||
break;
|
||||
case "system/alert":
|
||||
await HandleSystemAlert(payload);
|
||||
break;
|
||||
default:
|
||||
_logger.LogDebug($"未处理的MQTT主题: {topic}");
|
||||
break;
|
||||
}
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger.LogError(ex, $"处理主题 '{topic}' 的消息时出错");
|
||||
}
|
||||
}
|
||||
|
||||
private async Task SubscribeToTopicsAsync()
|
||||
{
|
||||
if (!_mqttClient.IsConnected)
|
||||
{
|
||||
_logger.LogWarning("无法订阅主题：MQTT客户端未连接");
|
||||
return;
|
||||
}
|
||||
|
||||
try
|
||||
{
|
||||
// 获取配置类生成的订阅选项
|
||||
var subscribeOptions = _mqttConfig.GetSubscribeToTopicsAsync();
|
||||
|
||||
if (subscribeOptions == null || !subscribeOptions.TopicFilters.Any())
|
||||
{
|
||||
_logger.LogInformation("没有配置要订阅的MQTT主题");
|
||||
return;
|
||||
}
|
||||
|
||||
_logger.LogInformation(
|
||||
$"正在订阅MQTT主题: {string.Join(", ", subscribeOptions.TopicFilters.Select(f => f.Topic))}"
|
||||
);
|
||||
|
||||
// 使用强类型的订阅选项进行订阅
|
||||
var result = await _mqttClient.SubscribeAsync(subscribeOptions);
|
||||
|
||||
foreach (var item in result.Items)
|
||||
{
|
||||
_logger.LogInformation(
|
||||
$"订阅主题 '{item.TopicFilter.Topic}' 结果: {item.ResultCode}"
|
||||
);
|
||||
}
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger.LogError(ex, "订阅MQTT主题时出错");
|
||||
ScheduleReconnect();
|
||||
}
|
||||
}
|
||||
|
||||
private void ScheduleReconnect()
|
||||
{
|
||||
// 实现指数退避算法
|
||||
var delaySeconds = Math.Min(
|
||||
DefaultReconnectDelaySeconds * (int)Math.Pow(2, Math.Min(_reconnectAttempts, 5)),
|
||||
MaxReconnectDelaySeconds
|
||||
);
|
||||
|
||||
_logger.LogInformation($"计划在 {delaySeconds} 秒后尝试重新连接MQTT代理服务器");
|
||||
|
||||
// 使用Timer替代Task.Run，更好地控制资源
|
||||
_reconnectTimer?.Dispose();
|
||||
_reconnectTimer = new Timer(
|
||||
async _ =>
|
||||
{
|
||||
if (_disposed || _disposeCts.IsCancellationRequested)
|
||||
{
|
||||
_logger.LogDebug("跳过重连：服务正在关闭");
|
||||
return;
|
||||
}
|
||||
|
||||
try
|
||||
{
|
||||
await ConnectAsync(_disposeCts.Token);
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger.LogError(ex, "计划的MQTT重连失败");
|
||||
}
|
||||
},
|
||||
null,
|
||||
TimeSpan.FromSeconds(delaySeconds),
|
||||
Timeout.InfiniteTimeSpan
|
||||
);
|
||||
}
|
||||
|
||||
private Task HandleDeviceMessage(string topic, string payload)
|
||||
{
|
||||
_logger.LogInformation($"处理设备消息: {topic} - {payload}");
|
||||
// 这里添加设备消息处理逻辑
|
||||
return Task.CompletedTask;
|
||||
}
|
||||
|
||||
private Task HandleSystemAlert(string payload)
|
||||
{
|
||||
_logger.LogWarning($"系统警报: {payload}");
|
||||
// 这里添加系统警报处理逻辑
|
||||
return Task.CompletedTask;
|
||||
}
|
||||
|
||||
public async Task PublishAsync(
|
||||
string topic,
|
||||
string payload,
|
||||
MqttQualityOfServiceLevel qos = MqttQualityOfServiceLevel.AtLeastOnce,
|
||||
bool retain = false
|
||||
)
|
||||
{
|
||||
if (string.IsNullOrEmpty(topic))
|
||||
{
|
||||
throw new ArgumentNullException(nameof(topic), "MQTT主题不能为空");
|
||||
}
|
||||
|
||||
if (payload == null)
|
||||
{
|
||||
throw new ArgumentNullException(nameof(payload), "MQTT消息内容不能为空");
|
||||
}
|
||||
|
||||
await _connectionLock.WaitAsync(_disposeCts.Token);
|
||||
try
|
||||
{
|
||||
if (!_mqttClient.IsConnected)
|
||||
{
|
||||
_logger.LogWarning("发布消息前需要连接MQTT代理服务器");
|
||||
//await ConnectAsync(_disposeCts.Token);
|
||||
throw new Exception("发布消息前需要连接MQTT代理服务器");
|
||||
}
|
||||
|
||||
var message = new MqttApplicationMessageBuilder()
|
||||
.WithTopic(topic)
|
||||
.WithPayload(payload)
|
||||
.WithQualityOfServiceLevel(qos)
|
||||
.WithRetainFlag(retain)
|
||||
.Build();
|
||||
|
||||
_logger.LogDebug($"准备发布MQTT消息 - 主题: {topic}, QoS: {qos}, 保留: {retain}");
|
||||
|
||||
try
|
||||
{
|
||||
var result = await _mqttClient.PublishAsync(message, _disposeCts.Token);
|
||||
_logger.LogInformation($"已发布MQTT消息 - 主题: {topic}, 结果: {result.ReasonCode}");
|
||||
}
|
||||
catch (OperationCanceledException)
|
||||
{
|
||||
_logger.LogWarning("MQTT发布操作被取消");
|
||||
throw;
|
||||
}
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger.LogError(ex, $"发布MQTT消息失败 - 主题: {topic}");
|
||||
throw;
|
||||
}
|
||||
finally
|
||||
{
|
||||
_connectionLock.Release();
|
||||
}
|
||||
}
|
||||
|
||||
private void CleanupClientResources()
|
||||
{
|
||||
try
|
||||
{
|
||||
_mqttClient.ApplicationMessageReceivedAsync -= OnMessageReceivedAsync;
|
||||
_mqttClient.ConnectedAsync -= OnConnectedAsync;
|
||||
_mqttClient.DisconnectedAsync -= OnDisconnectedAsync;
|
||||
|
||||
if (_mqttClient.IsConnected)
|
||||
{
|
||||
_mqttClient.DisconnectAsync().GetAwaiter().GetResult();
|
||||
}
|
||||
|
||||
_mqttClient.Dispose();
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger.LogError(ex, "清理MQTT客户端资源时出错");
|
||||
}
|
||||
}
|
||||
|
||||
public void Dispose()
|
||||
{
|
||||
Dispose(true);
|
||||
GC.SuppressFinalize(this);
|
||||
}
|
||||
|
||||
protected virtual void Dispose(bool disposing)
|
||||
{
|
||||
if (!_disposed)
|
||||
{
|
||||
if (disposing)
|
||||
{
|
||||
_disposeCts.Cancel();
|
||||
_reconnectTimer?.Dispose();
|
||||
_connectionLock.Dispose();
|
||||
CleanupClientResources();
|
||||
_disposeCts.Dispose();
|
||||
}
|
||||
|
||||
_disposed = true;
|
||||
_logger.LogDebug("MQTT服务已释放所有资源");
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
97
ZR.Common/MqttHelper/MyMqttConfig.cs
Normal file
97
ZR.Common/MqttHelper/MyMqttConfig.cs
Normal file
@@ -0,0 +1,97 @@
|
||||
using Infrastructure.Attribute;
|
||||
using Microsoft.Extensions.Configuration;
|
||||
using Microsoft.Extensions.Logging;
|
||||
using MQTTnet;
|
||||
using MQTTnet.Client;
|
||||
using MQTTnet.Packets;
|
||||
using MQTTnet.Protocol;
|
||||
using MQTTnet.Server;
|
||||
using System;
|
||||
using System.Collections.Generic;
|
||||
using System.Linq;
|
||||
using System.Text;
|
||||
using System.Threading.Tasks;
|
||||
using static Org.BouncyCastle.Math.EC.ECCurve;
|
||||
|
||||
namespace ZR.Common.MqttHelper
|
||||
{
|
||||
[AppService(ServiceLifetime = LifeTime.Singleton)]
|
||||
public class MyMqttConfig
|
||||
{
|
||||
// 配置常量
|
||||
private const string MqttConfigSection = "MqttConfig";
|
||||
private const string TopicsConfigKey = "Topics";
|
||||
private const int DefaultReconnectDelaySeconds = 5;
|
||||
private const int MaxReconnectAttempts = 10;
|
||||
private const int MaxReconnectDelaySeconds = 60;
|
||||
|
||||
private readonly ILogger<MyMqttConfig> _logger;
|
||||
private readonly IConfiguration _configuration;
|
||||
|
||||
public MyMqttConfig(ILogger<MyMqttConfig> logger, IConfiguration configuration)
|
||||
{
|
||||
_logger = logger ?? throw new ArgumentNullException(nameof(logger));
|
||||
_configuration = configuration;
|
||||
}
|
||||
|
||||
public MqttClientOptions BuildMqttClientOptions()
|
||||
{
|
||||
var mqttConfig = _configuration.GetSection(MqttConfigSection);
|
||||
var builder = new MqttClientOptionsBuilder()
|
||||
.WithClientId(mqttConfig["ClientId"] ?? Guid.NewGuid().ToString())
|
||||
.WithTcpServer(mqttConfig["Server"], mqttConfig.GetValue<int>("Port", 1883))
|
||||
.WithCleanSession();
|
||||
return builder.Build();
|
||||
}
|
||||
|
||||
public MqttClientSubscribeOptions GetSubscribeToTopicsAsync()
|
||||
{
|
||||
var topicsSection = _configuration.GetSection($"{MqttConfigSection}:{TopicsConfigKey}");
|
||||
var topicConfigs = topicsSection.Get<List<TopicSubscriptionConfig>>() ?? new List<TopicSubscriptionConfig>();
|
||||
|
||||
if (!topicConfigs.Any())
|
||||
{
|
||||
_logger.LogInformation("没有配置要订阅的MQTT主题");
|
||||
return null; // 或返回空配置，根据业务需求决定
|
||||
}
|
||||
|
||||
var subscribeOptionsBuilder = new MqttClientSubscribeOptionsBuilder();
|
||||
|
||||
foreach (var config in topicConfigs)
|
||||
{
|
||||
var topicFilter = config.ToMqttTopicFilter();
|
||||
subscribeOptionsBuilder.WithTopicFilter(topicFilter); // 直接添加单个主题过滤器
|
||||
}
|
||||
|
||||
return subscribeOptionsBuilder.Build();
|
||||
}
|
||||
|
||||
|
||||
|
||||
// 主题订阅配置类
|
||||
private class TopicSubscriptionConfig
|
||||
{
|
||||
public string Topic { get; set; }
|
||||
public MqttQualityOfServiceLevel QualityOfServiceLevel { get; set; } = MqttQualityOfServiceLevel.AtLeastOnce;
|
||||
public bool NoLocal { get; set; } = false;
|
||||
public bool RetainAsPublished { get; set; } = false;
|
||||
public MqttRetainHandling RetainHandling { get; set; } = MqttRetainHandling.SendAtSubscribe;
|
||||
|
||||
public MqttTopicFilter ToMqttTopicFilter()
|
||||
{
|
||||
return new MqttTopicFilterBuilder()
|
||||
.WithTopic(Topic)
|
||||
.WithQualityOfServiceLevel(QualityOfServiceLevel)
|
||||
.WithNoLocal(NoLocal)
|
||||
.WithRetainAsPublished(RetainAsPublished)
|
||||
.WithRetainHandling(RetainHandling)
|
||||
.Build();
|
||||
}
|
||||
|
||||
public override string ToString()
|
||||
{
|
||||
return $"{Topic}@QoS{QualityOfServiceLevel}";
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user