使用 MQTTnet 快速实现 MQTT 通讯

1 什么是 MQTT ？

MQTT（Message Queuing Telemetry Transport，消息队列遥测传输）是 IBM 开发的一个即时通信协议，有可能成为物联网的重要组成部分。MQTT 是基于二进制消息的发布/订阅编程模式的消息协议，现在已经成为 OASIS 规范，因为规范很简单，很是适合须要低功耗和网络带宽有限的 IoT 场景。MQTT官网

2 MQTTnet

MQTTnet 是一个基于 MQTT 通讯的高性能 .NET 开源库，它同时支持 MQTT 服务器端和客户端。并且做者也保持更新，目前支持新版的.NET core，这也是选择 MQTTnet 的缘由。 MQTTnet 在 Github 并非下载最多的 .NET 的 MQTT 开源库，其余的还 MqttDotNet、nMQTT、M2MQTT 等

MQTTnet is a high performance .NET library for MQTT based communication. It provides a MQTT client and a MQTT server (broker). The implementation is based on the documentation from http://mqtt.org/.

3 建立项目并导入类库

这里咱们使用 Visual Studio 2017 建立一个空解决方案，并在其中添加两个项目，即一个服务端和一个客户端，服务端项目模板选择最新的 .NET Core 控制台应用，客户端项目选择传统的 WinForm 窗体应用程序。.NET Core 项目模板以下图所示：

在解决方案在右键单击-选择“管理解决方案的 NuGet 程序包”-在“浏览”选项卡下面搜索 MQTTnet，为服务端项目和客户端项目都安装上 MQTTnet 库，当前最新稳定版为 2.4.0。项目结构以下图所示：

4 服务端

MQTT 服务端主要用于与多个客户端保持链接，并处理客户端的发布和订阅等逻辑。通常不多直接从服务端发送消息给客户端（可使用 mqttServer.Publish(appMsg); 直接发送消息），多数状况下服务端都是转发主题匹配的客户端消息，在系统中起到一个中介的做用。

4.1 建立服务端并启动

建立服务端最简单的方式是采用 MqttServerFactory 对象的 CreateMqttServer 方法来实现，该方法须要一个 MqttServerOptions 参数。

var options = new MqttServerOptions(); var mqttServer = new MqttServerFactory().CreateMqttServer(options);

经过上述方式建立了一个 IMqttServer 对象后，调用其 StartAsync 方法便可启动 MQTT 服务。值得注意的是：以前版本采用的是 Start 方法，做者也是紧跟 C# 语言新特性，能使用异步的地方也都改成异步方式。

await mqttServer.StartAsync();

4.2 验证客户端

在 MqttServerOptions 选项中，你可使用 ConnectionValidator 来对客户端链接进行验证。好比客户端ID标识 ClientId，用户名 Username 和密码 Password 等。

var options = new MqttServerOptions { ConnectionValidator = c => { if (c.ClientId.Length < 10) { return MqttConnectReturnCode.ConnectionRefusedIdentifierRejected; } if (c.Username != "xxx" || c.Password != "xxx") { return MqttConnectReturnCode.ConnectionRefusedBadUsernameOrPassword; } return MqttConnectReturnCode.ConnectionAccepted; } };

4.3 相关事件

服务端支持 ClientConnected、ClientDisconnected 和 ApplicationMessageReceived 事件，分别用来检查客户端链接、客户端断开以及接收客户端发来的消息。

其中 ClientConnected 和 ClientDisconnected 事件的事件参数一个客户端链接对象 ConnectedMqttClient，经过该对象能够获取客户端ID标识 ClientId 和 MQTT 版本 ProtocolVersion。

ApplicationMessageReceived 的事件参数包含了客户端ID标识 ClientId 和 MQTT 应用消息 MqttApplicationMessage 对象，经过该对象能够获取主题 Topic、QoS QualityOfServiceLevel 和消息内容 Payload 等信息。

5 客户端

MQTT 与 HTTP 不一样，后者是基于请求/响应方式的，服务器端没法直接发送数据给客户端。而 MQTT 是基于发布/订阅模式的,全部的客户端均与服务端保持链接状态。

那么客户端之间是如何通讯的呢？

具体逻辑是：某些客户端向服务端订阅它感兴趣（主题）的消息，另外一些客户端向服务端发布（主题）消息，服务端将订阅和发布的主题进行匹配，并将消息转发给匹配经过的客户端。

5.1 建立客户端并链接

使用 MQTTnet 建立 MQTT 也很是简单，只须要使用 MqttClientFactory 对象的 CreateMqttClient 方法便可。

var mqttClient = new MqttClientFactory().CreateMqttClient();

建立客户端对象后，调用其异步方法 ConnectAsync 来链接到服务端。

await mqttClient.ConnectAsync(options);

