1. 编程号首页
  2. 编程基础

MQTT学习笔记（C#）-MQTTnet

编程小号 • • 编程基础

MQTT学习笔记（C#）-MQTTnet代码地址：https://gitee.com/qq28069933146_admin/csharp_networkprotocol_research 演示地址：C#-MQTT调用示例演示 一，什么是MQTT： MQTT(消息队列遥测传输)是IBM开发的即时通讯协议，是一个基于客户端-服务器的消息发布

MQTT学习笔记（C#）-MQTTnet

  代码地址：https://gitee.com/qq28069933146_admin/csharp_networkprotocol_research

  演示地址：C#-MQTT调用示例演示

一，什么是MQTT：

　　MQTT(消息队列遥测传输)是IBM开发的即时通讯协议，是一个基于客户端–服务器的消息发布/订阅极其轻量级的消息传输协议。

　　它工作在 TCP/IP协议族上，专门为网络受限设备、低宽带以及高延迟和不可靠的网络而设计的（传输特点：至多一次、至少一次、只有一次）。使其在物联网、小型设备、移动应用等方面有较广泛的应用。

 1.MQTT报文：

MQTT学习笔记（C#）-MQTTnet

　　详情见：https://www.cnblogs.com/hayasi/p/7743356.html

 2.MQTT三种身份：发布者（客户端）、代理（服务器）、订阅者（客户端）。

 MQTT学习笔记（C#）-MQTTnet

二，示例

　　可以选择MQTTnet包或者DotNetty.Codecs.Mqtt包进行学习，这里我们选用Mqtt进行学习。

　　VS2022

　　Net5

　　MQTTnet 4.1.4.563

 1.服务端:

　　①服务器对象初始化（new MqttFactory().CreateMqttServer(new MqttServerOptionsBuilder().Build());）

  MqttServerOptionsBuilder mqttServerOptionsBuilder = new MqttServerOptionsBuilder();  // MQTT服务器配置
  mqttServerOptionsBuilder.WithDefaultEndpoint();
  mqttServerOptionsBuilder.WithDefaultEndpointBoundIPAddress(IPAddress.Parse(ip));  // 设置Server的IP
  mqttServerOptionsBuilder.WithDefaultEndpointPort(port);                           // 设置Server的端口号
  //mqttServerOptionsBuilder.WithEncryptedEndpointPort(port);                        // 使用加密的端点端口
  mqttServerOptionsBuilder.WithPersistentSessions(withPersistentSessions);  // 持续会话
  mqttServerOptionsBuilder.WithConnectionBacklog(2000);                     // 最大连接数
  //mqttServerOptionsBuilder.WithConnectionValidator(c =>  // 鉴权-方法失效
  //{
  //    if (c.Username != uName || c.Password != uPwd)
  //    {
  //        c.ReasonCode = MqttConnectReasonCode.BadUserNameOrPassword;
  //    }
  //})
 
  MqttServerOptions mqttServerOptions = mqttServerOptionsBuilder.Build();
  _MqttServer = new MqttFactory().CreateMqttServer(mqttServerOptions);  // 创建服务（配置）

　　　补充：MqttServerOptionsBuilder的属性(注：有些方法4.1.4.563版本不支持了)

函数名 功能说明
Build() 构建配置参数
WithApplicationMessageInterceptor() 允许处理来自客户端的所有已发布消息
WithClientId() 服务端发布消息时使用的ClientId
WithConnectionBacklog() 设置要保留的连接数
WithConnectionValidator() 验证连接
WithDefaultCommunicationTimeout() 设置默认的通信超时
WithDefaultEndpoint() 使用默认端点
WithDefaultEndpointBoundIPAddress() 使用默认端点IPv4地址
WithDefaultEndpointBoundIPV6Address() 使用默认端点IPv6地址
WithDefaultEndpointPort() 使用默认端点端口
WithEncryptedEndpoint() 使用加密的端点
WithEncryptedEndpointBoundIPAddress() 使用加密的端点IPv4地址
WithEncryptedEndpointBoundIPV6Address() 使用加密的端点IPv6地址
WithEncryptedEndpointPort() 使用加密的端点端口
WithEncryptionCertificate() 使用证书进行SSL连接
WithEncryptionSslProtocol() 使用SSL协议级别
WithMaxPendingMessagesPerClient() 每个客户端允许最多未决消息
WithPersistentSessions() 保持会话
WithStorage() 使用存储
WithSubscriptionInterceptor() 允许处理来自客户端的所有订阅
WithoutDefaultEndpoint() 禁用默认端点
WithoutEncryptedEndpoint() 禁用默认(SSL)端点

　　②服务器开启（await mqttServer.StartAsync()）

  await _MqttServer.StartAsync();  // 开启服务

  ③服务器关闭（await _MqttServer.StopAsync()）

  foreach (var clientStatus in _MqttServer.GetClientsAsync().Result)
  {
      await clientStatus.DisconnectAsync();
  }
  await _MqttServer.StopAsync();
  _MqttServer = null;

　　④是否对客户端的进行验证（账号密码-原方法失效）

  //mqttServerOptionsBuilder.WithConnectionValidator(c =>  // 鉴权-方法失效
  //{
  //    if (c.Username != uName || c.Password != uPwd)
  //    {
  //        c.ReasonCode = MqttConnectReasonCode.BadUserNameOrPassword;
  //    }
  //})

