MQTTNET mqtt客户端 mqtt发布 mqtt订阅 .net C#MQTT编程06--MQTT服务器和客户端(winform版)

Lyoune 博客 2024-06-04 1 0

1、前言

介绍完基础理论部分，下面在Windows平台上搭建一个简单的MQTT应用，进行简单的应用，整体架构如下图所示；

消息模型：

运用MQTT协议，设备可以很方便地连接到物联网云服务，管理设备并处理数据，最后应用到各种业务场景，如下图所示

前面介绍过，MQTT可以运行在几乎所有的平台，windows,linux什么的都可以，各种语言都有实现MQTT的组件，.net也好，Java也好，都有封装好的mqtt服务器和客户端组件或插件来实现，本系列是在.net平台下实现mqtt的服务器和客户端。

常见的MQTT服务器包括Eclipse Mosquitto、EMQ X、HiveMQ、RabbitMQ、MQTTNET等，本系列文章都是基于.net平台的mqtt服务通信，开发环境vs2022，.net framework4.8。

2、服务器搭建

1、创建项目方案

2、添加库引用

3、UI布局

4、控件代码

“启动”按钮

停止按钮

窗体加载

全部完整代码

注意这里面用到了多线程及任务task，委托action，异步async及await的技术，在mqtt中必须使用这些技术，否则界面会卡死。

using MQTTnet.Client.Receiving;

using MQTTnet.Server;

using MQTTnet;

using System;

using System.Collections.Generic;

using System.ComponentModel;

using System.Data;

using System.Drawing;

using System.Linq;

using System.Text;

using System.Threading.Tasks;

using System.Windows.Forms;

using MQTTnet.Protocol;

namespace MQTTNETServerForms

