场景

MQTT

MQTT（Message Queuing Telemetry Transport，消息队列遥测传输协议），是一种基于发布/订阅（publish/subscribe）模式的"轻量级"通讯协议，该协议构建于TCP/IP协议上，由IBM在1999年发布。MQTT最大优点在于，可以以极少的代码和有限的带宽，为连接远程设备提供实时可靠的消息服务。作为一种低开销、低带宽占用的即时通讯协议，使其在物联网、小型设备、移动应用等方面有较广泛的应用。

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

MQTTnet

MQTTnet 是一个基于 MQTT 通信的高性能 .NET 开源库，它同时支持 MQTT 服务器端和客户端。

注：

博客：
https://blog.csdn.net/badao_liumang_qizhi
关注公众号
霸道的程序猿
获取编程相关电子书、教程推送与免费下载。

实现

打开VS,新建mqtt的服务端项目

选择将解决方案和项目放在同一个目录下

然后选择新建控制台应用程序，并且选择目标框架为.NET Core3.1

然后在此解决方案下右击新建项目

新建Winform项目，作为Mqtt的客户端

然后在解决方案上右击选择-管理解决方案的Nuget 程序包

然后在浏览中搜索MQTTnet,右边勾选需要添加依赖的项目，这里客户端和服务端都需要添加。然后下面选择指定版本，这里是2.4.0，点击安装

安装过程中会提示预览更改，点击确定，以及会提示接受协议。

安装成功后，就可以在项目下看到依赖项

然后编写服务端的代码，打开Program.cs,修改如下