　　⑤给客户端发送数据（原方法失效）

  /// <summary>
  /// 发送消息-未写（原方法失效）
  /// </summary>
  /// <param name="Topic">主题</param>
  /// <param name="msg">消息</param>
  /// <returns></returns>
  public Task SedMessage(string Topic, string msg)
  {
      try
      {
          //var clients = _MqttServer.GetClientsAsync().Result;

          //foreach (var client in clients)
          //{

          //}
      }
      catch { }
      return Task.CompletedTask;
  }

　　⑥获取所有的客户端（_MqttServer.GetClientsAsync().Result.ToList()）

  _MqttServer.GetClientsAsync().Result.ToList()

  ⑦服务器开启/关闭事件（StartedAsync与StoppedAsync）

  _MqttServer.StartedAsync += StartedHandle;  // 服务器开启事件

  /// <summary>
  /// 开启Server的处理程序
  /// </summary>
  private Task StartedHandle(EventArgs arg)
  {
      return Task.CompletedTask;
  }

  _MqttServer.StoppedAsync += StoppedHandle;  // 服务器关闭事件

  /// <summary>
  /// 关闭Server的处理程序
  /// </summary>
  private Task StoppedHandle(EventArgs arg)
  {
      return Task.CompletedTask;
  }

　　⑧客户端连接/断开的处理事件

  _MqttServer.ClientConnectedAsync += ClientConnectedHandle;        // 设置客户端连接成功后的处理程序

  /// <summary>
  /// 设置客户端连接成功后的处理程序
  /// </summary>
  private Task ClientConnectedHandle(ClientConnectedEventArgs arg)
  {
      return Task.CompletedTask;
  }

  _MqttServer.ClientDisconnectedAsync += ClientDisconnectedHandle;  // 设置客户端断开后的处理程序

  /// <summary>
  /// 设置客户端断开后的处理程序
  /// </summary>
  private Task ClientDisconnectedHandle(ClientDisconnectedEventArgs arg)
  {
      return Task.CompletedTask;
  }

　　⑨消息被订阅/被退订事件

  _MqttServer.ClientSubscribedTopicAsync += ClientSubscribedTopicHandle;      // 设置消息订阅通知
  _MqttServer.ClientUnsubscribedTopicAsync += ClientUnsubscribedTopicHandle;  // 设置消息退订通知

  /// <summary>
  /// 设置消息订阅通知
  /// </summary>
  private Task ClientSubscribedTopicHandle(ClientSubscribedTopicEventArgs arg)
  {
      //if (!arg.Equals("admin"))
      //{
      //    var client = clients.Where(a => a.Id == arg.ClientId).FirstOrDefault();
      //    client?.DisconnectAsync();

      //    return Task.CompletedTask;
      //}
      return Task.CompletedTask;
  }

  /// <summary>
  /// 设置消息退订通知
  /// </summary>
  private Task ClientUnsubscribedTopicHandle(ClientUnsubscribedTopicEventArgs arg)
  {
      return Task.CompletedTask;
  }

  ⑩接收到消息时的处理程序

  _MqttServer.ApplicationMessageNotConsumedAsync += ApplicationMessageNotConsumedHandle;  // 设置消息处理程序

  /// <summary>
  /// 设置消息处理程序
  /// </summary>
  private Task ApplicationMessageNotConsumedHandle(ApplicationMessageNotConsumedEventArgs arg)
  {
      _Callback?.Invoke(new ResultData_MQTT()
      {
          ResultCode = -1,
          ResultMsg = DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss.fff") + $">>>客户端：'{arg.SenderId}'发布了消息：主题：'{arg.ApplicationMessage.Topic}'！内容：'{Encoding.UTF8.GetString(arg.ApplicationMessage.Payload)}'；服务质量：{arg.ApplicationMessage.QualityOfServiceLevel}；保留：{arg.ApplicationMessage.Retain}"
      });

      return Task.CompletedTask;
  }

 2.客户端：

　　①客户端对象初始化并连接（_MqttClient.ConnectAsync(new MqttClientOptionsBuilder().Build())）

  MqttClientOptionsBuilder mqttClientOptionsBuilder = new MqttClientOptionsBuilder();
  mqttClientOptionsBuilder.WithTcpServer(mqttServerUrl, port);          // 设置MQTT服务器地址
  if (!string.IsNullOrEmpty(userName))
  {
      mqttClientOptionsBuilder.WithCredentials(userName, userPassword);  // 设置鉴权参数
  }
  mqttClientOptionsBuilder.WithClientId(Guid.NewGuid().ToString("N"));  // 设置客户端序列号
  MqttClientOptions options = mqttClientOptionsBuilder.Build();
 
  _MqttClient = new MqttFactory().CreateMqttClient();
  _MqttClient.ConnectedAsync += ConnectedHandle;        // 服务器连接事件
  _MqttClient.DisconnectedAsync += DisconnectedHandle;  // 服务器断开事件（可以写入重连事件）
  _MqttClient.ApplicationMessageReceivedAsync += ApplicationMessageReceivedHandle;  // 发送消息事件
  await _MqttClient.ConnectAsync(options);  // 连接

