.NET Core集成MQTTnet实现MQTT服务端

一，什么是MQTT

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

MQTT是一个基于客户端-服务器的消息发布/订阅传输协议。MQTT协议是轻量、简单、开放和易于实现的，这些特点使它适用范围非常广泛。在很多情况下，包括受限的环境中，如：机器与机器（M2M）通信和物联网（IoT）。其在，通过卫星链路通信传感器、偶尔拨号的医疗设备、智能家居、及一些小型化设备中已广泛使用。——引自百度百科

二，常见的MQTT软件

MQTT官网上有很多推荐，博友们可以慢慢看，附上一篇别人的开源 MQTT Broker 对比文章https://wivwiv.com/post/best-mqtt-broker

三，MQTT通讯类型

名字	值	流向	描述
CONNECT	1	C->S	客户端请求与服务端建立连接
CONNACK	2	S->C	服务端确认连接建立
PUBLISH	3	CóS	发布消息
PUBACK	4	CóS	收到发布消息确认
PUBREC	5	CóS	发布消息收到
PUBREL	6	CóS	发布消息释放
PUBCOMP	7	CóS	发布消息完成
SUBSCRIBE	8	C->S	订阅请求
SUBACK	9	S->C	订阅确认
UNSUBSCRIBE	10	C->S	取消订阅
UNSUBACK	11	S->C	取消订阅确认
PING	12	C->S	客户端发送PING(连接保活)命令
PINGRSP	13	S->C	PING命令回复
DISCONNECT	14	C->S	断开连接

四，MQTT QoS等级

QoS(Quality of Service)，服务质量。MQTT中有三种Qos：

QoS0，At most once，至多一次；

QoS1，At least once，至少一次；

QoS2，Exactly once，确保只有一次。

QoS 是消息的发送方（Sender）和接受方（Receiver）之间达成的一个协议：

l QoS0 代表，Sender 发送的一条消息，Receiver 最多能收到一次，也就是说 Sender 尽力向 Receiver 发送消息，如果发送失败，也就算了；

Sender 向 Receiver 发送一个包含消息数据的 PUBLISH 包，然后不管结果如何，丢弃掉已发送的 PUBLISH 包，一条消息的发送完成。

l QoS1 代表，Sender 发送的一条消息，Receiver 至少能收到一次，也就是说 Sender 向 Receiver 发送消息，如果发送失败，会继续重试，直到 Receiver 收到消息为止，

QoS 要保证消息至少到达 Sender 一次，所以有一个应答的机制。

Sender 向 Receiver 发送一个带有消息数据的 PUBLISH 包，并在本地保存这个 PUBLISH 包。
Receiver 收到 PUBLISH 包以后，向 Sender 发送一个 PUBACK 数据包，PUBACK 数据包没有消息体（Payload），在可变头中（Variable header）中有一个包标识（Packet Identifier），和它收到的 PUBLISH 包中的 Packet Identifier 一致。
Sender 收到 PUBACK 之后，根据 PUBACK 包中的 Packet Identifier 找到本地保存的 PUBLISH 包，然后丢弃掉，一次消息的发送完成。
如果 Sender 在一段时间内没有收到 PUBLISH 包对应的 PUBACK，它将该 PUBLISH 包的 DUP 标识设为 1（代表是重新发送的 PUBLISH 包），然后重新发送该 PUBLISH 包。重复这个流程，直到收到 PUBACK，然后执行第 3 步。

l QoS2 代表，Sender 发送的一条消息，Receiver 确保能收到而且只收到一次，也就是说 Sender 尽力向 Receiver 发送消息，如果发送失败，会继续重试，直到 Receiver 收到消息为止，同时保证 Receiver 不会因为消息重传而收到重复的消息。

