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服務器";}}}
}