理解MQTT协议与使用MQTTNet库进行高效开发

大家好，我是讯享网，很高兴认识大家。

引言

MQTT（Message Queuing Telemetry Transport）是一种轻量级的发布/订阅消息协议，专为低带宽、高延迟或不可靠网络环境下的物联网设备通信而设计。其基于TCP/IP协议栈，具有低功耗、低带宽占用和可靠传输等特点，非常适合于传感器、移动设备以及远程控制等应用场景。

MQTT协议基础

发布/订阅模式：MQTT使用发布/订阅模型，其中发布者发送消息，订阅者接收消息。这使得消息传递更加灵活，可以根据不同的需求设置多个订阅者。
轻量级：MQTT协议非常轻量，减少了网络流量和传输数据的大小。这使得它在低带宽、高延迟或不稳定的环境中表现良好。
主题（Topic）分发：消息通过主题进行分发，订阅者可以订阅感兴趣的主题，从而接收到相应的消息。主题的层次结构允许灵活的消息路由。
异步通信：MQTT支持异步通信，发布者和订阅者之间不需要建立持久连接。这提高了通信的效率和可用性。
QoS（Quality of Service）级别：MQTT提供了三种QoS级别，以满足不同应用的需求。QoS 0表示最多传送一次，QoS 1表示可靠传送，QoS 2表示确保传送。

MQTTNet库介绍

MQTTNet是一个高性能、易于使用的开源.NET库，用于实现MQTT协议。它提供了全面的API接口，支持.NET Standard，可以在多种.NET平台上运行，包括但不限于.NET Framework、.NET Core、UWP和Xamarin等。

MQTTNet库的主要功能特性包括：

完整MQTT协议支持：完全遵循MQTT 3.1.1和5.0版本规范，提供丰富的连接、发布、订阅、断开等功能。
灵活的配置选项：用户可以根据需求自定义连接参数，如保持连接、清理会话、服务质量等级等。
异步编程模型：利用C#的async/await机制，确保在处理大量并发连接和消息时仍能保持高效性能。
事件驱动机制：通过订阅各种事件（如连接成功、消息接收、客户端断开等），开发者可以轻松构建响应式的应用程序。

下面来使用MQTTNet写个例子

服务采用Aspnetcore webapi 方式启用，使用了tcp和websokcet连接

// 服务端 #region MQTT var mqttConfig = ConfigJson.GetAll<MqttConfig>("mqttconfig", env.EnvironmentName); //添加MQTT 服务启动 builder.WebHost.UseKestrel(options => { options.ListenAnyIP(mqttConfig.Port, l => l.UseMqtt()); options.ListenAnyIP(mqttConfig.WsPort); }); //注册成windows/linux服务 builder.Host.UseWindowsService(); builder.Host.UseSystemd(); //注册成单例 builder.Services.AddSingleton(_ => new MqttService(mqttConfig)); //MQTT builder.Services.AddHostedMqttServer(optionsBuilder => { optionsBuilder.WithDefaultEndpoint(); optionsBuilder.WithDefaultEndpointPort(mqttConfig.Port); // 设置 服务端 端口号 optionsBuilder.WithConnectionBacklog(mqttConfig.MaxConnection);// 最大连接数 }); builder.Services.AddMqttConnectionHandler(); builder.Services.AddConnections(); #endregion MQTT // websokcet app.UseEndpoints(endpoints => { endpoints.MapControllers(); endpoints.MapConnectionHandler<MqttConnectionHandler>( "/mqtt", httpConnectionDispatcherOptions => httpConnectionDispatcherOptions.WebSockets.SubProtocolSelector = protocolList => protocolList.FirstOrDefault() ?? string.Empty); }); app.UseMqttServer(server => { var mqttService = app.Services.GetRequiredService<MqttService>(); server.ClientConnectedAsync += mqttService.ClientConnectedAsync; //客户端连接事件 server.ClientDisconnectedAsync += mqttService.ClientDisconnectedAsync; // 客户端关闭事件 server.ApplicationMessageNotConsumedAsync += mqttService.ApplicationMessageNotConsumedAsync; // 消息接收事件 server.ClientSubscribedTopicAsync += mqttService.ClientSubscribedTopicAsync; // 客户端订阅主题事件 server.ClientUnsubscribedTopicAsync += mqttService.ClientUnsubscribedTopicAsync; // 客户端取消订阅事件 server.StartedAsync += mqttService.StartedAsync; // 启动后事件 server.StoppedAsync += mqttService.StoppedAsync; // 关闭后事件 server.InterceptingPublishAsync += mqttService.InterceptApplicationMessagePublishAsync; // 验证发布事件 server.ValidatingConnectionAsync += mqttService.ValidatingConnectionAsync; // 用户名和密码验证有关 server.InterceptingSubscriptionAsync += mqttService.InterceptSubscriptionAsync; //验证订阅事件 });

讯享网

讯享网// 客户端 using MQTTnet; using MQTTnet.Client; using System.Text; var mqttFactory = new MqttFactory(); var mqttClient = mqttFactory.CreateMqttClient(); // 配置TCP 连接 ，验证使用账户密码 var options = new MqttClientOptionsBuilder() .WithTcpServer("127.0.0.1", 1688) .WithClientId("Client_01") .WithCredentials("Client_01", "Client_01") .Build(); //连接 await mqttClient.ConnectAsync(options, CancellationToken.None); //接受消息 mqttClient.ApplicationMessageReceivedAsync += e => { Console.WriteLine(" 接受订阅消息 "); Console.WriteLine($"+ Topic = {e.ApplicationMessage.Topic}"); Console.WriteLine($"+ Payload = {Encoding.UTF8.GetString(e.ApplicationMessage.PayloadSegment)}"); Console.WriteLine($"+ QoS = {e.ApplicationMessage.QualityOfServiceLevel}"); Console.WriteLine($"+ Retain = {e.ApplicationMessage.Retain}"); Console.WriteLine(); return Task.CompletedTask; }; //订阅 var mqttSubscribeOptions = mqttFactory.CreateSubscribeOptionsBuilder() .WithTopicFilter(f => { f.WithTopic("Cient_01/ledMsg/v1/control"); }) .Build(); var response = await mqttClient.SubscribeAsync(mqttSubscribeOptions, CancellationToken.None); Console.WriteLine($"MQTT客户端订阅了主题....."); //发布 for (int i = 0; i < 10; i++) { await Task.Delay(1000); var applicationMessage = new MqttApplicationMessageBuilder() .WithTopic("Cient_01/ledMsg/v1/control") .WithPayload($"我发布了Cient_01/ledMsg/v1/control消息_{i}") .Build(); await mqttClient.PublishAsync(applicationMessage, CancellationToken.None); } Console.WriteLine();

上述代码，使用了aspnetcore webapi启动服务端，使用控制台程序作为测试客户端，效果图如下：

客户端接受订阅消息
讯享网
服务端接受处理消息