MQTTServer服务器根据MQTTClient客户端已订阅的主题推送分发消息

网络读卡器介绍：https://item.taobao.com/item.htm?ft=t&id=22173428704&spm=a21dvs.23580594.0.0.52de2c1bgK3bgZ

本示例使用了MQTTNet插件

C# MQTTNETServer 源码

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 MqttServerOptionsBuilder optionBuilder;private IMqttServer server;//mqtt服务器对象List<TopicItem> Topics = new List<TopicItem>();public Form1(){InitializeComponent();}private void Form1_Load(object sender, EventArgs e){//创建服务器对象server = new MqttFactory().CreateMqttServer();server.ApplicationMessageReceivedHandler = new MqttApplicationMessageReceivedHandlerDelegate(new Action<MqttApplicationMessageReceivedEventArgs>(Server_ApplicationMessageReceived));//绑定消息接收事件server.ClientConnectedHandler = new MqttServerClientConnectedHandlerDelegate(new Action<MqttServerClientConnectedEventArgs>(Server_ClientConnected));//绑定客户端连接事件server.ClientDisconnectedHandler = new MqttServerClientDisconnectedHandlerDelegate(new Action<MqttServerClientDisconnectedEventArgs>(Server_ClientDisconnected));//绑定客户端断开事件server.ClientSubscribedTopicHandler = new MqttServerClientSubscribedHandlerDelegate(new Action<MqttServerClientSubscribedTopicEventArgs>(Server_ClientSubscribedTopic));//绑定客户端订阅主题事件server.ClientUnsubscribedTopicHandler = new MqttServerClientUnsubscribedTopicHandlerDelegate(new Action<MqttServerClientUnsubscribedTopicEventArgs>(Server_ClientUnsubscribedTopic));//绑定客户端退订主题事件server.StartedHandler = new MqttServerStartedHandlerDelegate(new Action<EventArgs>(Server_Started));//绑定服务端启动事件server.StoppedHandler = new MqttServerStoppedHandlerDelegate(new Action<EventArgs>(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(">>> 服务端已启动！");Invoke(new Action(() => {this.button1.Text = "停止服务";}));}/// 绑定服务端停止事件private void Server_Stopped(EventArgs e){WriteLog(">>> 服务端已停止！");Invoke(new Action(() => {this.button1.Text = "启动MQTT服务";}));}/// 显示日志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 button1_Click(object sender, EventArgs e){if (button1.Text == "启动MQTT服务")   /// 启动服务{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);}else{if (server != null)   //停止服务{                    server.StopAsync();}}}}
}

C# MQTTNETClient 源码

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;
using static System.Windows.Forms.Design.AxImporter;
using System.Net.Security;namespace MQTTNETClientForms
{public partial class Form1 : Form{private MqttFactory factory;private IManagedMqttClient mqttClient;//客户端mqtt对象private MqttClientOptionsBuilder mqttClientOptions;private ManagedMqttClientOptionsBuilder options;private bool connstate;public Form1(){InitializeComponent();}private void Form1_Load(object sender, EventArgs e){}/// 显示日志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 (connstate == false){WriteLog(">>> 请先与MQTT服务器建立连接！");return;}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 级别0await mqttClient.SubscribeAsync(new TopicFilterBuilder().WithTopic(this.txtTopic.Text).WithAtLeastOnceQoS().Build());//恰好一次， QoS 级别1 WriteLog($">>> 成功订阅");}/// 发布private async void btnPub_Click(object sender, EventArgs e){if (connstate==false){WriteLog(">>> 请先与MQTT服务器建立连接！");return;}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}");}private async void button1_Click(object sender, EventArgs e){if (button1.Text == "连接到MQTT服务器"){connstate = false;factory = new MqttFactory();mqttClient = factory.CreateManagedMqttClient();//创建客户端对象//绑定断开事件mqttClient.UseDisconnectedHandler(async ee =>{                   WriteLog("与服务器之间的连接断开了，正在尝试重新连接");Invoke(new Action(() => {this.button1.Text = "连接到MQTT服务器";}));// 等待 5s 时间await Task.Delay(TimeSpan.FromSeconds(5));try{if (factory == null) { factory = new MqttFactory() ; }//创建客户端对象 if (mqttClient == null) { mqttClient = factory.CreateManagedMqttClient(); }//创建客户端对象 mqttClient.UseConnectedHandler(tt =>{connstate = true;WriteLog(">>> 连接到服务成功");Invoke(new Action(() => {this.button1.Text = "断开与MQTT服务器的连续";}));});}catch (Exception ex){connstate = false;WriteLog($"重新连接服务器失败:{ex}");Invoke(new Action(() => {this.button1.Text = "连接到MQTT服务器";}));}});//绑定接收事件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 =>{connstate =true;WriteLog(">>> 连接到服务成功");Invoke(new Action(() => {this.button1.Text = "断开与MQTT服务器的连续";}));});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);                }else{if (mqttClient != null){if (mqttClient.IsStarted){await mqttClient.StopAsync();}mqttClient.Dispose();connstate = false;}button1.Text = "连接到MQTT服务器";}}}
}