　　②与服务器断开连接（await _MqttClient.DisconnectAsync()）

  await _MqttClient.DisconnectAsync();
  _MqttClient.Dispose();
  _MqttClient = null;

  ③与服务器重新连接（await _MqttClient.ReconnectAsync()）

  await _MqttClient.ReconnectAsync();

  ④订阅与退订主题（）

  MqttTopicFilter topicFilter = new MqttTopicFilterBuilder().WithTopic(topic).Build();
  await _MqttClient.SubscribeAsync(topicFilter, CancellationToken.None);  // 订阅

  await MqttClientExtensions.UnsubscribeAsync(_MqttClient, topic, CancellationToken.None);  // 退订

  ⑤发布消息

  MqttApplicationMessageBuilder mqttApplicationMessageBuilder = new MqttApplicationMessageBuilder();  // 设置内容
  mqttApplicationMessageBuilder.WithTopic(topic);          // 主题
  mqttApplicationMessageBuilder.WithPayload(msg);          // 信息
  mqttApplicationMessageBuilder.WithRetainFlag(retained);  // 保留
  MqttApplicationMessage messageObj = mqttApplicationMessageBuilder.Build();

  await _MqttClient.PublishAsync(messageObj, CancellationToken.None);  // 发送

  ⑥与服务器连接/断开事件

  _MqttClient.ConnectedAsync += ConnectedHandle;        // 服务器连接事件
  _MqttClient.DisconnectedAsync += DisconnectedHandle;  // 服务器断开事件（可以写入重连事件）

  /// <summary>
  /// 服务器连接事件
  /// </summary>
  private Task ConnectedHandle(MqttClientConnectedEventArgs arg)
  {
      return Task.CompletedTask;
  }

  /// <summary>
  /// 服务器断开事件（可以写入重连事件）
  /// </summary>
  private Task DisconnectedHandle(MqttClientDisconnectedEventArgs arg)
  {
      return Task.CompletedTask;
  }

  ⑦订阅/退订事件

  // 略-原方法失效

  ⑧发送消息事件

  _MqttClient.ApplicationMessageReceivedAsync += ApplicationMessageReceivedHandle;  // 发送消息事件

  /// <summary>
  /// 发送消息事件
  /// </summary>
  private Task ApplicationMessageReceivedHandle(MqttApplicationMessageReceivedEventArgs arg)
  {
      string resultMsg = DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss.fff") + $">>>MQTTClient'{arg.ClientId}'内容：'{Encoding.UTF8.GetString(arg.ApplicationMessage.Payload)}'；主题：'{arg.ApplicationMessage.Topic}'，消息等级Qos：[{arg.ApplicationMessage.QualityOfServiceLevel}]，是否保留：[{arg.ApplicationMessage.Retain}]",

      return Task.CompletedTask;
  }

三、MQTTHelper

/**
*┌──────────────────────────────────────────────────────────────┐
*│　描    述：MQTT通讯相关的工具类（MQTTnet 4.1.4.563）
*│　作    者：执笔小白                                              
*│　版    本：1.0                                       
*│　创建时间：2023-3-18 10:40:56                            
*└──────────────────────────────────────────────────────────────┘
*┌──────────────────────────────────────────────────────────────┐
*│　命名空间: MqttnetServerWin                               
*│　类    名：MQTTHelper                                     
*└──────────────────────────────────────────────────────────────┘
*/
using MQTTnet;
using MQTTnet.Client;
using MQTTnet.Packets;
using MQTTnet.Server;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Net;
using System.Text;
using System.Threading;
using System.Threading.Tasks;

namespace MqttnetServerWin
{
    /// <summary>
    /// MQTT通讯相关的工具类
    /// </summary>
    public class MQTTHelper
    {
        #region 变量
        /// <summary>
        /// 记录日志、输出、保存等操作
        /// </summary>
        private Action<ResultData_MQTT>? _Callback = null;
        #endregion 变量

        #region Server
        /// <summary>
        /// MQTT服务
        /// </summary>
        MqttServer _MqttServer = null;

