#|MQTTnet 实现MQTT 客户端和服务端 #|协议相关

服务端：

class Program { private static MqttServer mqttServer = null; static void Main(string[] args) { MqttNetTrace.TraceMessagePublished += MqttNetTrace_TraceMessagePublished; new Thread(StartMqttServer).Start(); while (true) { var inputString = Console.ReadLine().ToLower().Trim(); if (inputString == "exit") { mqttServer?.StopAsync(); Console.WriteLine("MQTT服务已停止！"); break; } 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 { var options = new MqttServerOptions { ConnectionValidator = p => { if (p.ClientId == "c001") { if (p.Username != "u001" || p.Password != "p001") { return MqttConnectReturnCode.ConnectionRefusedBadUsernameOrPassword; } }return MqttConnectReturnCode.ConnectionAccepted; } }; mqttServer = new MqttServerFactory().CreateMqttServer(options) as MqttServer; mqttServer.ApplicationMessageReceived += MqttServer_ApplicationMessageReceived; mqttServer.ClientConnected += MqttServer_ClientConnected; mqttServer.ClientDisconnected += MqttServer_ClientDisconnected; } catch (Exception ex) { Console.WriteLine(ex.Message); return; } }mqttServer.StartAsync(); Console.WriteLine("MQTT服务启动成功！"); }private static void MqttServer_ClientConnected(object sender, MqttClientConnectedEventArgs e) { Console.WriteLine($"客户端[{e.Client.ClientId}]已连接，协议版本：{e.Client.ProtocolVersion}"); }private static void MqttServer_ClientDisconnected(object sender, MqttClientDisconnectedEventArgs e) { Console.WriteLine($"客户端[{e.Client.ClientId}]已断开连接！"); }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}"); }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); }*/ } }

客户端

class Program { private static MqttClient mqttClient = null; static string topic = "测试1/测试2/测试3"; static void Main(string[] args) {mqttClient = new MqttClientFactory().CreateMqttClient() as MqttClient; mqttClient.ApplicationMessageReceived += MqttClient_ApplicationMessageReceived; mqttClient.Connected += MqttClient_Connected; mqttClient.Disconnected += MqttClient_Disconnected; var options = new MqttClientTcpOptions { Server = "127.0.0.1", ClientId = Guid.NewGuid().ToString().Substring(0, 5), UserName = "u001", Password = "p001", CleanSession = true }; mqttClient.ConnectAsync(options); Console.ReadLine(); Console.WriteLine("Hello World!"); } private static void MqttClient_ApplicationMessageReceived(object sender, MqttApplicationMessageReceivedEventArgs e) {Console.WriteLine($">> {Encoding.UTF8.GetString(e.ApplicationMessage.Payload)}{Environment.NewLine}"); } private static void MqttClient_Connected(object sender, EventArgs e) {Console.WriteLine("已连接到MQTT服务器！" + Environment.NewLine); mqttClient.SubscribeAsync(new List { new TopicFilter(topic, MqttQualityOfServiceLevel.AtMostOnce) }); Console.WriteLine($"已订阅[{topic}]主题" + Environment.NewLine); //开始发布消息 string inputString = "你好么"; var appMsg = new MqttApplicationMessage(topic, Encoding.UTF8.GetBytes(inputString), MqttQualityOfServiceLevel.AtMostOnce, false); mqttClient.PublishAsync(appMsg); } private static void MqttClient_Disconnected(object sender, EventArgs e) {Console.WriteLine("已断开MQTT连接！" + Environment.NewLine); } }

文章图片

文章图片

【#|MQTTnet 实现MQTT 客户端和服务端】源码下载地址：MQTT
MQTT Windows 端验证程序： http://mqttfx.jensd.de/index.php/download
https://download.csdn.net/download/kesshei/10906614