调用该方法时须要传递一个 MqttClientTcpOptions 对象（以前的版本是在建立对象时使用该选项），该选项包含了客户端ID标识 ClientId、服务端地址（可使用IP地址或域名）Server、端口号 Port、用户名 UserName、密码 Password 等信息。

var options = new MqttClientTcpOptions { Server = "127.0.0.1", ClientId = "c001", UserName = "u001", Password = "p001", CleanSession = true };

5.2 相关事件

客户端支持 Connected、Disconnected 和 ApplicationMessageReceived 事件，用来处理客户端与服务端链接、客户端从服务端断开以及客户端收到消息的事情。

5.2 订阅消息

客户端链接到服务端以后，可使用 SubscribeAsync 异步方法订阅消息，该方法能够传入一个可枚举或可变参数的主题过滤器 TopicFilter 参数，主题过滤器包含主题名和 QoS 等级。

mqttClient.SubscribeAsync(new List<TopicFilter> { new TopicFilter("家/客厅/空调/#", MqttQualityOfServiceLevel.AtMostOnce) });

5.3 发布消息

发布消息前须要先构建一个消息对象 MqttApplicationMessage，最直接的方法是使用其实构造函数，传入主题、内容、Qos 等参数。

var appMsg = new MqttApplicationMessage("家/客厅/空调/开关", Encoding.UTF8.GetBytes("消息内容"), MqttQualityOfServiceLevel.AtMostOnce, false);

获得 MqttApplicationMessage 消息对象后，经过客户端对象调用其 PublishAsync 异步方法进行消息发布。

mqttClient.PublishAsync(appMsg);

6 跟踪消息

MQTTnet 提供了一个静态类 MqttNetTrace 来对消息进行跟踪，该类可用于服务端和客户端。MqttNetTrace 的事件 TraceMessagePublished 用于跟踪服务端和客户端应用的日志消息，好比启动、中止、心跳、消息订阅和发布等。事件参数 MqttNetTraceMessagePublishedEventArgs 包含了线程ID ThreadId、来源 Source、日志级别 Level、日志消息 Message、异常信息 Exception 等。

MqttNetTrace.TraceMessagePublished += MqttNetTrace_TraceMessagePublished;

private static void MqttNetTrace_TraceMessagePublished(object sender, MqttNetTraceMessagePublishedEventArgs e) { Console.WriteLine($">> 线程ID：{e.ThreadId} 来源：{e.Source} 跟踪级别：{e.Level} 消息: {e.Message}"); if (e.Exception != null) { Console.WriteLine(e.Exception); } }

同时 MqttNetTrace 类还提供了4个不一样消息等级的静态方法，Verbose、Information、Warning 和 Error，用于给出不一样级别的日志消息，该消息将会在 TraceMessagePublished 事件中输出，你可使用 e.Level 进行过虑。

7 运行效果

如下分别是服务端、客户端1和客户端2的运行效果，其中客户端1和客户端2只是同一个项目运行了两个实例。客户端1用于订阅传感器的“温度”数据，并模拟上位机（如 APP 等）发送开关控制命令；客户端2订阅上位机传来的“开关”控制命令，并模拟温度传感器上报温度数据。

7.1 服务端

7.2 客户端1

7.2 客户端2

8 Demo代码

8.1 服务端代码

using MQTTnet; using MQTTnet.Core.Adapter; using MQTTnet.Core.Diagnostics; using MQTTnet.Core.Protocol; using MQTTnet.Core.Server; using System; using System.Text; using System.Threading; namespace MqttServerTest { class Program { private static MqttServer mqttServer = null; static void Main(string[] args) { MqttNetTrace.TraceMessagePublished += MqttNetTrace_TraceMessagePublished; new Thread(StartMqttServer).Start(); while (true) { var inputString = Console.ReadLine().ToLower().Trim(); if (inputString == "exit") { mqttServer?.StopAsync(); Console.WriteLine("MQTT服务已中止！"); break; } else if (inputString == "clients") { foreach (var item in mqttServer.GetConnectedClients()) { Console.WriteLine($"客户端标识：{item.ClientId}，协议版本：{item.ProtocolVersion}"); } } else { Console.WriteLine($"命令[{inputString}]无效！"); } } } private static void StartMqttServer() { if (mqttServer == null) { try { var options = new MqttServerOptions { ConnectionValidator = p => { if (p.ClientId == "c001") { if (p.Username != "u001" || p.Password != "p001") { return MqttConnectReturnCode.ConnectionRefusedBadUsernameOrPassword; } } return MqttConnectReturnCode.ConnectionAccepted; } }; mqttServer = new MqttServerFactory().CreateMqttServer(options) as MqttServer; mqttServer.ApplicationMessageReceived += MqttServer_ApplicationMessageReceived; mqttServer.ClientConnected += MqttServer_ClientConnected; mqttServer.ClientDisconnected += MqttServer_ClientDisconnected; } catch (Exception ex) { Console.WriteLine(ex.Message); return; } } mqttServer.StartAsync(); Console.WriteLine("MQTT服务启动成功！"); } private static void MqttServer_ClientConnected(object sender, MqttClientConnectedEventArgs e) { Console.WriteLine($"客户端[{e.Client.ClientId}]已链接，协议版本：{e.Client.ProtocolVersion}"); } private static void MqttServer_ClientDisconnected(object sender, MqttClientDisconnectedEventArgs e) { Console.WriteLine($"客户端[{e.Client.ClientId}]已断开链接！"); } private static void MqttServer_ApplicationMessageReceived(object sender, MqttApplicationMessageReceivedEventArgs e) { Console.WriteLine($"客户端[{e.ClientId}]>> 主题：{e.ApplicationMessage.Topic} 负荷：{Encoding.UTF8.GetString(e.ApplicationMessage.Payload)} Qos：{e.ApplicationMessage.QualityOfServiceLevel} 保留：{e.ApplicationMessage.Retain}"); } private static void MqttNetTrace_TraceMessagePublished(object sender, MqttNetTraceMessagePublishedEventArgs e) { /*Console.WriteLine($">> 线程ID：{e.ThreadId} 来源：{e.Source} 跟踪级别：{e.Level} 消息: {e.Message}"); if (e.Exception != null) { Console.WriteLine(e.Exception); }*/ } } }