        /// <summary>
        /// 创建MQTTServer并运行
        /// </summary>
        public async Task<ResultData_MQTT> CreateMQTTServerAndStart(MqttServerOptionsBuilder mqttServerOptionsBuilder, Action<ResultData_MQTT>? callback)
        {
            ResultData_MQTT resultData_MQTT = new ResultData_MQTT();

            _Callback = callback;
            try
            {
                MqttServerOptions mqttServerOptions = mqttServerOptionsBuilder.Build();
                _MqttServer = new MqttFactory().CreateMqttServer(mqttServerOptions);  // 创建服务（配置）

                _MqttServer.StartedAsync += StartedHandle;  // 服务器开启事件
                _MqttServer.StoppedAsync += StoppedHandle;  // 服务器关闭事件
                _MqttServer.ClientConnectedAsync += ClientConnectedHandle;        // 设置客户端连接成功后的处理程序
                _MqttServer.ClientDisconnectedAsync += ClientDisconnectedHandle;  // 设置客户端断开后的处理程序
                _MqttServer.ClientSubscribedTopicAsync += ClientSubscribedTopicHandle;      // 设置消息订阅通知
                _MqttServer.ClientUnsubscribedTopicAsync += ClientUnsubscribedTopicHandle;  // 设置消息退订通知
                _MqttServer.ApplicationMessageNotConsumedAsync += ApplicationMessageNotConsumedHandle;  // 设置消息处理程序

                await _MqttServer.StartAsync();  // 开启服务

                if (_MqttServer.IsStarted)
                {
                    resultData_MQTT = new ResultData_MQTT()
                    {
                        ResultCode = 1,
                        ResultMsg = DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss.fff") + ">>>执行了开启MQTTServer_成功！"
                    };
                }
                else
                {
                    resultData_MQTT = new ResultData_MQTT()
                    {
                        ResultCode = -1,
                        ResultMsg = DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss.fff") + ">>>执行了开启MQTTServer_失败！"
                    };
                }
            }
            catch (Exception ex)
            {
                resultData_MQTT = new ResultData_MQTT()
                {
                    ResultCode = -1,
                    ResultMsg = DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss.fff") + ">>>执行了开启MQTTServer_失败！错误信息：" + ex.Message
                };
            }

            _Callback?.Invoke(resultData_MQTT);
            return resultData_MQTT;
        }

        /// <summary>
        /// 简易创建MQTTServer并运行-不使用加密
        /// </summary>
        /// <param name="ip">IP</param>
        /// <param name="port">端口</param>
        /// <param name="withPersistentSessions">是否保持会话</param>
        /// <param name="callback">处理方法</param>
        /// <returns></returns>
        public async Task<ResultData_MQTT> CreateMQTTServerAndStart(string ip, int port, bool withPersistentSessions, Action<ResultData_MQTT>? callback)
        {
            ResultData_MQTT resultData_MQTT = new ResultData_MQTT();
            _Callback = callback;

            try
            {
                MqttServerOptionsBuilder mqttServerOptionsBuilder = new MqttServerOptionsBuilder();  // MQTT服务器配置
                mqttServerOptionsBuilder.WithDefaultEndpoint();
                mqttServerOptionsBuilder.WithDefaultEndpointBoundIPAddress(IPAddress.Parse(ip));  // 设置Server的IP
                mqttServerOptionsBuilder.WithDefaultEndpointPort(port);                           // 设置Server的端口号
                //mqttServerOptionsBuilder.WithEncryptedEndpointPort(port);                        // 使用加密的端点端口
                mqttServerOptionsBuilder.WithPersistentSessions(withPersistentSessions);  // 持续会话
                mqttServerOptionsBuilder.WithConnectionBacklog(2000);                     // 最大连接数
                //mqttServerOptionsBuilder.WithConnectionValidator(c =>  // 鉴权-方法失效
                //{
                //    if (c.Username != uName || c.Password != uPwd)
                //    {
                //        c.ReasonCode = MqttConnectReasonCode.BadUserNameOrPassword;
                //    }
                //})

                MqttServerOptions mqttServerOptions = mqttServerOptionsBuilder.Build();
                _MqttServer = new MqttFactory().CreateMqttServer(mqttServerOptions);  // 创建服务（配置）

                _MqttServer.StartedAsync += StartedHandle;  // 服务器开启事件
                _MqttServer.StoppedAsync += StoppedHandle;  // 服务器关闭事件
                _MqttServer.ClientConnectedAsync += ClientConnectedHandle;        // 设置客户端连接成功后的处理程序
                _MqttServer.ClientDisconnectedAsync += ClientDisconnectedHandle;  // 设置客户端断开后的处理程序
                _MqttServer.ClientSubscribedTopicAsync += ClientSubscribedTopicHandle;      // 设置消息订阅通知
                _MqttServer.ClientUnsubscribedTopicAsync += ClientUnsubscribedTopicHandle;  // 设置消息退订通知
                _MqttServer.ValidatingConnectionAsync += ValidatingConnectionHandle;                    // 鉴权-未完
                _MqttServer.ApplicationMessageNotConsumedAsync += ApplicationMessageNotConsumedHandle;  // 设置消息处理程序

                await _MqttServer.StartAsync();  // 开启服务

                if (_MqttServer.IsStarted)
                {
                    resultData_MQTT = new ResultData_MQTT()
                    {
                        ResultCode = 1,
                        ResultMsg = DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss.fff") + ">>>执行了开启MQTTServer_成功！"
                    };
                }
                else
                {
                    resultData_MQTT = new ResultData_MQTT()
                    {
                        ResultCode = -1,
                        ResultMsg = DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss.fff") + ">>>执行了开启MQTTServer_失败！"
                    };
                }
            }
            catch (Exception ex)
            {
                resultData_MQTT = new ResultData_MQTT()
                {
                    ResultCode = -1,
                    ResultMsg = DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss.fff") + ">>>执行了开启MQTTServer_失败！错误信息：" + ex.Message
                };
            }

            _Callback?.Invoke(resultData_MQTT);
            return resultData_MQTT;
        }