{

public partial class Form1 : Form

{

private IMqttServer server;//mqtt服务器对象

List Topics = new List();

public Form1()

{

InitializeComponent();

}

private void Form1_Load(object sender, EventArgs e)

{

//创建服务器对象

server = new MqttFactory().CreateMqttServer();

server.ApplicationMessageReceivedHandler = new MqttApplicationMessageReceivedHandlerDelegate(new Action(Server_ApplicationMessageReceived));//绑定消息接收事件

server.ClientConnectedHandler = new MqttServerClientConnectedHandlerDelegate(new Action(Server_ClientConnected));//绑定客户端连接事件

server.ClientDisconnectedHandler = new MqttServerClientDisconnectedHandlerDelegate(new Action(Server_ClientDisconnected));//绑定客户端断开事件

server.ClientSubscribedTopicHandler = new MqttServerClientSubscribedHandlerDelegate(new Action(Server_ClientSubscribedTopic));//绑定客户端订阅主题事件

server.ClientUnsubscribedTopicHandler = new MqttServerClientUnsubscribedTopicHandlerDelegate(new Action(Server_ClientUnsubscribedTopic));//绑定客户端退订主题事件

server.StartedHandler = new MqttServerStartedHandlerDelegate(new Action(Server_Started));//绑定服务端启动事件

server.StoppedHandler = new MqttServerStoppedHandlerDelegate(new Action(Server_Stopped));//绑定服务端停止事件

}

///

/// 绑定消息接收事件

///

private void Server_ApplicationMessageReceived(MqttApplicationMessageReceivedEventArgs e)

{

string msg = e.ApplicationMessage.ConvertPayloadToString();

WriteLog(">>> 收到消息：" + msg + "，QoS =" + e.ApplicationMessage.QualityOfServiceLevel + "，客户端=" + e.ClientId + "，主题：" + e.ApplicationMessage.Topic);

}

///

/// 绑定客户端连接事件

///

private void Server_ClientConnected(MqttServerClientConnectedEventArgs e)

{

Task.Run(new Action(() =>

{

lbClients.BeginInvoke(new Action(() =>

{

lbClients.Items.Add(e.ClientId);

}));

WriteLog(">>> 客户端" + e.ClientId + "连接");

}

///

/// 绑定客户端断开事件

///

private void Server_ClientDisconnected(MqttServerClientDisconnectedEventArgs e)

{

Task.Run(new Action(() =>

{

lbClients.BeginInvoke(new Action(() =>

{

lbClients.Items.Remove(e.ClientId);

}));

WriteLog(">>> 客户端" + e.ClientId + "断开");

}

///

/// 绑定客户端订阅主题事件

///

private void Server_ClientSubscribedTopic(MqttServerClientSubscribedTopicEventArgs e)

{

Task.Run(new Action(() =>

{

var topic = Topics.FirstOrDefault(t => t.Topic == e.TopicFilter.Topic);

if (topic == null)

{

topic = new TopicItem { Topic = e.TopicFilter.Topic, Count = 0 };

Topics.Add(topic);

}

if (!topic.Clients.Exists(c => c == e.ClientId))

{

topic.Clients.Add(e.ClientId);

topic.Count++;

}

lvTopic.Invoke(new Action(() =>

{

this.lvTopic.Items.Clear();

}));

foreach (var item in this.Topics)

{

lvTopic.Invoke(new Action(() =>

{

this.lvTopic.Items.Add($"{item.Topic}:{item.Count}");

}));

}

}));

WriteLog(">>> 客户端" + e.ClientId + "订阅主题" + e.TopicFilter.Topic);

}

///

/// 绑定客户端退订主题事件

///

private void Server_ClientUnsubscribedTopic(MqttServerClientUnsubscribedTopicEventArgs e)

{

Task.Run(new Action(() =>

{

var topic = Topics.FirstOrDefault(t => t.Topic == e.TopicFilter);

if (topic != null)

{

topic.Count--;

topic.Clients.Remove(e.ClientId);

}

this.lvTopic.Items.Clear();

foreach (var item in this.Topics)

{

this.lvTopic.Items.Add($"{item.Topic}:{item.Count}");

}

}));

WriteLog(">>> 客户端" + e.ClientId + "退订主题" + e.TopicFilter);

}

///

/// 绑定服务端启动事件

///

private void Server_Started(EventArgs e)

{

WriteLog(">>> 服务端已启动！");

}

///

/// 绑定服务端停止事件

///

private void Server_Stopped(EventArgs e)

{

WriteLog(">>> 服务端已停止！");

}

///

/// 显示日志

///

public void WriteLog(string message)

{

if (txtMsg.InvokeRequired)

{

txtMsg.Invoke(new Action(() =>

{

txtMsg.Text = "";

txtMsg.Text = (message + "\r");

}));

}

else

{

txtMsg.Text = "";

txtMsg.Text = (message + "\r");

}

///

/// 启动

///

[Obsolete]

private async void btnStart_Click(object sender, EventArgs e)

{

var optionBuilder = new MqttServerOptionsBuilder()

.WithDefaultEndpointBoundIPAddress(System.Net.IPAddress.Parse(this.txtIP.Text))

.WithDefaultEndpointPort(int.Parse(this.txtPort.Text))

.WithDefaultCommunicationTimeout(TimeSpan.FromMilliseconds(5000))

.WithConnectionValidator(t =>

{

string un = "", pwd = "";

un = this.txtUname.Text;

pwd = this.txtUpwd.Text;

if (t.Username != un || t.Password != pwd)

{

t.ReturnCode = MqttConnectReturnCode.ConnectionRefusedBadUsernameOrPassword;

}

else

{

t.ReturnCode = MqttConnectReturnCode.ConnectionAccepted;

}

});

var option = optionBuilder.Build();

//启动

await server.StartAsync(option);

WriteLog(">>> 服务器启动成功");

}

///

/// 停止

///

private void btnStop_Click(object sender, EventArgs e)

{

if (server != null)

{

server.StopAsync();

}

注意这个端口，帐号，密码可以自己决定，Ip地址也是。

注意这里面的代码，服务器上必须注册绑定实现的几个事件：消息接收事件，客户端连接事件，客户端断开事件，客户端订阅主题事件，客户端退订主题事件，服务端启动事件，服务端停止事件

server.ApplicationMessageReceivedHandler = new MqttApplicationMessageReceivedHandlerDelegate(new Action(Server_ApplicationMessageReceived));//绑定消息接收事件

server.ClientConnectedHandler = new MqttServerClientConnectedHandlerDelegate(new Action(Server_ClientConnected));//绑定客户端连接事件

server.ClientDisconnectedHandler = new MqttServerClientDisconnectedHandlerDelegate(new Action(Server_ClientDisconnected));//绑定客户端断开事件

server.ClientSubscribedTopicHandler = new MqttServerClientSubscribedHandlerDelegate(new Action(Server_ClientSubscribedTopic));//绑定客户端订阅主题事件

server.ClientUnsubscribedTopicHandler = new MqttServerClientUnsubscribedTopicHandlerDelegate(new Action(Server_ClientUnsubscribedTopic));//绑定客户端退订主题事件

server.StartedHandler = new MqttServerStartedHandlerDelegate(new Action(Server_Started));//绑定服务端启动事件

server.StoppedHandler = new MqttServerStoppedHandlerDelegate(new Action(Server_Stopped));//绑定服务端停止事件

启动测试服务器

启动成功。

3、客户端创建

1、添加项目

2、添加库引用

注意这里添加的与服务器不一样，别混错了

3、UI布局

布局使用的是常规的label，textbox，button

4、控件代码

连接代码

订阅代码

发布代码

完整代码

using MQTTnet.Client.Options;

using MQTTnet.Client;

using MQTTnet.Extensions.ManagedClient;

using System;

using System.Collections.Generic;

using System.ComponentModel;

using System.Data;

using System.Drawing;

using System.Linq;

using System.Text;

using System.Threading.Tasks;

using System.Windows.Forms;

using MQTTnet;

namespace MQTTNETClientForms

{

public partial class Form1 : Form

{

private IManagedMqttClient mqttClient;//客户端mqtt对象

public Form1()

{

InitializeComponent();

}

///

/// 连接

///

private async void btnConn_Click(object sender, EventArgs e)

{

var mqttClientOptions = new MqttClientOptionsBuilder()

.WithClientId(this.txtId.Text)

.WithTcpServer(this.txtIP.Text, int.Parse(this.txtPort.Text))

.WithCredentials(this.txtName.Text, this.txtUpwd.Text);

var options = new ManagedMqttClientOptionsBuilder()

.WithAutoReconnectDelay(TimeSpan.FromSeconds(5))

.WithClientOptions(mqttClientOptions.Build())

.Build();

//开启

await mqttClient.StartAsync(options);

}

///

/// 断开

///

private async void btnClose_Click(object sender, EventArgs e)

{

if (mqttClient != null)

{

if (mqttClient.IsStarted)

{

await mqttClient.StopAsync();

}

mqttClient.Dispose();

}

private void Form1_Load(object sender, EventArgs e)

{

var factory = new MqttFactory();

mqttClient = factory.CreateManagedMqttClient();//创建客户端对象

//绑定断开事件

mqttClient.UseDisconnectedHandler(async ee =>

{

WriteLog("与服务器之间的连接断开了，正在尝试重新连接");

// 等待 5s 时间

await Task.Delay(TimeSpan.FromSeconds(5));

try

{

mqttClient.UseConnectedHandler(tt =>

{

WriteLog(">>> 连接到服务成功");

});

}

catch (Exception ex)

{

WriteLog($"重新连接服务器失败:{ex}");

}

});

//绑定接收事件

mqttClient.UseApplicationMessageReceivedHandler(aa =>

{

try

{

string msg = aa.ApplicationMessage.ConvertPayloadToString();

WriteLog(">>> 消息：" + msg + "，QoS =" + aa.ApplicationMessage.QualityOfServiceLevel + "，客户端=" + aa.ClientId + "，主题：" + aa.ApplicationMessage.Topic);

}

catch (Exception ex)

{

WriteLog($"+ 消息 = " + ex.Message);

}

});

//绑定连接事件

mqttClient.UseConnectedHandler(ee =>

{

WriteLog(">>> 连接到服务成功");

});

}

///

/// 显示日志

///

private void WriteLog(string message)

{

if (txtMsg.InvokeRequired)

{

txtMsg.Invoke(new Action(() =>

{

txtMsg.Text = (message);

}));

}

else

{

txtMsg.Text = (message);

}

///

/// 订阅

///

[Obsolete]

private async void btnSub_Click(object sender, EventArgs e)

{

if (string.IsNullOrWhiteSpace(this.txtTopic.Text))

{

WriteLog(">>> 请输入主题");

return;

}

//在 MQTT 中有三种 QoS 级别：

//At most once(0) 最多一次

//At least once(1) 至少一次

//Exactly once(2) 恰好一次

//await mqttClient.SubscribeAsync(new TopicFilterBuilder().WithTopic(this.tbTopic.Text).WithAtMostOnceQoS().Build());//最多一次， QoS 级别0

await mqttClient.SubscribeAsync(new TopicFilterBuilder().WithTopic(this.txtTopic.Text).WithAtLeastOnceQoS().Build());//恰好一次， QoS 级别1

WriteLog($">>> 成功订阅");

}

///

/// 发布

///

private async void btnPub_Click(object sender, EventArgs e)

{

if (string.IsNullOrWhiteSpace(this.txtTopik.Text))

{

WriteLog(">>> 请输入主题");

return;

}

var result = await mqttClient.PublishAsync(

this.txtTopik.Text,

this.txtContent.Text,

MQTTnet.Protocol.MqttQualityOfServiceLevel.AtLeastOnce);//恰好一次， QoS 级别1

WriteLog($">>> 主题：{this.txtTopik.Text}，消息：{this.txtContent.Text}，结果： {result.ReasonCode}");

}

4、运行测试

生成编译解决方案，成功后开始测试

1、启动服务器

2、启动客户端

找到生成的客户端debug目录下的.exe文件

3、测试连接

连接成功，服务器看到客户端上线了

4、测试订阅

再运行一个客户端，连接服务器

5、测试发布

c1向cced主题发布一个消息，结果是c1,c2都收到了消息

同样，c2发布一个消息，c1,c2都收到了消息

6、测试下线

c1关闭，服务器马上知道了

5、小结

基于mqttnet的组件搭建的mqtt服务器和客户端通信成功，发布和订阅都ko,ko,ko。

讲解不易，分析不易，原创不易，整理不易，伙伴们动动你的金手指，你的支持是我最大的动力。

好文链接

评论可见，请评论后查看内容，谢谢！！！评论后请刷新页面。

本文由用户于 2024-06-04 发布在夸智网，如有疑问，请联系我们。
本文链接：https://www.kuazhi.com/post/714025796.html

夸智网

MQTTNET mqtt客户端 mqtt发布 mqtt订阅 .net C#MQTT编程06--MQTT服务器和客户端(winform版)

大数据医疗中使用的大数据方法（大数据医疗中使用的大数据方法有哪些）

开发语言 C#打印控件ReportViewer使用总结（四）

发表评论取消回复

夸智网

MQTTNET mqtt客户端 mqtt发布 mqtt订阅 .net C#MQTT编程06--MQTT服务器和客户端(winform版)

大数据医疗中使用的大数据方法（大数据医疗中使用的大数据方法有哪些）

开发语言 C#打印控件ReportViewer使用总结（四）

相关文章

发表评论取消回复