一、服务端

首先，我们先来看下服务端的功能实现，主要类文件为MyMqttServer.cs，在服务端，当用户链接之后，有个用户信息验证的过程，该过程可以用，也可以不用，在连接验证器里面有个名称为MqttConnectReasonCode的状态信息，MqttConnectReasonCode的状态说明如下：

public enum MqttConnectReasonCode
{Success = 0,                       // 成功UnspecifiedError = 128,            // 未知的错误MalformedPacket = 129,             // 数据包缺失ProtocolError = 130,               // 协议有误ImplementationSpecificError = 131, // 实现细节有误UnsupportedProtocolVersion = 132,  // 协议版本不支持ClientIdentifierNotValid = 133,    // 客户端标识码无效BadUserNameOrPassword = 134,       // 错误的用户名或密码NotAuthorized = 135,               // 未授权ServerUnavailable = 136,           // 服务器不可用ServerBusy = 137,                  // 服务器正忙Banned = 138,                      // 已禁用BadAuthenticationMethod = 140,     // 验证方法有误TopicNameInvalid = 144,            // 无效的Topic名称PacketTooLarge = 149,              // 数据包过大QuotaExceeded = 151,               // 超出配额PayloadFormatInvalid = 153,        // Payload格式有误RetainNotSupported = 154,          // 不支持保留QoSNotSupported = 155,             // 不支持QoSUseAnotherServer = 156,            // 使用其他服务器ServerMoved = 157,                 // 服务器移动ConnectionRateExceeded = 159       // 连接速率超出限定
}

MyMqttServer.cs 代码如下：