        /// <summary>
        /// 关闭MQTTServer
        /// </summary>
        public async Task<ResultData_MQTT> StopMQTTServer()
        {
            ResultData_MQTT resultData_MQTT = new ResultData_MQTT();

            try
            {
                if (_MqttServer == null)
                {
                    resultData_MQTT = new ResultData_MQTT()
                    {
                        ResultCode = -1,
                        ResultMsg = DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss.fff") + ">>>执行了关闭MQTTServer_出错！MQTTServer未在运行。"
                    };
                }
                else
                {
                    foreach (var clientStatus in _MqttServer.GetClientsAsync().Result)
                    {
                        await clientStatus.DisconnectAsync();
                    }
                    await _MqttServer.StopAsync();
                    _MqttServer = null;

                    resultData_MQTT = new ResultData_MQTT()
                    {
                        ResultCode = 1,
                        ResultMsg = DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss.fff") + ">>>执行了关闭MQTTServer_成功！"
                    };
                }
            }
            catch (Exception ex)
            {
                resultData_MQTT = new ResultData_MQTT()
                {
                    ResultCode = -1,
                    ResultMsg = DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss.fff") + ">>>执行了关闭MQTTServer_失败！错误信息：" + ex.Message
                };
            }

            _Callback?.Invoke(resultData_MQTT);
            return resultData_MQTT;
        }

        /// <summary>
        /// 获取所有的客户端
        /// </summary>
        public List<MqttClientStatus> GetClientsAsync()
        {
            return _MqttServer.GetClientsAsync().Result.ToList();
        }

        /// <summary>
        /// 发送消息-未写
        /// </summary>
        /// <param name="Topic">主题</param>
        /// <param name="msg">消息</param>
        /// <returns></returns>
        public Task SedMessage(string Topic, string msg)
        {
            try
            {
                //var clients = _MqttServer.GetClientsAsync().Result;

                //foreach (var client in clients)
                //{

                //}
            }
            catch { }
            return Task.CompletedTask;
        }
        #region 处理事件
        /// <summary>
        /// 开启Server的处理程序
        /// </summary>
        private Task StartedHandle(EventArgs arg)
        {
            _Callback?.Invoke(new()
            {
                ResultCode = 1,
                ResultMsg = DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss.fff") + ">>>MQTTServer已开启！"
            });
            return Task.CompletedTask;
        }

        /// <summary>
        /// 关闭Server的处理程序
        /// </summary>
        private Task StoppedHandle(EventArgs arg)
        {
            _Callback?.Invoke(new()
            {
                ResultCode = 1,
                ResultMsg = DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss.fff") + ">>>MQTTServer已关闭！"
            });
            return Task.CompletedTask;
        }

        /// <summary>
        /// 设置客户端连接成功后的处理程序
        /// </summary>
        private Task ClientConnectedHandle(ClientConnectedEventArgs arg)
        {
            var clients = _MqttServer.GetClientsAsync().Result;

            _Callback?.Invoke(new()
            {
                ResultCode = 1,
                ResultMsg = DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss.fff") + $">>>客户端'{arg.ClientId}'已成功连接！当前客户端连接数：{clients?.Count}个。"
            });
            return Task.CompletedTask;
        }

        /// <summary>
        /// 设置客户端断开后的处理程序
        /// </summary>
        private Task ClientDisconnectedHandle(ClientDisconnectedEventArgs arg)
        {
            var clients = _MqttServer.GetClientsAsync().Result;
            _Callback?.Invoke(new ResultData_MQTT()
            {
                ResultCode = 1,
                ResultMsg = DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss.fff") + $">>>客户端'{arg.ClientId}'已断开连接！当前客户端连接数：{clients?.Count}个。"
            });

            return Task.CompletedTask;
        }

        /// <summary>
        /// 设置消息订阅通知
        /// </summary>
        private Task ClientSubscribedTopicHandle(ClientSubscribedTopicEventArgs arg)
        {
            //if (!arg.Equals("admin"))
            //{
            //    var client = clients.Where(a => a.Id == arg.ClientId).FirstOrDefault();
            //    client?.DisconnectAsync();

            //    return Task.CompletedTask;
            //}

            _Callback?.Invoke(new ResultData_MQTT()
            {
                ResultCode = 1,
                ResultMsg = DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss.fff") + $">>>客户端'{arg.ClientId}'订阅了主题'{arg.TopicFilter.Topic}'，主题服务质量：'{arg.TopicFilter.QualityOfServiceLevel}'！"
            });

            return Task.CompletedTask;
        }

        /// <summary>
        /// 设置消息退订通知
        /// </summary>
        private Task ClientUnsubscribedTopicHandle(ClientUnsubscribedTopicEventArgs arg)
        {
            _Callback?.Invoke(new ResultData_MQTT()
            {
                ResultCode = 1,
                ResultMsg = DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss.fff") + $">>>客户端{arg.ClientId}退订了主题{arg.TopicFilter}！"
            });

            return Task.CompletedTask;
        }

        /// <summary>
        /// 鉴权-未写完
        /// </summary>
        /// <returns></returns>
        private Task ValidatingConnectionHandle(ValidatingConnectionEventArgs arg)  // 鉴权
        {
            if (arg.UserName != "Admin" || arg.Password != "Admin123")
            {

            }
            return Task.CompletedTask;
        }