Sender 发送 QoS 为 2 的 PUBLISH 数据包，数据包 Packet Identifier 为 P，并在本地保存该 PUBLISH 包；
Receiver 收到 PUBLISH 数据包以后，在本地保存 PUBLISH 包的 Packet Identifier P，并回复 Sender 一个 PUBREC 数据包，PUBREC 数据包可变头中的 Packet Identifier 为 P，没有消息体（Payload）；
当 Sender 收到 PUBREC，它就可以安全地丢弃掉初始的 Packet Identifier 为 P 的 PUBLISH 数据包，同时保存该 PUBREC 数据包，同时回复 Receiver 一个 PUBREL 数据包，PUBREL 数据包可变头中的 Packet Identifier 为 P，没有消息体；如果 Sender 在一定时间内没有收到 PUBREC，它会把 PUBLISH 包的 DUP 标识设为 1，重新发送该 PUBLISH 数据包（Payload）；
当 Receiver 收到 PUBREL 数据包，它可以丢弃掉保存的 PUBLISH 包的 Packet Identifier P，并回复 Sender 一个 PUBCOMP 数据包，PUBCOMP 数据包可变头中的 Packet Identifier 为 P，没有消息体（Payload）；
当 Sender 收到 PUBCOMP 包，那么它认为数据包传输已完成，它会丢弃掉对应的 PUBREC 包。如果 Sender 在一定时间内没有收到 PUBCOMP 包，它会重新发送 PUBREL 数据包。

注：QoS 是 Sender 和 Receiver 之间达成的协议，不是 Publisher 和 Subscriber 之间达成的协议。也就是说 Publisher 发布一条 QoS1 的消息，只能保证 Broker 能至少收到一次这个消息；至于对应的 Subscriber 能否至少收到一次这个消息，还要取决于 Subscriber 在 Subscribe 的时候和 Broker 协商的 QoS 等级。

五，QoS 和会话（Session）

如果 Client 想接收离线消息，必须使用持久化的会话（Clean Session = 0）连接到 Broker，这样 Broker 才会存储 Client 在离线期间没有确认接收的 QoS 大于 1 的消息。

六，QoS降级问题

在 MQTT 协议中，从 Broker 到 Subscriber 这段消息传递的实际 QoS 等于：Publisher 发布消息时指定的 QoS 等级和 Subscriber 在订阅时与 Broker 协商的 QoS 等级，这两个 QoS 等级中的最小那一个。

七，QoS选择

在以下情况下你可以选择 QoS0：

Client 和 Broker 之间的网络连接非常稳定，例如一个通过有线网络连接到 Broker 的测试用 Client；
可以接受丢失部分消息，比如你有一个传感器以非常短的间隔发布状态数据，所以丢一些也可以接受；
不需要离线消息。

在以下情况下你应该选择 QoS1：

你需要接收所有的消息，而且你的应用可以接受并处理重复的消息；
你无法接受 QoS2 带来的额外开销，QoS1 发送消息的速度比 QoS2 快很多。

在以下情况下你应该选择 QoS2：

你的应用必须接收到所有的消息，而且你的应用在重复的消息下无法正常工作，同时你也能接受 QoS2 带来的额外开销。

八，MQTT的retain标志位

当我们使用MQTT客户端发布消息（PUBLISH）时，如果将RETAIN标志位设置为true，那么MQTT服务器会将最近收到的一条RETAIN标志位为true的消息保存在服务器端（内存或文件）。

特别注意：MQTT服务器只会为每一个Topic保存最近收到的一条RETAIN标志位为true的消息！也就是说，如果MQTT服务器上已经为某个Topic保存了一条Retained消息，当客户端再次发布一条新的Retained消息，那么服务器上原来的那条消息会被覆盖！

每当MQTT客户端连接到MQTT服务器并订阅了某个topic，如果该topic下有Retained消息，那么MQTT服务器会立即向客户端推送该条Retained消息。

九，MQTT的will（遗愿消息）

想一下以下场景，你的设备向服务端发送了在线的消息后突然爆炸了，它还没来得及和服务端说它爆炸了就死了，这样会勿让我们以为它还在线，但其实它已经挂了。有没有方法让客户端非正常断线后通知服务端呢？有的，就是使用遗愿消息，