8.2 客户端代码

using MQTTnet; using MQTTnet.Core; using MQTTnet.Core.Client; using MQTTnet.Core.Packets; using MQTTnet.Core.Protocol; using System; using System.Collections.Generic; using System.Text; using System.Threading.Tasks; using System.Windows.Forms; namespace MqttClientWin { public partial class FmMqttClient : Form { private MqttClient mqttClient = null; public FmMqttClient() { InitializeComponent(); Task.Run(async () => { await ConnectMqttServerAsync(); }); } private async Task ConnectMqttServerAsync() { if (mqttClient == null) { mqttClient = new MqttClientFactory().CreateMqttClient() as MqttClient; mqttClient.ApplicationMessageReceived += MqttClient_ApplicationMessageReceived; mqttClient.Connected += MqttClient_Connected; mqttClient.Disconnected += MqttClient_Disconnected; } try { var options = new MqttClientTcpOptions { Server = "127.0.0.1", ClientId = Guid.NewGuid().ToString().Substring(0, 5), UserName = "u001", Password = "p001", CleanSession = true }; await mqttClient.ConnectAsync(options); } catch (Exception ex) { Invoke((new Action(() => { txtReceiveMessage.AppendText($"链接到MQTT服务器失败！" + Environment.NewLine + ex.Message + Environment.NewLine); }))); } } private void MqttClient_Connected(object sender, EventArgs e) { Invoke((new Action(() => { txtReceiveMessage.AppendText("已链接到MQTT服务器！" + Environment.NewLine); }))); } private void MqttClient_Disconnected(object sender, EventArgs e) { Invoke((new Action(() => { txtReceiveMessage.AppendText("已断开MQTT链接！" + Environment.NewLine); }))); } private void MqttClient_ApplicationMessageReceived(object sender, MqttApplicationMessageReceivedEventArgs e) { Invoke((new Action(() => { txtReceiveMessage.AppendText($">> {Encoding.UTF8.GetString(e.ApplicationMessage.Payload)}{Environment.NewLine}"); }))); } private void BtnSubscribe_ClickAsync(object sender, EventArgs e) { string topic = txtSubTopic.Text.Trim(); if (string.IsNullOrEmpty(topic)) { MessageBox.Show("订阅主题不能为空！"); return; } if (!mqttClient.IsConnected) { MessageBox.Show("MQTT客户端还没有链接！"); return; } mqttClient.SubscribeAsync(new List<TopicFilter> { new TopicFilter(topic, MqttQualityOfServiceLevel.AtMostOnce) }); txtReceiveMessage.AppendText($"已订阅[{topic}]主题" + Environment.NewLine); txtSubTopic.Enabled = false; btnSubscribe.Enabled = false; } private void BtnPublish_Click(object sender, EventArgs e) { string topic = txtPubTopic.Text.Trim(); if (string.IsNullOrEmpty(topic)) { MessageBox.Show("发布主题不能为空！"); return; } string inputString = txtSendMessage.Text.Trim(); var appMsg = new MqttApplicationMessage(topic, Encoding.UTF8.GetBytes(inputString), MqttQualityOfServiceLevel.AtMostOnce, false); mqttClient.PublishAsync(appMsg); } } }

9 参考

• 快速搭建MQTT服务器(MQTTnet和Apache Apollo)
• MQTTnet
• “mqtt” - 可译网
• MQTT Essentials