        /// <summary>
        /// 设置消息处理程序
        /// </summary>
        private Task ApplicationMessageNotConsumedHandle(ApplicationMessageNotConsumedEventArgs arg)
        {
            _Callback?.Invoke(new ResultData_MQTT()
            {
                ResultCode = -1,
                ResultMsg = DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss.fff") + $">>>客户端：'{arg.SenderId}'发布了消息：主题：'{arg.ApplicationMessage.Topic}'！内容：'{Encoding.UTF8.GetString(arg.ApplicationMessage.Payload)}'；服务质量：{arg.ApplicationMessage.QualityOfServiceLevel}；保留：{arg.ApplicationMessage.Retain}"
            });

            return Task.CompletedTask;
        }
        #endregion 处理事件
        #endregion Server

        #region Client
        /// <summary>
        /// 客户端
        /// </summary>
        IMqttClient _MqttClient = null;

        /// <summary>
        /// 创建MQTTClient并运行
        /// </summary>
        /// <param name="mqttClientOptionsBuilder">MQTTClient连接配置</param>
        /// <param name="callback">信息处理逻辑</param>
        /// <returns></returns>
        public async Task<ResultData_MQTT> CreateMQTTClientAndStart(MqttClientOptionsBuilder mqttClientOptionsBuilder, Action<ResultData_MQTT>? callback)
        {
            ResultData_MQTT resultData_MQTT = new ResultData_MQTT();

            _Callback = callback;
            try
            {
                MqttClientOptions options = mqttClientOptionsBuilder.Build();

                _MqttClient = new MqttFactory().CreateMqttClient();
                _MqttClient.ConnectedAsync += ConnectedHandle;        // 服务器连接事件
                _MqttClient.DisconnectedAsync += DisconnectedHandle;  // 服务器断开事件（可以写入重连事件）
                _MqttClient.ApplicationMessageReceivedAsync += ApplicationMessageReceivedHandle;  // 发送消息事件
                await _MqttClient.ConnectAsync(options);  // 连接

                if (_MqttClient.IsConnected)
                {
                    resultData_MQTT = new ResultData_MQTT()
                    {
                        ResultCode = 1,
                        ResultMsg = DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss.fff") + ">>>执行了开启MQTTClient_成功！"
                    };
                }
                else
                {
                    resultData_MQTT = new ResultData_MQTT()
                    {
                        ResultCode = -1,
                        ResultMsg = DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss.fff") + ">>>执行了开启MQTTClient_失败！"
                    };
                }
            }
            catch (Exception ex)
            {
                resultData_MQTT = new ResultData_MQTT()
                {
                    ResultCode = -1,
                    ResultMsg = DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss.fff") + ">>>执行了开启MQTTClient_失败！错误信息：" + ex.Message
                };
            }

            _Callback?.Invoke(resultData_MQTT);
            return resultData_MQTT;
        }

        /// <summary>
        /// 简易创建MQTTClient并运行
        /// </summary>
        /// <param name="mqttServerUrl">mqttServer的Url</param>
        /// <param name="port">mqttServer的端口</param>
        /// <param name="userName">认证用用户名</param>
        /// <param name="userPassword">认证用密码</param>
        /// <param name="callback">信息处理逻辑</param>
        /// <returns></returns>
        public async Task<ResultData_MQTT> CreateMQTTClientAndStart(string mqttServerUrl, int port, string userName, string userPassword, Action<ResultData_MQTT>? callback)
        {
            ResultData_MQTT resultData_MQTT = new ResultData_MQTT();

            _Callback = callback;
            try
            {
                MqttClientOptionsBuilder mqttClientOptionsBuilder = new MqttClientOptionsBuilder();
                mqttClientOptionsBuilder.WithTcpServer(mqttServerUrl, port);          // 设置MQTT服务器地址
                if (!string.IsNullOrEmpty(userName))
                {
                    mqttClientOptionsBuilder.WithCredentials(userName, userPassword);  // 设置鉴权参数
                }
                mqttClientOptionsBuilder.WithClientId(Guid.NewGuid().ToString("N"));  // 设置客户端序列号
                MqttClientOptions options = mqttClientOptionsBuilder.Build();

                _MqttClient = new MqttFactory().CreateMqttClient();
                _MqttClient.ConnectedAsync += ConnectedHandle;        // 服务器连接事件
                _MqttClient.DisconnectedAsync += DisconnectedHandle;  // 服务器断开事件（可以写入重连事件）
                _MqttClient.ApplicationMessageReceivedAsync += ApplicationMessageReceivedHandle;  // 发送消息事件
                await _MqttClient.ConnectAsync(options);  // 连接

                if (_MqttClient.IsConnected)
                {
                    resultData_MQTT = new ResultData_MQTT()
                    {
                        ResultCode = 1,
                        ResultMsg = DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss.fff") + ">>>执行了开启MQTTClient_成功！"
                    };
                }
                else
                {
                    resultData_MQTT = new ResultData_MQTT()
                    {
                        ResultCode = -1,
                        ResultMsg = DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss.fff") + ">>>执行了开启MQTTClient_失败！"
                    };
                }
            }
            catch (Exception ex)
            {
                resultData_MQTT = new ResultData_MQTT()
                {
                    ResultCode = -1,
                    ResultMsg = DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss.fff") + ">>>执行了开启MQTTClient_失败！错误信息：" + ex.Message
                };
            }
            _Callback?.Invoke(resultData_MQTT);
            return resultData_MQTT;
        }