using MQTTnet;
using MQTTnet.Protocol;
using MQTTnet.Server;
using System;
using System.Collections.Generic;
using System.Text;
namespace Mqtt_Server
{/// <summary>/// 消息调用的事件/// </summary>/// <param name="sender"></param>public delegate void InformHandle(object sender);public class MyMqttServer{#region 消息/// <summary>/// 返回的消息/// </summary>public event InformHandle MqttMessage;/// <summary>/// 发送消息/// </summary>/// <param name="data"></param>public virtual void ToSignal(object data){if (MqttMessage != null) MqttMessage(data);}#endregion#region 环境变量/// <summary>/// mqtt服务/// </summary>private MqttServer mqttServer { get; set; }/// <summary>/// 配置信息/// </summary>public MqttConfigs Config { get; set; }#endregionpublic bool StartServer(){try{var options = new MqttServerOptionsBuilder();// 连接记录数，默认 一般为2000options.WithConnectionBacklog(2000);// 默认端口是1883,这里可以自己设置options.WithDefaultEndpointPort(Config.tcpPort);// 连接验证器options.WithConnectionValidator((e) => ConnectionValidationHandler(e));// 持续会话options.WithPersistentSessions();// 会话超时时间？options.WithDefaultCommunicationTimeout(TimeSpan.FromMilliseconds(60000));// 每个客户端主题存1000000条数据options.WithMaxPendingMessagesPerClient(1000000);var mqttFactory = new MqttFactory();mqttServer = mqttFactory.CreateMqttServer() as MqttServer;// 服务启动消息事件mqttServer.StartedHandler = new MqttServerStartedHandlerDelegate(OnStartedHandler);// 服务停止消息事件mqttServer.StoppedHandler = new MqttServerStoppedHandlerDelegate(OnStoppedHandler);// 客户端连接事件mqttServer.UseClientConnectedHandler(UseClientConnected);// 客户端断开事件mqttServer.UseClientDisconnectedHandler(UseClientDisconnected);// 客户端消息事件mqttServer.UseApplicationMessageReceivedHandler(UseApplicationMessageReceived);// 开启订阅事件mqttServer.ClientSubscribedTopicHandler = new MqttServerClientSubscribedTopicHandlerDelegate(OnClientSubscribedTopic);// 取消订阅事件mqttServer.ClientUnsubscribedTopicHandler = new MqttServerClientUnsubscribedTopicHandlerDelegate(OnClientUnsubscribedTopic);// 启动服务mqttServer.StartAsync(options.Build());return true;}catch (Exception ex){ToSignal(new MqttSignal() { Type = 0, Data = $"客户端尝试连接出错：{ex.Message}" });}return false;}/// <summary>/// 连接验证器，进行用户名和密码的验证，也可以不验证/// </summary>/// <returns></returns>private void ConnectionValidationHandler(MqttConnectionValidatorContext e){if (e.Username == Config.mqttUser && e.Password == Config.mqttPassword){e.ReasonCode = MqttConnectReasonCode.Success;ToSignal($"{e.ClientId} 登录成功");return;}else{e.ReasonCode = MqttConnectReasonCode.BadUserNameOrPassword;ToSignal($"{e.ClientId} 用户名或密码错误，登陆失败");return;}}/// <summary>/// 服务启动消息事件/// </summary>/// <param name="e"></param>private void OnStartedHandler(EventArgs e){ToSignal("服务启动消息事件");}/// <summary>/// 服务停止消息事件/// </summary>/// <param name="e"></param>private void OnStoppedHandler(EventArgs e){ToSignal("服务停止消息事件");}/// <summary>/// 客户端连接事件/// </summary>/// <param name="e"></param>private void UseClientConnected(MqttServerClientConnectedEventArgs e){ToSignal($"客户端[{e.ClientId}]已连接");}/// <summary>/// 客户端断开事件/// </summary>/// <param name="e"></param>private void UseClientDisconnected(MqttServerClientDisconnectedEventArgs e){ToSignal($"客户端[{e.ClientId}]断开 Type:{e.DisconnectType}");}/// <summary>/// 客户端消息事件/// </summary>/// <param name="e"></param>private void UseApplicationMessageReceived(MqttApplicationMessageReceivedEventArgs e){var msg = e.ApplicationMessage;var topic = msg.Topic;var payload = msg.ConvertPayloadToString();ToSignal($"客户端消息事件:\r\nTopic:{topic}\r\nPayload:{payload}");}/// <summary>/// 开启订阅事件/// </summary>/// <param name="e"></param>private void OnClientSubscribedTopic(MqttServerClientSubscribedTopicEventArgs e){// 向前端汇报客户端端ID和主题ToSignal($"开启订阅事件:{e.ClientId}\r\nTopic:{e.TopicFilter.ToString()}");ToSignal(new MqttSignal() { Type = 3, Data = new List<string> { e.ClientId, e.TopicFilter.Topic } });}/// <summary>/// 取消订阅事件/// </summary>/// <param name="e"></param>private void OnClientUnsubscribedTopic(MqttServerClientUnsubscribedTopicEventArgs e){// 向前端汇报客户端端ID和主题ToSignal($"取消订阅事件:{e.ClientId} Topic:{e.TopicFilter}");ToSignal(new MqttSignal() { Type = 4, Data = new List<string> { e.ClientId, e.TopicFilter } });}/// <summary>/// 发送消息/// </summary>/// <param name="topic">客户端主题</param>/// <param name="msg">消息</param>public bool Send(string topic, object msg){try{if (msg.GetType() == typeof(string) && msg.GetType() == typeof(byte[])){ToSignal(new MqttSignal() { Type = 0, Data = "消息格式错误" });return false;}var message = new MqttApplicationMessage(){Topic = topic, // 客户端主题Payload = (msg.GetType() == typeof(string)) ? Encoding.UTF8.GetBytes(msg as string) : msg as byte[], // 发布的消息QualityOfServiceLevel = (MqttQualityOfServiceLevel)0, // 消息等级Retain = false  // 是否保留};mqttServer.PublishAsync(message);ToSignal(new MqttSignal() { Type = 1, Data = string.Format("服务端[{0}]发布主题[{1}]成功！", mqttServer.Options.ClientId, topic) });return true;}catch (Exception ex){ToSignal(new MqttSignal() { Type = 0, Data = string.Format("服务端[{0}]发布主题[{1}]异常！>{2}", mqttServer.Options.ClientId, topic, ex.Message) });}return false;}public void MqttClose(){if (mqttServer == null) return;try{mqttServer?.StopAsync();mqttServer = null;ToSignal($"服务关闭成功.");}catch (Exception e){ToSignal($"服务关闭失败：{e.Message}");}}}#region 反馈信号/// <summary>/// 反馈信号/// </summary>public class MqttSignal{/// <summary>/// 类型/// <para>0：异常消息</para>/// <para>1：日志消息</para>/// <para>2：接收到数据</para>/// <para>3：链接消息</para>/// <para>4：断开消息</para>/// </summary>public int Type { get; set; }/// <summary>/// 数据/// </summary>public object Data { get; set; }}#endregion#region 配置信息/// <summary>/// 配置信息/// </summary>public class MqttConfigs{/// <summary>/// 链接端口/// </summary>public int tcpPort;/// <summary>/// 链接用户名/// </summary>public string mqttUser;/// <summary>/// 链接密码/// </summary>public string mqttPassword;};#endregion
}

