HYZ.MqttClient 0.0.3

基于开源MQTTnet封装MqttClient库,适用net8.0,net9.0,net10.0

快速开始 二选一

// 添加Mqtt服务
   builder.Services.AddHYZMqttServices(builder.Configuration.GetSection("MqttConfig"));

// 添加Mqtt服务
 var mqttConfigOption = builder.Configuration.GetSection("MqttConfig").Get<MqttConfig>() ?? new MqttConfig();
builder.Services.AddHYZMqttServices(mqttConfig =>
{
    mqttConfig.Server = mqttConfigOption.Server;
    mqttConfig.Port = mqttConfigOption.Port;
    mqttConfig.Username = mqttConfigOption.Username;
    mqttConfig.Password = mqttConfigOption.Password;
});

如何使用

    /// <summary>
    /// Mqtt测试
    /// </summary>
    [Route("api/[controller]")]
    [ApiController]
    public class MqttTestController : ControllerBase
    {
        private readonly IMqttClientHelper _mqttClientHelper;
        private readonly ILogger<MqttTestController> _logger;

        /// <summary>
        /// 构造函数
        /// </summary>
        public MqttTestController(
            IMqttClientHelper mqttClientHelper,
            ILogger<MqttTestController> logger,)
        {
            _mqttClientHelper = mqttClientHelper;
            _logger = logger;

            // 订阅消息接收事件
            _mqttClientHelper.MessageReceived += OnMessageReceived;
        }

        /// <summary>
        /// 订阅主题
        /// </summary>
        /// <param name="request"></param>
        /// <returns></returns>
        [HttpPost("subscribe")]
        public async Task<IActionResult> Subscribe([FromBody] SubscribeRequest request)
        {
            if (string.IsNullOrWhiteSpace(request.Topic))
            {
                return BadRequest("Topic is required");
            }

            var success = await _mqttClientHelper.SubscribeAsync(
                request.Topic,
                request.QualityOfServiceLevel ?? MqttQualityOfServiceLevel.AtMostOnce
            );

            return Ok(new { Success = success });
        }

         /// <summary>
        /// 测试Mqtt,需先订阅主题，否则无法收到信息
        /// </summary>
        /// <param name="request"></param>
        /// <returns></returns>
        [HttpPost("test")]
        public async Task<IActionResult> Test([FromBody] TestRequest request)
        {
            if (!_mqttClientHelper.IsConnected)
            {
                // 连接Mqtt
                bool connected = await _mqttClientHelper.ConnectFromConfigAsync();
                if(!connected)
                  return BadRequest(new { Error = "MQTT client is not connected" });
            }

            // 发布测试消息
            var message = $"{{\"message\":\"{request.Message}\",\"timestamp\":\"{DateTime.UtcNow:O}\"}}";

            var publishSuccess = await _mqttClientHelper.PublishAsync(
                request.Topic,
                message,
                 request.QualityOfServiceLevel ?? MqttQualityOfServiceLevel.AtMostOnce
            );

            if (!publishSuccess)
            {
                return StatusCode(500, new { Error = "Failed to publish message" });
            }

            return Ok(new
            {
                Success = true,
                Topic = request.Topic,
                Message = message,
                Timestamp = DateTime.UtcNow
            });
        }

        /// <summary>
        /// 接受消息事件
        /// </summary>
        /// <param name="sender"></param>
        /// <param name="e"></param>
        private void OnMessageReceived(object? sender, MqttApplicationMessageReceivedEventArgs e)
        {
            try
            {
                string payload = e.ApplicationMessage.ConvertPayloadToString() ?? string.Empty;
                _logger.LogInformation("收到 MQTT 消息 - 主题: {Topic}, 内容: {Content}",
                    e.ApplicationMessage.Topic, payload);
            }
            catch (Exception ex)
            {
                _logger.LogError(ex, "处理 MQTT 消息时出错");
            }
        }
     }

  /// <summary>
    /// MQTT 后台服务
    /// </summary>
    public class MqttBackgroundService : BackgroundService
    {
        private readonly IMqttClientHelper _mqttClientHelper;
        private readonly ILogger<MqttBackgroundService> _logger;
        private readonly MqttConfig? _config;

        /// <summary>
        /// 构造函数
        /// </summary>
        public MqttBackgroundService(
            IMqttClientHelper mqttClientHelper,
            ILogger<MqttBackgroundService> logger,
            IOptions<MqttConfig>? config = null)
        {
            _mqttClientHelper = mqttClientHelper ?? throw new ArgumentNullException(nameof(mqttClientHelper));
            _logger = logger ?? throw new ArgumentNullException(nameof(logger));
            _config = config?.Value;

            // 订阅事件
            _mqttClientHelper.MessageReceived += OnMessageReceived;
            _mqttClientHelper.ConnectionStateChanged += OnConnectionStateChanged;
            _mqttClientHelper.ReconnectStatusChanged += OnReconnectStatusChanged;
            
        }

        /// <summary>
        /// 服务执行
        /// </summary>
        /// <param name="stoppingToken"></param>
        /// <returns></returns>
        protected override async Task ExecuteAsync(CancellationToken stoppingToken)
        {
            _logger.LogInformation("MQTT 后台服务启动");
            // 连接到 MQTT 服务器
            while (!stoppingToken.IsCancellationRequested)
            {
                try
                {
                    if (_config != null && !string.IsNullOrEmpty(_config.Server))
                    {
                        _logger.LogInformation("正在连接到 MQTT 服务器: {Server}:{Port}",
                            _config.Server, _config.Port);

                        var connected = await _mqttClientHelper.ConnectFromConfigAsync(stoppingToken);

                        if (connected)
                        {
                            _logger.LogInformation("MQTT 连接成功");

                            // 订阅额外的主题（如果有）
                            await SubscribeAdditionalTopicsAsync(stoppingToken);

                            // 保持运行状态
                            while (!stoppingToken.IsCancellationRequested && _mqttClientHelper.IsConnected)
                            {
                                await Task.Delay(TimeSpan.FromSeconds(10), stoppingToken);
                            }

                            if (!_mqttClientHelper.IsConnected)
                            {
                                _logger.LogWarning("MQTT 连接断开，等待重连...");
                                await Task.Delay(TimeSpan.FromSeconds(5), stoppingToken);
                            }
                        }
                        else
                        {
                            _logger.LogWarning("MQTT 连接失败，5 秒后重试...");
                            await Task.Delay(TimeSpan.FromSeconds(5), stoppingToken);
                        }
                    }
                    else
                    {
                        _logger.LogWarning("MQTT 配置无效，30 秒后重试...");
                        await Task.Delay(TimeSpan.FromSeconds(30), stoppingToken);
                    }
                }
                catch (OperationCanceledException)
                {
                    break;
                }
                catch (Exception ex)
                {
                    _logger.LogError(ex, "MQTT 后台服务发生错误");
                    await Task.Delay(TimeSpan.FromSeconds(10), stoppingToken);
                }
            }

            _logger.LogInformation("MQTT 后台服务停止");
        }

        /// <summary>
        /// 停止服务
        /// </summary>
        /// <param name="cancellationToken"></param>
        /// <returns></returns>
        public override async Task StopAsync(CancellationToken cancellationToken)
        {
            _logger.LogInformation("正在停止 MQTT 后台服务...");
            // 断开 MQTT 连接
            await _mqttClientHelper.DisconnectAsync(cancellationToken);

            await base.StopAsync(cancellationToken);
        }

        /// <summary>
        /// 订阅额外的主题
        /// </summary>
        private async Task SubscribeAdditionalTopicsAsync(CancellationToken cancellationToken)
        {
            // TODO：这里可以添加额外的主题订阅逻辑
            // 例如：从数据库或其他配置源获取主题
            var additionalTopics = new List<string>();

            if (additionalTopics.Count != 0)
            {
                _logger.LogInformation("开始订阅额外的主题...");
                await _mqttClientHelper.SubscribeMultipleAsync(additionalTopics, cancellationToken: cancellationToken);
            }
        }

        /// <summary>
        /// 消息接收事件处理
        /// </summary>
        private void OnMessageReceived(object? sender, MqttApplicationMessageReceivedEventArgs e)
        {
            try
            {
                string payload = e.ApplicationMessage.ConvertPayloadToString() ?? string.Empty;

                _logger.LogDebug("收到消息 - 主题: {Topic}, 内容: {Content}",
                    e.ApplicationMessage.Topic, payload);

                // TODO：处理订阅收到的消息
            }
            catch (Exception ex)
            {
                _logger.LogError(ex, "处理消息时出错");
            }
        }

        /// <summary>
        /// 连接状态变化事件处理
        /// </summary>
        private void OnConnectionStateChanged(object? sender, bool isConnected)
        {
            if (isConnected)
            {
                _logger.LogInformation("MQTT 连接已建立");
            }
            else
            {
                _logger.LogWarning("MQTT 连接已断开");
            }
        }


        /// <summary>
        /// 重连状态变化事件处理
        /// </summary>
        /// <param name="reconnectStatus"></param>
        /// <param name="sender"></param>
        public void OnReconnectStatusChanged(object? sender, ReconnectStatus reconnectStatus)
        {
            _logger.LogInformation($"MQTT 连接状态：{reconnectStatus.ToString()}");
        }
    }