        /// <summary>
        /// 关闭MQTTClient
        /// </summary>
        public async Task DisconnectAsync_Client()
        {
            ResultData_MQTT resultData_MQTT = new ResultData_MQTT();
            try
            {
                if (_MqttClient != null && _MqttClient.IsConnected)
                {
                    await _MqttClient.DisconnectAsync();
                    _MqttClient.Dispose();
                    _MqttClient = null;

                    resultData_MQTT = new ResultData_MQTT()
                    {
                        ResultCode = 1,
                        ResultMsg = DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss.fff") + ">>>执行了关闭MQTTClient_成功！"
                    };
                }
                else
                {
                    resultData_MQTT = new ResultData_MQTT()
                    {
                        ResultCode = -1,
                        ResultMsg = DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss.fff") + ">>>执行了关闭MQTTClient_失败！MQTTClient未开启连接！"
                    };
                }
            }
            catch (Exception ex)
            {
                resultData_MQTT = new ResultData_MQTT()
                {
                    ResultCode = -1,
                    ResultMsg = DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss.fff") + ">>>执行了关闭MQTTClient_失败！错误信息：" + ex.Message
                };
            }
            _Callback?.Invoke(resultData_MQTT);
        }

        /// <summary>
        /// 重连
        /// </summary>
        /// <returns></returns>
        public async Task ReconnectAsync_Client()
        {
            ResultData_MQTT resultData_MQTT = new ResultData_MQTT();
            try
            {
                if (_MqttClient != null)
                {
                    await _MqttClient.ReconnectAsync();
                    resultData_MQTT = new ResultData_MQTT()
                    {
                        ResultCode = 1,
                        ResultMsg = DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss.fff") + ">>>执行了MQTTClient重连_成功！"
                    };
                }
                else
                {
                    resultData_MQTT = new ResultData_MQTT()
                    {
                        ResultCode = -1,
                        ResultMsg = DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss.fff") + ">>>执行了MQTTClient重连_失败！未设置MQTTClient连接！"
                    };
                }
            }
            catch (Exception ex)
            {
                resultData_MQTT = new ResultData_MQTT()
                {
                    ResultCode = -1,
                    ResultMsg = DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss.fff") + ">>>执行了MQTTClient重连_失败！错误信息：" + ex.Message
                };
            }
            _Callback?.Invoke(resultData_MQTT);
        }

        /// <summary>
        /// 订阅
        /// </summary>
        /// <param name="topic">主题</param>
        public async void SubscribeAsync_Client(string topic)
        {
            ResultData_MQTT resultData_MQTT = new ResultData_MQTT();
            try
            {
                MqttTopicFilter topicFilter = new MqttTopicFilterBuilder().WithTopic(topic).Build();
                await _MqttClient.SubscribeAsync(topicFilter, CancellationToken.None);

                resultData_MQTT = new ResultData_MQTT()
                {
                    ResultCode = 1,
                    ResultMsg = DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss.fff") + $">>>MQTTClient执行了订阅'{topic}'_成功！"
                };
            }
            catch (Exception ex)
            {
                resultData_MQTT = new ResultData_MQTT()
                {
                    ResultCode = -1,
                    ResultMsg = DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss.fff") + $">>>MQTTClient执行了订阅'{topic}'_失败！错误信息：" + ex.Message
                };
            }
            _Callback?.Invoke(resultData_MQTT);
        }
        /// <summary>
        /// 退订阅
        /// </summary>
        /// <param name="topic">主题</param>
        public async void UnsubscribeAsync_Client(string topic)
        {
            ResultData_MQTT resultData_MQTT = new ResultData_MQTT();
            try
            {
                await MqttClientExtensions.UnsubscribeAsync(_MqttClient, topic, CancellationToken.None);
                resultData_MQTT = new ResultData_MQTT()
                {
                    ResultCode = 1,
                    ResultMsg = DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss.fff") + $">>>MQTTClient执行了退订'{topic}'_成功！"
                };
            }
            catch (Exception ex)
            {
                resultData_MQTT = new ResultData_MQTT()
                {
                    ResultCode = -1,
                    ResultMsg = DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss.fff") + $">>>MQTTClient执行退订'{topic}'_失败！错误信息：" + ex.Message
                };
            }
            _Callback?.Invoke(resultData_MQTT);
        }