using System;
using MQTTnet;
using MQTTnet.Core.Adapter;
using MQTTnet.Core.Diagnostics;
using MQTTnet.Core.Protocol;
using MQTTnet.Core.Server;
using System.Text;
using System.Threading;namespace MqttnetServer
{class Program{//MQtt服务端private static MqttServer mqttServer = null;static void Main(string[] args){//MQTTnet 提供了一个静态类 MqttNetTrace 来对消息进行跟踪//MqttNetTrace 的事件 TraceMessagePublished 用于跟踪服务端和客户端应用的日志消息，比如启动、停止、心跳、消息订阅和发布等MqttNetTrace.TraceMessagePublished += MqttNetTrace_TraceMessagePublished;//启动服务端new Thread(StartMqttServer).Start();while (true){//获取输入字符var inputString = Console.ReadLine().ToLower().Trim();//exit则停止服务if (inputString == "exit"){mqttServer.StopAsync();Console.WriteLine("MQTT服务已停止！");break;}//clients则输出所有客户端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{//在 MqttServerOptions 选项中，你可以使用 ConnectionValidator 来对客户端连接进行验证。//比如客户端ID标识 ClientId，用户名 Username 和密码 Password 等。var options = new MqttServerOptions{ConnectionValidator = p =>{if (p.ClientId == "c001"){if (p.Username != "u001" || p.Password != "p001"){return MqttConnectReturnCode.ConnectionRefusedBadUsernameOrPassword;}}return MqttConnectReturnCode.ConnectionAccepted;}};//创建服务端最简单的方式是采用 MqttServerFactory 对象的 CreateMqttServer 方法来实现，该方法需要一个//MqttServerOptions 参数。mqttServer = new MqttServerFactory().CreateMqttServer(options) as MqttServer;//服务端支持 ClientConnected、ClientDisconnected 和 ApplicationMessageReceived 事件，//分别用来检查客户端连接、客户端断开以及接收客户端发来的消息。//ApplicationMessageReceived 的事件参数包含了客户端ID标识 ClientId 和 MQTT 应用消息 MqttApplicationMessage 对象，//通过该对象可以获取主题 Topic、QoS QualityOfServiceLevel 和消息内容 Payload 等信息mqttServer.ApplicationMessageReceived += MqttServer_ApplicationMessageReceived;//ClientConnected 和 ClientDisconnected 事件的事件参数一个客户端连接对象 ConnectedMqttClient//通过该对象可以获取客户端ID标识 ClientId 和 MQTT 版本 ProtocolVersion。mqttServer.ClientConnected += MqttServer_ClientConnected;mqttServer.ClientDisconnected += MqttServer_ClientDisconnected;}catch (Exception ex){Console.WriteLine(ex.Message); return;}}//创建了一个 IMqttServer 对象后，调用其 StartAsync 方法即可启动 MQTT 服务mqttServer.StartAsync();Console.WriteLine("MQTT服务启动成功！");}//ClientConnected 和 ClientDisconnected 事件的事件参数一个客户端连接对象 ConnectedMqttClient//通过该对象可以获取客户端ID标识 ClientId 和 MQTT 版本 ProtocolVersion。private static void MqttServer_ClientConnected(object sender, MqttClientConnectedEventArgs e){Console.WriteLine($"客户端[{e.Client.ClientId}]已连接，协议版本：{e.Client.ProtocolVersion}");}//ClientConnected 和 ClientDisconnected 事件的事件参数一个客户端连接对象 ConnectedMqttClient//通过该对象可以获取客户端ID标识 ClientId 和 MQTT 版本 ProtocolVersion。private static void MqttServer_ClientDisconnected(object sender, MqttClientDisconnectedEventArgs e){Console.WriteLine($"客户端[{e.Client.ClientId}]已断开连接！");}//ApplicationMessageReceived 的事件参数包含了客户端ID标识 ClientId 和 MQTT 应用消息 MqttApplicationMessage 对象，//通过该对象可以获取主题 Topic、QoS QualityOfServiceLevel 和消息内容 Payload 等信息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}");}//事件参数 MqttNetTraceMessagePublishedEventArgs 包含了线程ID ThreadId、来源 Source、日志级别 Level、日志消息 Message、异常信息 Exception 等。//MqttNetTrace 类还提供了4个不同消息等级的静态方法，Verbose、Information、Warning 和 Error，//用于给出不同级别的日志消息，该消息将会在 TraceMessagePublished 事件中输出，//你可以使用 e.Level 进行过虑。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); }}}
}

然后启动服务端

然后修改客户端的窗体页面，添加如下控件布局

连接按钮的点击事件代码为

        //连接到Mqtt服务器private void button_Connect_Click(object sender, EventArgs e){if (string.IsNullOrEmpty(textServerAddress.Text)){MessageBox.Show("服务器地址不能为空");}else {Task.Run(async () => {await ConnectMqttServerAsync();});}}

订阅并保存到文件按钮的点击事件为

       //订阅主题按钮点击事件private void btnSubscribe_Click(object sender, EventArgs e){string topic = txtSubTopic.Text.Trim();if (string.IsNullOrEmpty(topic)){MessageBox.Show("订阅主题不能为空！");return;}if (!mqttClient.IsConnected){MessageBox.Show("MQTT客户端尚未连接！");return;}//客户端连接到服务端之后，可以使用 SubscribeAsync 异步方法订阅消息，//该方法可以传入一个可枚举或可变参数的主题过滤器 TopicFilter 参数，主题过滤器包含主题名和 QoS 等级。mqttClient.SubscribeAsync(new List<TopicFilter> {new TopicFilter(topic, MqttQualityOfServiceLevel.AtMostOnce)});txtReceiveMessage.AppendText($"已订阅[{topic}]主题" + Environment.NewLine);}

取消订阅按钮的点击事件为

        //取消订阅按钮点击事件private void btn_cancle_sub_Click(object sender, EventArgs e){string topic = txtSubTopic.Text.Trim();mqttClient.UnsubscribeAsync(new List<String> {topic});}

发布按钮的点击事件为

        //发布主题private void button2_Click_1(object sender, EventArgs e){string topic = txtPubTopic.Text.Trim();if (string.IsNullOrEmpty(topic)){MessageBox.Show("发布主题不能为空！"); return;}string inputString = txtSendMessage.Text.Trim();//发布消息前需要先构建一个消息对象 MqttApplicationMessage，//最直接的方法是使用其实构造函数，传入主题、内容、Qos 等参数。var appMsg = new MqttApplicationMessage(topic, Encoding.UTF8.GetBytes(inputString), MqttQualityOfServiceLevel.AtMostOnce, false);//得到 MqttApplicationMessage 消息对象后，//通过客户端对象调用其 PublishAsync 异步方法进行消息发布。mqttClient.PublishAsync(appMsg);}

完整客户端示例代码为

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.IO;
using System.Text;
using System.Threading.Tasks;
using System.Windows.Forms;namespace MqttnetClient
{public partial class Form1 : Form{//MQTTk客户端private MqttClient mqttClient = null;public Form1(){InitializeComponent();}private async Task ConnectMqttServerAsync(){if (mqttClient == null){//使用 MQTTnet 创建 MQTT 也非常简单//只需要使用 MqttClientFactory 对象的 CreateMqttClient 方法即可mqttClient = new MqttClientFactory().CreateMqttClient() as MqttClient;//客户端支持 Connected、Disconnected 和 ApplicationMessageReceived 事件，//用来处理客户端与服务端连接、客户端从服务端断开以及客户端收到消息的事情。mqttClient.ApplicationMessageReceived += MqttClient_ApplicationMessageReceived;mqttClient.Connected += MqttClient_Connected;mqttClient.Disconnected += MqttClient_Disconnected;}try{//调用ConnectAsync方法时需要传递一个 MqttClientTcpOptions 对象//选项包含了客户端ID标识 ClientId、服务端地址（可以使用IP地址或域名）Server、端口号 Port、//用户名 UserName、密码 Password 等信息。var options = new MqttClientTcpOptions { };if (!string.IsNullOrEmpty(textPort.Text)){options = new MqttClientTcpOptions{Server = textServerAddress.Text,ClientId = Guid.NewGuid().ToString().Substring(0, 5),UserName = textusername.Text,Password = textpassword.Text,Port = int.Parse(textPort.Text),CleanSession = true};}else {options = new MqttClientTcpOptions{Server = textServerAddress.Text,ClientId = Guid.NewGuid().ToString().Substring(0, 5),UserName = textusername.Text,Password = textpassword.Text,CleanSession = true};}//创建客户端对象后，调用其异步方法 ConnectAsync 来连接到服务端。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}");//将收到的消息追加到相对路径record.txt文件if (!File.Exists("record.txt")){File.Create("record.txt");}else{File.AppendAllText("record.txt", "\r\n");File.AppendAllText("record.txt", Encoding.Default.GetString(e.ApplicationMessage.Payload));}})));}//订阅主题按钮点击事件private void btnSubscribe_Click(object sender, EventArgs e){string topic = txtSubTopic.Text.Trim();if (string.IsNullOrEmpty(topic)){MessageBox.Show("订阅主题不能为空！");return;}if (!mqttClient.IsConnected){MessageBox.Show("MQTT客户端尚未连接！");return;}//客户端连接到服务端之后，可以使用 SubscribeAsync 异步方法订阅消息，//该方法可以传入一个可枚举或可变参数的主题过滤器 TopicFilter 参数，主题过滤器包含主题名和 QoS 等级。mqttClient.SubscribeAsync(new List<TopicFilter> {new TopicFilter(topic, MqttQualityOfServiceLevel.AtMostOnce)});txtReceiveMessage.AppendText($"已订阅[{topic}]主题" + Environment.NewLine);}//发布主题private void button2_Click_1(object sender, EventArgs e){string topic = txtPubTopic.Text.Trim();if (string.IsNullOrEmpty(topic)){MessageBox.Show("发布主题不能为空！"); return;}string inputString = txtSendMessage.Text.Trim();//发布消息前需要先构建一个消息对象 MqttApplicationMessage，//最直接的方法是使用其实构造函数，传入主题、内容、Qos 等参数。var appMsg = new MqttApplicationMessage(topic, Encoding.UTF8.GetBytes(inputString), MqttQualityOfServiceLevel.AtMostOnce, false);//得到 MqttApplicationMessage 消息对象后，//通过客户端对象调用其 PublishAsync 异步方法进行消息发布。mqttClient.PublishAsync(appMsg);}//连接到Mqtt服务器private void button_Connect_Click(object sender, EventArgs e){if (string.IsNullOrEmpty(textServerAddress.Text)){MessageBox.Show("服务器地址不能为空");}else {Task.Run(async () => {await ConnectMqttServerAsync();});}}//取消订阅按钮点击事件private void btn_cancle_sub_Click(object sender, EventArgs e){string topic = txtSubTopic.Text.Trim();mqttClient.UnsubscribeAsync(new List<String> {topic});}}
}

运行客户端

这里服务器地址就是本机地址，端口或者用户名和密码根据自己需要

然后就可以进行主题的订阅和发布了

然后在客户端bin目录下找到record.txt,可以看到追加到文件成功。

示例代码下载

https://download.csdn.net/download/BADAO_LIUMANG_QIZHI/19431291

参考文章：

https://www.cnblogs.com/kuige/articles/7724786.html

Winform中使用MQTTnet实现MQTT的服务端和客户端之间的通信以及将订阅的消息保存到文件相关推荐

java贪吃蛇客户端服务器_java Socket套接字TCP编程开发服务端和客户端之间的通信 - 贪吃蛇学院-专业IT技术平台...
超级简单,没有太多实质内容的Socket服务端,客户端小程序先运行server 再运行client OK 服务端代码如下: public class Server { public static v ...
Java中利用socket实现简单的服务端与客户端的通信（中级）——实现任意双向通信
本文计划采用socket实现客户端和服务端的任意双向通信,即客户端可以随时给服务端发消息,服务端也可以随时给客户端发消息,最终结果就是一个类似与QQ的聊天软件的功能. 以下代码可以直接拷贝到Eclip ...
Java中利用socket实现简单的服务端与客户端的通信（基础级）
在上一篇文章中,简单的介绍了java中入门级的socket编程,简单的实现了客户端像服务器端发送数据,服务器端将数据接收并显示在控制台,没有涉及多线程.上一篇文章的链接:Java中利用socket实现 ...
Java中利用socket实现简单的服务端与客户端的通信（入门级）
Java编程中,要想要使用网络通信,就离不开Socket编程,在此对socket进行简单的介绍.首先声明,这是一个入门级的介绍,仅仅简单的实现了客户端向服务端发送数据,服务端正常的接收数据,当接收到特 ...
oracle 设置监听和服务,oracle服务端和客户端之间的网络监听设置
下面为您介绍的是oracle服务端和客户端之间的网络监听设置,如果您在oracle服务端和客户端之间的网络监听设置方面遇到过类似的问题,不妨一看. oracle服务端和客户端之间的网络监听设置假如我 ...
oracle 设置监听和服务,oracle服务端和客户端之间的网络监听如何设置呢？
racle服务端和客户端之间的网络监听设置假如我现在有两个数据库kkman和orcl,现在我想在服务端设立两个监听来处理从客户端发过来的连接请求. ------------------------- ...
[C语言]一个很实用的服务端和客户端进行UDP通信的实例
前段时间发了个TCP通信的例子,现在再来一个UDP通信的例子.这些可以作为样本程序,用到开发中."裸写"socket老是记不住步骤,经常被鄙视-- 下面的例子很简单,写一个UDP的 ...
Socket代码实现服务端和客户端之间通信
服务端代码 // Socket_connection.cpp : 此文件包含 "main" 函数.程序执行将在此处开始并结束. // #ifndef UNICODE #define ...
SpringBoot(23) 集成socket.io服务端和客户端实现通信
一.前言 websocket和socket.io区别? websocket 一种让客户端和服务器之间能进行双向实时通信的技术使用时,虽然主流浏览器都已经支持,但仍然可能有不兼容的情况适合用于cli ...

Winform中使用MQTTnet实现MQTT的服务端和客户端之间的通信以及将订阅的消息保存到文件

场景

MQTT

MQTTnet

实现

示例代码下载

Winform中使用MQTTnet实现MQTT的服务端和客户端之间的通信以及将订阅的消息保存到文件相关推荐

最新文章

热门文章