在建立与服务端的连接时约定好遗愿消息，服务端会存储这个消息，当客户端非正常断线时则会向约定好的主题发送遗愿消息，同样，它也可以设置为retian。

十，基于MQTTnet实现MQTT服务端

新建一个控制台程序

安装nuget包MQTTnet，.NET Core下安装最新版本即可

修改Program.cs，编写实现代码

  1 using MQTTnet;
  2 using MQTTnet.Client.Receiving;
  3 using MQTTnet.Protocol;
  4 using MQTTnet.Server;
  5 using System;
  6 using System.Collections.Generic;
  7 using System.Text;
  8 using System.Threading.Tasks;
  9 
 10 namespace MQTT
 11 {
 12     class Program
 13     {
 14         static void Main(string[] args)
 15         {
 16             MqttServerClass serverClass = new MqttServerClass();
 17             serverClass.StartMqttServer().Wait();
 18             Console.ReadLine();
 19         }
 20     }
 21     public static class Config
 22     {
 23         public static int Port { get; set; } = 1883;
 24         public static string UserName { get; set; } = "Username";
 25         public static string Password { get; set; } = "Password";
 26 
 27     }
 28     public class UserInstance
 29     {
 30         public string ClientId { get; set; }
 31         public string UserName { get; set; }
 32         public string Password { get; set; }
 33     }
 34     public class MqttServerClass
 35     {
 36         private IMqttServer mqttServer;
 37         private List<MqttApplicationMessage> messages = new List<MqttApplicationMessage>();
 38 
 39         public async Task StartMqttServer()
 40         {
 41             try
 42             {
 43                 if (mqttServer == null)
 44                 {
 45                     var optionsBuilder = new MqttServerOptionsBuilder()
 46                     .WithDefaultEndpoint()
 47                     .WithDefaultEndpointPort(Config.Port)
 48                     //连接拦截器
 49                     .WithConnectionValidator(
 50                         c =>
 51                         {
 52                             var flag = c.Username == Config.UserName && c.Password == Config.Password;
 53                             if (!flag)
 54                             {
 55                                 c.ReasonCode = MqttConnectReasonCode.BadUserNameOrPassword;
 56                                 return;
 57                             }
 58                             //设置代码为 Success
 59                             c.ReasonCode = MqttConnectReasonCode.Success;
 60                             //instances.Add(new UserInstance()  //缓存到内存的List集合当中
 61                             //{
 62                             //    ClientId = c.ClientId,
 63                             //    UserName = c.Username,
 64                             //    Password = c.Password
 65                             //});
 66                         })
 67                     //订阅拦截器
 68                     .WithSubscriptionInterceptor(
 69                         c =>
 70                         {
 71                             if (c == null) return;
 72                             c.AcceptSubscription = true;
 73                         })
 74                     //应用程序消息拦截器
 75                     .WithApplicationMessageInterceptor(
 76                         c =>
 77                         {
 78                             if (c == null) return;
 79                             c.AcceptPublish = true;
 80                         })
 81                     //clean sesison是否生效
 82                    .WithPersistentSessions();
 83 
 84                     mqttServer = new MqttFactory().CreateMqttServer();
 85 
 86                     //客户端断开连接拦截器
 87                     //mqttServer.UseClientDisconnectedHandler(c =>
 88                     //{
 89                     //    //var user = instances.FirstOrDefault(t => t.ClientId == c.ClientId);
 90                     //    //if (user != null)
 91                     //    //{
 92                     //    //    instances.Remove(user);
 93                     //    //}
 94                     //});
 95 
 96                     //服务开始
 97                     mqttServer.StartedHandler = new MqttServerStartedHandlerDelegate(OnMqttServerStarted);
 98                     //服务停止
 99                     mqttServer.StoppedHandler = new MqttServerStoppedHandlerDelegate(OnMqttServerStopped);
100                     //客户端连接
101                     mqttServer.ClientConnectedHandler = new MqttServerClientConnectedHandlerDelegate(OnMqttServerClientConnected);
102                     //客户端断开连接（此事件会覆盖拦截器）
103                     mqttServer.ClientDisconnectedHandler = new MqttServerClientDisconnectedHandlerDelegate(OnMqttServerClientDisconnected);
104                     //客户端订阅
105                     mqttServer.ClientSubscribedTopicHandler = new MqttServerClientSubscribedHandlerDelegate(OnMqttServerClientSubscribedTopic);
106                     //客户端取消订阅
107                     mqttServer.ClientUnsubscribedTopicHandler = new MqttServerClientUnsubscribedTopicHandlerDelegate(OnMqttServerClientUnsubscribedTopic);
108                     //服务端收到消息
109                     mqttServer.ApplicationMessageReceivedHandler = new MqttApplicationMessageReceivedHandlerDelegate(OnMqttServerApplicationMessageReceived);
110 
111                     await mqttServer.StartAsync(optionsBuilder.Build());
112 
113                     //主动发送消息到客户端
114                     //await mqttServer.PublishAsync(new
115                     //     MqttApplicationMessage
116                     //{
117                     //    Topic = "testtopic",
118                     //    Payload = Encoding.UTF8.GetBytes("dsdsd")
119                     //});
120                     //mqttServer.GetClientStatusAsync();
121                     //mqttServer.GetRetainedApplicationMessagesAsync();
122                     //mqttServer.GetSessionStatusAsync();
123 
124                 }
125             }
126             catch (Exception ex)
127             {
128                 Console.WriteLine($"MQTT Server start fail.>{ex.Message}");
129             }
130         }
131         private void OnMqttServerStarted(EventArgs e)
132         {
133             if (mqttServer.IsStarted)
134             {
135                 Console.WriteLine("MQTT服务启动完成！");
136             }
137         }
138         private void OnMqttServerStopped(EventArgs e)
139         {
140             if (!mqttServer.IsStarted)
141             {
142                 Console.WriteLine("MQTT服务停止完成！");
143             }
144         }
145         private void OnMqttServerClientConnected(MqttServerClientConnectedEventArgs e)
146         {
147             Console.WriteLine($"客户端[{e.ClientId}]已连接");
148         }
149         private void OnMqttServerClientDisconnected(MqttServerClientDisconnectedEventArgs e)
150         {
151             Console.WriteLine($"客户端[{e.ClientId}]已断开连接！");
152         }
153         private void OnMqttServerClientSubscribedTopic(MqttServerClientSubscribedTopicEventArgs e)
154         {
155             Console.WriteLine($"客户端[{e.ClientId}]已成功订阅主题[{e.TopicFilter}]！");
156         }
157         private void OnMqttServerClientUnsubscribedTopic(MqttServerClientUnsubscribedTopicEventArgs e)
158         {
159             Console.WriteLine($"客户端[{e.ClientId}]已成功取消订阅主题[{e.TopicFilter}]！");
160         }
161         private void OnMqttServerApplicationMessageReceived(MqttApplicationMessageReceivedEventArgs e)
162         {
163             messages.Add(e.ApplicationMessage);
164             Console.WriteLine(DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss:fff"));
165             Console.WriteLine($"客户端[{e.ClientId}]>> Topic[{e.ApplicationMessage.Topic}] Payload[{Encoding.UTF8.GetString(e.ApplicationMessage.Payload ?? new byte[] { })}] Qos[{e.ApplicationMessage.QualityOfServiceLevel}] Retain[{e.ApplicationMessage.Retain}]");
166         }
167     }
168 }

View Code

下载客户端测试工具MQTTX，安装即用。

启动程序，启动MQTTX

在MQTTX种新建连接client a，键入Broker地址，端口，账号和密码，然后点击连接

新建主题订阅，订阅主题testtopic/#，#代表所有testtopic下所有的子主题都订阅

新建多一个连接client b，通过client b发布消息给client a

可看到Client a已收到消息。

文章介绍MQTT的部分引用许多人的文章，由于来源太多且我也不知道看的是否为原创，因此不细细列出来了。

——我从来不水帖，我只是代码的搬运工。