C# 基于MQTT创建客户端的可靠数据传输

引言

MQTT是tcpip的应用层协议，这里我们简单介绍一下MQTT的基本概念，并用C# 描述客户端的订阅和发布。

MQTT简介

• MQTT(Message Queuing Telemetry Transport）即 消息队列遥测传输，是ISO 标准(ISO/IEC PRF 20922)下基于发布/订阅范式的消息协议。它也是工作在 TCP/IP协议族上，是为硬件性能低下的远程设备以及网络状况糟糕的情况下而设计的发布/订阅型消息协议。

• MQTT协议是轻量、简单、开放和易于实现的，这些特点使它适用范围非常广泛。在很多情况下，包括受限的环境中，如：机器与机器（M2M）通信和物联网（IoT）。其在，通过卫星链路通信传感器、偶尔拨号的医疗设备、智能家居、及一些小型化设备中已广泛使用。

• 通过MQTT协议，目前已经扩展出了数十个MQTT服务器端程序，可以通过PHP，JAVA，Python，C，C#等系统语言来向MQTT发送相关消息。

• MQTT 服务器是发布-订阅架构的转发中心，它可以非常简单地在Internet 服务器上实现。服务器分发消息，因此是推送者，客户端可以发布消息（发送方）、订阅消息（接收方）或两者兼而有之。客户端（也称为节点）是一种智能设备，如微控制器或具有 TCP/IP 堆栈和实现 MQTT 协议的软件的计算机。
  

• QoS（Quality of Service levels）服务质量是 MQTT 的一个重要特性。当我们使用时，连接已经在一定程度上是不是面向连接的。MQTT 在这里帮助避免信息丢失及其服务质量水平。 服务质量水平包括三个等级。

  1. AtMostOnce ——最多一次
  2. AtLeastOnce ——至少一次
  3. ExactlyOnce ——恰好一次，这个服务质量可以确保数据准确的发送到订阅端。

C# MQTT库引用

这里我使用是VS2022，在工具中的netGet包管理器中打开“管理解决方案包“，搜索MQTTnet，找到对应的包。这里注意要选择自己合适的版本！注意要选择自己合适的版本！注意要选择自己合适的版本！重要的话说3遍。我在win10环境下使用的VS2022,验证了多次才发现4.1.2.350是合适的。如下图所示。

代码和描述

1、 代码

using System;
using System.Collections.Generic; 
using System.Threading.Tasks; 
using MQTTnet;
using MQTTnet.Client; 
using MQTTnet.Protocol;
using MQTTnet.Server;

namespace WriteSN
{
  internal class Mqtt
  {
      public static byte[] Mqtt_Message_Received_Str = new byte[] { };
  	//构造函数
      public Mqtt()
      {
         //初始化全局变量
          Mqtt_Message_Received_Str = new byte[] { };
      }

      /// <summary>
      ///  发布和订阅MQTT专用主题授权获取数据
      /// </summary>
      /// <param name="Data">发布传输字符串</param>
      /// <returns></returns>
      public async Task MQTT_Client_Handle(string Data)
      {
  		//服务端地址
  		string broker = "broker.emqx.io";
  		//端口·
  		int port = 8090;
  		//客户端ID
  		string clientId = Guid.NewGuid().ToString();
  		//订阅和发布的主题
  		string  req_topic = "trace_id_req";//发
  		string  rsp_topic = "trace_id_rsp";//收 
  		//用户和密码
  		string username = "";//"admin";
  		string password = "";//public";
  		
  		// Create a MQTT client factory 创建客户端代理
  		var factory = new MqttFactory();
  		
  		// Create a MQTT client instance 创建客户端实例
  		var mqttClient = factory.CreateMqttClient();
  		
          // Create MQTT client options 为连接服务器创建客户端选项
          var options = new MqttClientOptionsBuilder().
              .WithTcpServer(broker, port) // MQTT broker address and port
              .WithCredentials(username, password) // Set username and password
              .WithClientId(clientId).WithWillQualityOfServiceLevel(MqttQualityOfServiceLevel.ExactlyOnce)//QOS=2
              .WithCleanSession()
              .Build();
          // Connect to MQTT broker //连接代理服务器
          var connectResult = await mqttClient.ConnectAsync(options);
      
          if (connectResult.ResultCode == MqttClientConnectResultCode.Success)
          {   
              //连接成功后打印
              Console.WriteLine("\r\nConnected to MQTT broker successfully.\r\n");

              // Subscribe to a topic //订阅一个要接收的主题
              await mqttClient.SubscribeAsync(rsp_topic, MqttQualityOfServiceLevel.ExactlyOnce);//QOS=2
              // Callback function when a message is received //添加接收回调的消息
              mqttClient.ApplicationMessageReceivedAsync += MqttClient_ApplicationMessageReceivedAsync;

              //=======================================================================================       
              //开始发布消息        
              var message = new MqttApplicationMessageBuilder()
                  .WithTopic(req_topic)
                  .WithPayload(Data)
                  .WithQualityOfServiceLevel(MqttQualityOfServiceLevel.ExactlyOnce)//QOS=2
                  .WithRetainFlag(false)
                  .Build();
              Console.WriteLine($"Send topic:{message.Topic}\r\nSend QoS:{message.QualityOfServiceLevel}\r\nSend RetainFlag: {message.Retain}\r\n");
              //发送一个主题消息
              await mqttClient.PublishAsync(message);
              // Wait for 1 second //自定义等待一段时间后关闭
              await Task.Delay(1000); 
              // Unsubscribe and disconnect//关闭接收主题
              await mqttClient.UnsubscribeAsync(rsp_topic);
              await mqttClient.DisconnectAsync();
              Console.WriteLine($"Unsubscribe and disconnect.\r\n");
          }
          else
          {
              Console.WriteLine($"Failed to connect to MQTT broker: {connectResult.ResultCode}");
          }
      }

     //回调函数，打印和传递订阅接收到数据
      private static Task MqttClient_ApplicationMessageReceivedAsync(MqttApplicationMessageReceivedEventArgs arg)
      {
          Mqtt_Message_Received_Str = arg.ApplicationMessage.Payload;

          Console.WriteLine($"Received message:\r\n{Utils.bytesToString(Mqtt_Message_Received_Str)}\r\n[{arg.ApplicationMessage.Payload.Length}]\r\n");
          Console.WriteLine("Received topic:" + $"{arg.ApplicationMessage.Topic}\t\nReceived QoS:{arg.ApplicationMessage.QualityOfServiceLevel}\r\nReceived RetainFlag: {arg.ApplicationMessage.Retain}\r\n");
          return Task.CompletedTask;
      }

  }
}

2、 描述

• 小结以上代码可以理解为3部分功能，创建连接、回调接收和发布消息。且3部分中均可以设置服务质量QoS（Quality of Service levels），确保可靠性传输，当然，服务器端也必须对标QOS设置：
  1、创建连接

  var connectResult = await mqttClient.ConnectAsync(options);

  2、回调接收

  mqttClient.ApplicationMessageReceivedAsync += MqttClient_ApplicationMessageReceivedAsync

  3、发布消息

  await mqttClient.PublishAsync(message);

  4、Qos设置
  参看代码备注 “//QOS=2”