C#使用 MQTTnet 快速实现 MQTT 通信

2024-06-04 16:52
文章标签 c# 实现 使用 快速 通信 .net mqtt netcore mqttnet

本文主要是介绍C#使用 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 等参数。

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

得到 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);
}
}
}

这篇关于C#使用 MQTTnet 快速实现 MQTT 通信的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！


http://www.chinasem.cn/article/1030574。

相关文章

Java中流式并行操作parallelStream的原理和使用方法

Java中流式并行操作parallelStream的原理和使用方法

《Java中流式并行操作parallelStream的原理和使用方法》本文详细介绍了Java中的并行流（parallelStream）的原理、正确使用方法以及在实际业务中的应用案例,并指出在使用并行流... 目录Java中流式并行操作parallelStream0. 问题的产生1. 什么是parallelS
阅读更多...
C++中unordered_set哈希集合的实现

C++中unordered_set哈希集合的实现

《C++中unordered_set哈希集合的实现》std::unordered_set是C++标准库中的无序关联容器,基于哈希表实现,具有元素唯一性和无序性特点,本文就来详细的介绍一下unorder... 目录一、概述二、头文件与命名空间三、常用方法与示例1. 构造与析构2. 迭代器与遍历3. 容量相关4
阅读更多...
Linux join命令的使用及说明

Linux join命令的使用及说明

《Linuxjoin命令的使用及说明》`join`命令用于在Linux中按字段将两个文件进行连接,类似于SQL的JOIN,它需要两个文件按用于匹配的字段排序,并且第一个文件的换行符必须是LF,`jo... 目录一. 基本语法二. 数据准备三. 指定文件的连接key四.-a输出指定文件的所有行五.-o指定输出
阅读更多...
Linux jq命令的使用解读

Linux jq命令的使用解读

《Linuxjq命令的使用解读》jq是一个强大的命令行工具,用于处理JSON数据,它可以用来查看、过滤、修改、格式化JSON数据,通过使用各种选项和过滤器,可以实现复杂的JSON处理任务... 目录一. 简介二. 选项2.1.2.2-c2.3-r2.4-R三. 字段提取3.1 普通字段3.2 数组字段四.
阅读更多...
C++中悬垂引用(Dangling Reference) 的实现

C++中悬垂引用(Dangling Reference) 的实现

《C++中悬垂引用(DanglingReference)的实现》C++中的悬垂引用指引用绑定的对象被销毁后引用仍存在的情况,会导致访问无效内存,下面就来详细的介绍一下产生的原因以及如何避免,感兴趣... 目录悬垂引用的产生原因1. 引用绑定到局部变量，变量超出作用域后销毁2. 引用绑定到动态分配的对象，对象
阅读更多...
Linux kill正在执行的后台任务 kill进程组使用详解

Linux kill正在执行的后台任务 kill进程组使用详解

《Linuxkill正在执行的后台任务kill进程组使用详解》文章介绍了两个脚本的功能和区别,以及执行这些脚本时遇到的进程管理问题,通过查看进程树、使用`kill`命令和`lsof`命令,分析了子... 目录零. 用到的命令一. 待执行的脚本二. 执行含子进程的脚本，并kill2.1 进程查看2.2 遇到的
阅读更多...
SpringBoot基于注解实现数据库字段回填的完整方案

SpringBoot基于注解实现数据库字段回填的完整方案

《SpringBoot基于注解实现数据库字段回填的完整方案》这篇文章主要为大家详细介绍了SpringBoot如何基于注解实现数据库字段回填的相关方法,文中的示例代码讲解详细,感兴趣的小伙伴可以了解... 目录数据库表pom.XMLRelationFieldRelationFieldMapping基础的一些代
阅读更多...
Java HashMap的底层实现原理深度解析

Java HashMap的底层实现原理深度解析

《JavaHashMap的底层实现原理深度解析》HashMap基于数组+链表+红黑树结构,通过哈希算法和扩容机制优化性能,负载因子与树化阈值平衡效率,是Java开发必备的高效数据结构,本文给大家介绍... 目录一、概述：HashMap的宏观结构二、核心数据结构解析1. 数组（桶数组）2. 链表节点（Node
阅读更多...
Java AOP面向切面编程的概念和实现方式

Java AOP面向切面编程的概念和实现方式

《JavaAOP面向切面编程的概念和实现方式》AOP是面向切面编程,通过动态代理将横切关注点（如日志、事务）与核心业务逻辑分离,提升代码复用性和可维护性,本文给大家介绍JavaAOP面向切面编程的概... 目录一、AOP 是什么？二、AOP 的核心概念与实现方式核心概念实现方式三、Spring AOP 的关
阅读更多...
详解SpringBoot+Ehcache使用示例

详解SpringBoot+Ehcache使用示例

《详解SpringBoot+Ehcache使用示例》本文介绍了SpringBoot中配置Ehcache、自定义get/set方式,并实际使用缓存的过程,文中通过示例代码介绍的非常详细,对大家的学习或者... 目录摘要概念内存与磁盘持久化存储：配置灵活性：编码示例引入依赖：配置ehcache.XML文件：配置
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2