Form1的代码如下：

using System;
using System.Collections.Generic;
using System.Threading.Tasks;
using System.Windows.Forms;
namespace Mqtt_Server
{public partial class Form1 : Form{/// <summary>/// Mqtt服务/// </summary>public MyMqttServer mqtt { get; set; }public Form1(){InitializeComponent();comboBox2.Items.Add("全部");comboBox2.SelectedIndex = 0;}private void button_启动服务_Click(object sender, EventArgs e){new Task(() =>{if (mqtt == null){// 参数配置MqttConfigs con = new MqttConfigs();con.tcpPort = int.Parse(textBox3.Text);con.mqttUser = textBox5.Text;con.mqttPassword = textBox6.Text;// 启动Mqtt服务mqtt = new MyMqttServer();mqtt.Config = con;mqtt.MqttMessage += Mqtt_MqttMessage; ;// 订阅消息mqtt.StartServer();}}).Start();}private void button_关闭服务_Click(object sender, EventArgs e){mqtt.MqttClose();mqtt = null;}private void button_清空日志_Click(object sender, EventArgs e){textBox1.Text = "";}private void button_发送消息_Click(object sender, EventArgs e){if (comboBox2.Text == "全部"){foreach (string s in comboBox2.Items){mqtt.Send(s, textBox2.Text);}}else{mqtt.Send(comboBox2.Text, textBox2.Text);}}/// <summary>/// mqtt消息回调/// </summary>/// <param name="sender"></param>private void Mqtt_MqttMessage(object sender){BeginInvoke(new Action(() =>{if (sender.GetType() == typeof(MqttSignal)){MqttSignal m = sender as MqttSignal;switch (m.Type){case 3:List<string> data1 = m.Data as List<string>;// 添加IDif (!comboBox1.Items.Contains(data1[0]))comboBox1.Items.Add(data1[0]);comboBox1.SelectedIndex = comboBox1.Items.Count - 1;// 添加主题if (!comboBox2.Items.Contains(data1[1]))comboBox2.Items.Add(data1[1]);comboBox2.SelectedIndex = comboBox2.Items.Count - 1;break;case 4:List<string> data2 = m.Data as List<string>;// 删除IDcomboBox1.Items.Remove(data2[0]);comboBox1.SelectedIndex = comboBox1.Items.Count - 1;// 删除主题comboBox2.Items.Remove(data2[1]);comboBox2.SelectedIndex = comboBox2.Items.Count - 1;break;default:textBox1.Text += m.Data.ToString() + "\r\n";break;}}if (sender.GetType() == typeof(string)){string m = sender as string;textBox1.Text += m + "\r\n";}}));}}
}

二、客户端

客户端的功能实现，主要类文件为MyMqttClient.cs，代码如下：

using MQTTnet;
using MQTTnet.Client;
using MQTTnet.Client.Connecting;
using MQTTnet.Client.Disconnecting;
using MQTTnet.Client.Options;
using MQTTnet.Client.Receiving;
using MQTTnet.Protocol;
using MQTTnet.Server;
using System;
using System.Text;
using System.Threading;
namespace Mqtt_Client
{/// <summary>/// 消息调用的事件/// </summary>/// <param name="sender"></param>public delegate void InformHandle(object sender);public class MyMqttClient{#region 消息/// <summary>/// 返回的消息/// </summary>public event InformHandle MqttMessage;/// <summary>/// 发送消息/// </summary>/// <param name="data"></param>public virtual void ToSignal(object data){if (MqttMessage != null) MqttMessage(data);}#endregion#region 环境变量/// <summary>/// mqtt连接信息/// </summary>public MqttClient mqttClient { get; set; }/// <summary>/// 配置信息/// </summary>public MqttConfigs Config { get; set; }#endregion#region 私有方法/// <summary>/// 时间差计算/// </summary>/// <param name="tim"></param>/// <returns></returns>private long TimeSecondDif(DateTime tim){return (DateTime.Now.Ticks - tim.Ticks) / 10000000;}#endregion#region 主体功能/// <summary>/// 启动连接/// </summary>/// <param name="conf"></param>/// <returns></returns>public bool StartClient(){try{var options = new MqttClientOptions{ClientId = Config.ClientId,ProtocolVersion = MQTTnet.Formatter.MqttProtocolVersion.V311,ChannelOptions = new MqttClientTcpOptions { Server = Config.tcpServer, Port = Config.tcpPort },WillDelayInterval = 10 // 延迟的事件};if (options.ChannelOptions == null){throw new InvalidOperationException();}options.Credentials = new MqttClientCredentials{Username = Config.mqttUser,Password = Encoding.UTF8.GetBytes(Config.mqttPassword)};options.CleanSession = true;options.KeepAlivePeriod = TimeSpan.FromSeconds(120);var mqttFactory = new MqttFactory();mqttClient = mqttFactory.CreateMqttClient() as MqttClient;// 连接成功回调mqttClient.ConnectedHandler = new MqttClientConnectedHandlerDelegate(OnMqttClientConnected);// 连接断开回调mqttClient.DisconnectedHandler = new MqttClientDisconnectedHandlerDelegate(OnMqttClientDisConnected);// 客户端接收回调mqttClient.ApplicationMessageReceivedHandler = new MqttApplicationMessageReceivedHandlerDelegate(OnSubscriberMessageReceived);mqttClient.ConnectAsync(options);ToSignal(new MqttSignal() { Type = 1, Data = $"客户端[{options.ClientId}]尝试连接..." });DateTime OutTime = DateTime.Now;while (mqttClient != null && !mqttClient.IsConnected){// 30秒链接超时if (TimeSecondDif(OutTime) >= 30) return false;Thread.Sleep(500);}Thread.Sleep(100);if (mqttClient == null){ToSignal(new MqttSignal() { Type = 0, Data = $"客户端连接超时..." });return false;}ToSignal(new MqttSignal() { Type = 1, Data = $"订阅消息..." });ClientSubscribeTopic(Config.topic);return true;}catch (Exception ex){ToSignal(new MqttSignal() { Type = 0, Data = $"客户端尝试连接出错：{ex.Message}" });}return false;}/// <summary>/// 关闭连接/// </summary>/// <returns></returns>public void ClientStop(){try{if (mqttClient == null){return;}mqttClient.DisconnectAsync();mqttClient = null;}catch (Exception ex){ToSignal(new MqttSignal() { Type = 0, Data = $"客户端尝试断开Server出错：{ex.Message}" });}}/// <summary>/// 连接成功回调/// </summary>/// <param name="e"></param>private void OnMqttClientConnected(MqttClientConnectedEventArgs e){ToSignal(new MqttSignal() { Type = 1, Data = $"客户端连接成功." });}/// <summary>/// 连接断开回调/// </summary>/// <param name="e"></param>private void OnMqttClientDisConnected(MqttClientDisconnectedEventArgs e){ToSignal(new MqttSignal() { Type = 1, Data = $"客户端断开连接." });ClientStop();}/// <summary>/// 订阅消息/// </summary>/// <param name="topic"></param>private void ClientSubscribeTopic(string topic){try{if (mqttClient == null) return;mqttClient.SubscribeAsync(topic);ToSignal(new MqttSignal() { Type = 1, Data = string.Format("客户端[{0}]订阅主题[{1}]成功！", mqttClient.Options.ClientId, topic) });}catch (Exception ex){ToSignal(new MqttSignal() { Type = 0, Data = $"客户端订阅主题出错：{ex.Message}" });}}/// <summary>/// 取消订阅消息/// </summary>/// <param name="topic"></param>public void MqttClose(string topic){try{if (mqttClient == null) return;mqttClient.UnsubscribeAsync(topic);ToSignal(new MqttSignal() { Type = 1, Data = string.Format("客户端[{0}]取消主题[{1}]成功！", mqttClient.Options.ClientId, topic) });}catch (Exception ex){ToSignal(new MqttSignal() { Type = 0, Data = $"客户端取消主题出错：{ex.Message}" });}}/// <summary>/// 客户端接收回调/// </summary>/// <param name="e"></param>private void OnSubscriberMessageReceived(MqttApplicationMessageReceivedEventArgs e){try{string json = Encoding.UTF8.GetString(e.ApplicationMessage.Payload);string Topic = e.ApplicationMessage.Topic;string QoS = e.ApplicationMessage.QualityOfServiceLevel.ToString();string Retained = e.ApplicationMessage.Retain.ToString();ToSignal(new MqttSignal() { Type = 2, Data = $"[接收] >> Topic:{Topic} QoS:{QoS} Retained:{Retained} json:{json}" });}catch (Exception ex){ToSignal(new MqttSignal() { Type = 0, Data = $"数据接收异常：{ex.Message}" });}}/// <summary>/// 发送消息/// </summary>/// <param name="topic">客户端主题</param>/// <param name="msg">消息</param>public bool Send(string topic, object msg){try{if (mqttClient == null) return false;if (msg.GetType() == typeof(string) && msg.GetType() == typeof(byte[])){ToSignal(new MqttSignal() { Type = 0, Data = "消息格式错误" });return false;}var message = new MqttApplicationMessage(){Topic = topic, // 客户端主题Payload = (msg.GetType() == typeof(string)) ? Encoding.UTF8.GetBytes(msg as string) : msg as byte[], // 发布的消息QualityOfServiceLevel = (MqttQualityOfServiceLevel)0, // 消息等级Retain = false  // 是否保留};mqttClient.PublishAsync(message);ToSignal(new MqttSignal() { Type = 1, Data = string.Format("客户端[{0}]发布主题[{1}]成功！", mqttClient.Options.ClientId, topic) });return true;}catch (Exception ex){ToSignal(new MqttSignal() { Type = 0, Data = string.Format("客户端[{0}]发布主题[{1}]异常！>{2}", mqttClient.Options.ClientId, topic, ex.Message) });}return false;}#endregion}#region 反馈信号/// <summary>/// 反馈信号/// </summary>public class MqttSignal{/// <summary>/// 类型/// <para>0：异常消息</para>/// <para>1：日志消息</para>/// <para>2：接收到数据</para>/// </summary>public int Type { get; set; }/// <summary>/// 数据/// </summary>public string Data { get; set; }}#endregion#region 配置信息// 参考配置// tcpPort = 1883;// tcpServer = "192.168.1.151";// mqttUser = "rakwireless";// mqttPassword = "rakwireless.com";// ClientId = "MQTT_FX_Client";// topic = "application/+/device/+/rx";/// <summary>/// 配置信息/// </summary>public class MqttConfigs{/// <summary>/// 链接端口/// </summary>public int tcpPort;/// <summary>/// 目标IP/// </summary>public string tcpServer;/// <summary>/// 链接用户名/// </summary>public string mqttUser;/// <summary>/// 链接密码/// </summary>public string mqttPassword;/// <summary>/// 客户端ID/// </summary>public string ClientId;/// <summary>/// 主题/// </summary>public string topic;};#endregion
}

Form1的代码如下：

using System;
using System.Threading.Tasks;
using System.Windows.Forms;
namespace Mqtt_Client
{public partial class Form1 : Form{/// <summary>/// Mqtt/// </summary>public MyMqttClient mqtt { get; set; }public Form1(){InitializeComponent();}private void button_启动链接_Click(object sender, EventArgs e){new Task(() =>{if (mqtt == null || mqtt.mqttClient == null || !mqtt.mqttClient.IsConnected){// 参数配置MqttConfigs con = new MqttConfigs();con.tcpPort = int.Parse(textBox3.Text);con.tcpServer = textBox4.Text;con.mqttUser = textBox5.Text;con.mqttPassword = textBox6.Text;con.ClientId = textBox7.Text;con.topic = textBox8.Text;// 启动Mqttmqtt = new MyMqttClient();mqtt.Config = con;mqtt.MqttMessage += Mqtt_MqttMessage;// 订阅消息mqtt.StartClient();}}).Start();}private void button_关闭链接_Click(object sender, EventArgs e){mqtt.MqttClose(textBox8.Text);mqtt = null;}private void button_清空消息_Click(object sender, EventArgs e){textBox1.Text = "";}private void button_发送消息_Click(object sender, EventArgs e){mqtt.Send(textBox8.Text, textBox2.Text);}/// <summary>/// Mqtt消息回调/// </summary>/// <param name="sender"></param>private void Mqtt_MqttMessage(object sender){BeginInvoke(new Action(() =>{if (sender.GetType() == typeof(MqttSignal)){MqttSignal m = sender as MqttSignal;textBox1.Text += m.Data + "\r\n";}}));}}
}

使用MQTTNet包实现客户端与服务端通讯相关推荐

php cannot bind port to socket,PHP基于socket实现客户端和服务端通讯功能
本文主要介绍了PHP基于socket实现的简单客户端和服务端通讯功能,可实现服务端接收客户端发送的字符串进行翻转操作后返回客户端的功能,需要的朋友可以参考下服务端: set_time_limit(0 ...
基于Socket实现客户端与服务端通讯
基于Socket实现客户端与服务端通讯 socket 概述 Socket,套接字就是两台主机之间逻辑连接的端点.TCP/IP协议是传输层协议,主要解决数据如何在网络中传输,而HTTP是应用层协议,主 ...
springboot实现SSE服务端主动向客户端推送数据，java服务端向客户端推送数据，kotlin模拟客户端向服务端推送数据
SSE服务端推送服务器向浏览器推送信息,除了 WebSocket,还有一种方法:Server-Sent Events(以下简称 SSE).本文介绍它的用法. 在很多业务场景中,会涉及到服务端向客户端 ...
RPC 笔记（03）— gRPC 概念、安装、编译、客户端和服务端示例
1. gRPC 概念 gRPC 是 Google 开源的一款高性能的 RPC 框架.GitHub 上介绍如下: gRPC is a modern, open source, high-performa ...
java socket同步通信,javasocket客户端与服务端同步通信实例
javasocket客户端与服务端同步通信实例工作中没涉及有关JA V A SOCKET编程的实际经历,但理论大概了解.想正明一下自已的对JA V A SOCKET理解写个通信TEST,通过TEST ...
一篇文章带你了解https是如何做到客户端与服务端之间安全通信
https是什么. 超文本传输安全协议(英语:Hypertext Transfer Protocol Secure,缩写:HTTPS,常称为HTTP over TLS,HTTP over SSL或HT ...
AIDL 客户端与服务端的双向通信
时隔一年半了,终于写下了这个续篇,我发现我的很多博客有头无尾,都是有前面一点点,后面就没写去了,也正在想办法都补上初涉IPC,了解AIDL的工作原理及使用方法今天聊聊的是客户端和服务端的相互通信, ...
SignalR 实现web浏览器客户端与服务端的推送功能
SignalR 是一个集成的客户端与服务器库,基于浏览器的客户端和基于 ASP.NET 的服务器组件可以借助它来进行双向多步对话. 换句话说,该对话可不受限制地进行单个无状态请求/响应数据交换:它将继 ...
Netty实战 IM即时通讯系统（十二）构建客户端与服务端pipeline
Netty实战 IM即时通讯系统(十二)构建客户端与服务端pipeline 零. 目录 IM系统简介 Netty 简介 Netty 环境配置服务端启动流程客户端启动流程实战: 客户端和服务端双向 ...

使用MQTTNet包实现客户端与服务端通讯

一、服务端

二、客户端

使用MQTTNet包实现客户端与服务端通讯相关推荐

最新文章

热门文章