        /// <summary>
        /// 发布消息( 必须在成功连接以后才生效 )
        /// </summary>
        /// <param name="topic">主题</param>
        /// <param name="msg">信息</param>
        /// <param name="retained">是否保留</param>
        /// <returns></returns>
        public async Task PublishAsync_Client(string topic, string msg, bool retained)
        {
            ResultData_MQTT resultData_MQTT = new ResultData_MQTT();

            try
            {
                MqttApplicationMessageBuilder mqttApplicationMessageBuilder = new MqttApplicationMessageBuilder();
                mqttApplicationMessageBuilder.WithTopic(topic);          // 主题
                mqttApplicationMessageBuilder.WithPayload(msg);          // 信息
                mqttApplicationMessageBuilder.WithRetainFlag(retained);  // 保留

                MqttApplicationMessage messageObj = mqttApplicationMessageBuilder.Build();

                if (_MqttClient.IsConnected)
                {
                    await _MqttClient.PublishAsync(messageObj, CancellationToken.None);

                    resultData_MQTT = new ResultData_MQTT()
                    {
                        ResultCode = 1,
                        ResultMsg = DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss.fff") + $">>>执行了发布信息_成功！主题:'{topic}'，信息:'{msg}'，是否保留:'{retained}'"
                    };
                }
                else
                {
                    // 未连接
                    resultData_MQTT = new ResultData_MQTT()
                    {
                        ResultCode = -1,
                        ResultMsg = DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss.fff") + ">>>执行了发布信息_失败！MQTTClient未开启连接！"
                    };
                }
            }
            catch (Exception ex)
            {
                resultData_MQTT = new ResultData_MQTT()
                {
                    ResultCode = -1,
                    ResultMsg = DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss.fff") + ">>>执行了发布信息_失败！错误信息：" + ex.Message
                };
            }

            _Callback?.Invoke(resultData_MQTT);
        }

        #region 事件
        /// <summary>
        /// 服务器连接事件
        /// </summary>
        private Task ConnectedHandle(MqttClientConnectedEventArgs arg)
        {
            _Callback?.Invoke(new ResultData_MQTT()
            {
                ResultCode = 1,
                ResultMsg = DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss.fff") + ">>>已连接到MQTT服务器！"
            });
            return Task.CompletedTask;
        }

        /// <summary>
        /// 服务器断开事件（可以写入重连事件）
        /// </summary>
        private Task DisconnectedHandle(MqttClientDisconnectedEventArgs arg)
        {
            _Callback?.Invoke(new ResultData_MQTT()
            {
                ResultCode = 1,
                ResultMsg = DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss.fff") + $">>>已断开与MQTT服务器连接！"
            });
            return Task.CompletedTask;
        }

        /// <summary>
        /// 发送消息事件
        /// </summary>
        private Task ApplicationMessageReceivedHandle(MqttApplicationMessageReceivedEventArgs arg)
        {
            _Callback?.Invoke(new ResultData_MQTT()
            {
                ResultCode = 1,
                ResultMsg = DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss.fff") + $">>>MQTTClient'{arg.ClientId}'内容：'{Encoding.UTF8.GetString(arg.ApplicationMessage.Payload)}'；主题：'{arg.ApplicationMessage.Topic}'，消息等级Qos：[{arg.ApplicationMessage.QualityOfServiceLevel}]，是否保留：[{arg.ApplicationMessage.Retain}]",
                ResultObject1 = arg.ApplicationMessage.Topic,
                ResultObject2 = Encoding.UTF8.GetString(arg.ApplicationMessage.Payload)
            });
            return Task.CompletedTask;
        }
        #endregion 事件
        #endregion Client
    }

    /// <summary>
    /// 信息载体
    /// </summary>
    public class ResultData_MQTT
    {
        /// <summary>
        /// 结果Code
        /// 正常1，其他为异常；0不作为回复结果
        /// </summary>
        public int ResultCode { get; set; } = 0;

        /// <summary>
        /// 结果信息
        /// </summary>
        public string ResultMsg { get; set; } = string.Empty;

        /// <summary>
        /// 扩展1
        /// </summary>
        public object? ResultObject1 { get; set; } = string.Empty;

        /// <summary>
        /// 扩展2
        /// </summary>
        public object? ResultObject2 { get; set; } = string.Empty;
    }
}

四、分布式MQTT消息服务器推荐

  EMQ X（简称 EMQ）：是一款完全开源，高度可伸缩，高可用的分布式 MQTT 消息服务器，同时也支持 CoAP/LwM2M 一站式 IoT 协议接入。可处理千万级别的并发客户端。

 

参考：

　　https://www.cnblogs.com/hayasi/p/7743356.html

　　https://www.cnblogs.com/dathlin/p/11631894.html

　　https://www.cnblogs.com/sxkgeek/p/9140180.html

　　https://www.shangmayuan.com/a/67b2f4a9f2c440e9b2db3a97.html

　　https://zhuanlan.zhihu.com/p/419561816

　　https://blog.csdn.net/qq_37258787/article/details/80183923

　　https://www.cnblogs.com/dathlin/p/11631894.html

　　https://www.jianshu.com/p/a371c6ac076b

今天的文章MQTT学习笔记（C#）-MQTTnet分享到此就结束了，感谢您的阅读，如果确实帮到您，您可以动动手指转发给其他人。

版权声明：本文内容由互联网用户自发贡献，该文观点仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容， 请发送邮件至 举报，一经查实，本站将立刻删除。
如需转载请保留出处：https://bianchenghao.cn/49565.html

赞 (0)
编程小号编程小号
0
上一篇 2023-09-02
下一篇 2023-09-02

相关推荐

发表回复

您的电子邮箱地址不会被公开。 必填项已用*标注