繁体

C#使用MQTT通信

白茶清欢

这里简单说一下，MQTT的功能就是一个发布/订阅的功能

例如，接下来，我们做一个例子。现在有A服务器作为发布方，B客户端作为订阅方。如果A发布了消息，那么会向B发送消息，反之亦然，从而进行通信的功能。

服务端代码

建立一个.net5的web api项目

nuget引用mqttnet

Snipaste_2024-07-16_11-45-28.png

MqttService代码

using MQTTnet;
using MQTTnet.Protocol;
using MQTTnet.Server;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
 
namespace mqtt
{
    public class MqttService
    {
        public static MqttServer _mqttServer { get; set; }
 
        public static void PublishData(string data)
        {
            var message = new MqttApplicationMessage
            {
                Topic = "topic1",
                Payload = Encoding.Default.GetBytes(data),
                QualityOfServiceLevel = MqttQualityOfServiceLevel.AtLeastOnce,
                Retain = true  // 服务端是否保留消息。true为保留，如果有新的订阅者连接，就会立马收到该消息。
            };
 
            _mqttServer.InjectApplicationMessage(new InjectedMqttApplicationMessage(message) // 发送消息给有订阅 topic_01的客户端
            {
                SenderClientId = "Server_01"
            }).GetAwaiter().GetResult();
        }
 
    }
}

MqttHostService代码

using Microsoft.Extensions.Hosting;
using MQTTnet;
using MQTTnet.Protocol;
using MQTTnet.Server;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Net;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
 
namespace mqtt
{
    public class MqttHostService : IHostedService, IDisposable
    {
        public void Dispose()
        {
 
        }
        const string ServerClientId = "SERVER";
        public Task StartAsync(CancellationToken cancellationToken)
        {
            MqttServerOptionsBuilder optionsBuilder = new MqttServerOptionsBuilder();
            optionsBuilder.WithDefaultEndpoint();
            //optionsBuilder.WithDefaultEndpointBoundIPAddress(IPAddress.Parse("127.0.0.1"));
            optionsBuilder.WithDefaultEndpointPort(10086); // 设置 服务端 端口号
            optionsBuilder.WithConnectionBacklog(1000); // 最大连接数
            MqttServerOptions options = optionsBuilder.Build();
 
            MqttService._mqttServer = new MqttFactory().CreateMqttServer(options);
 
            MqttService._mqttServer.ClientConnectedAsync += _mqttServer_ClientConnectedAsync; //客户端连接事件
            MqttService._mqttServer.ClientDisconnectedAsync += _mqttServer_ClientDisconnectedAsync; // 客户端关闭事件
            MqttService._mqttServer.ApplicationMessageNotConsumedAsync += _mqttServer_ApplicationMessageNotConsumedAsync; // 消息接收事件
 
            MqttService._mqttServer.ClientSubscribedTopicAsync += _mqttServer_ClientSubscribedTopicAsync; // 客户端订阅主题事件
            MqttService._mqttServer.ClientUnsubscribedTopicAsync += _mqttServer_ClientUnsubscribedTopicAsync; // 客户端取消订阅事件
            MqttService._mqttServer.StartedAsync += _mqttServer_StartedAsync; // 启动后事件
            MqttService._mqttServer.StoppedAsync += _mqttServer_StoppedAsync; // 关闭后事件
            MqttService._mqttServer.InterceptingPublishAsync += _mqttServer_InterceptingPublishAsync; // 消息接收事件
            MqttService._mqttServer.ValidatingConnectionAsync += _mqttServer_ValidatingConnectionAsync; // 用户名和密码验证有关
 
            MqttService._mqttServer.StartAsync();
            return Task.CompletedTask;
        }
 
        /// <summary>
        /// 客户端订阅主题事件
        /// </summary>
        /// <param name="arg"></param>
        /// <returns></returns>
        private Task _mqttServer_ClientSubscribedTopicAsync(ClientSubscribedTopicEventArgs arg)
        {
            Console.WriteLine($"ClientSubscribedTopicAsync：客户端ID=【{arg.ClientId}】订阅的主题=【{arg.TopicFilter}】 ");
            return Task.CompletedTask;
        }
 
        /// <summary>
        /// 关闭后事件
        /// </summary>
        /// <param name="arg"></param>
        /// <returns></returns>
        private Task _mqttServer_StoppedAsync(EventArgs arg)
        {
            Console.WriteLine($"StoppedAsync：MQTT服务已关闭……");
            return Task.CompletedTask;
        }
 
        /// <summary>
        /// 用户名和密码验证有关
        /// </summary>
        /// <param name="arg"></param>
        /// <returns></returns>
        private Task _mqttServer_ValidatingConnectionAsync(ValidatingConnectionEventArgs arg)
        {
            arg.ReasonCode = MqttConnectReasonCode.Success;
            if ((arg.Username ?? string.Empty) != "admin" || (arg.Password ?? String.Empty) != "123456")
            {
                arg.ReasonCode = MqttConnectReasonCode.Banned;
                Console.WriteLine($"ValidatingConnectionAsync：客户端ID=【{arg.ClientId}】用户名或密码验证错误 ");
 
            }
            return Task.CompletedTask;
        }
 
        /// <summary>
        /// 消息接收事件
        /// </summary>
        /// <param name="arg"></param>
        /// <returns></returns>
        private Task _mqttServer_InterceptingPublishAsync(InterceptingPublishEventArgs arg)
        {
            if (string.Equals(arg.ClientId, ServerClientId))
            {
                return Task.CompletedTask;
            }
 
            Console.WriteLine($"InterceptingPublishAsync：客户端ID=【{arg.ClientId}】 Topic主题=【{arg.ApplicationMessage.Topic}】 消息=【{Encoding.UTF8.GetString(arg.ApplicationMessage.Payload)}】 qos等级=【{arg.ApplicationMessage.QualityOfServiceLevel}】");
            return Task.CompletedTask;
 
        }
 
        /// <summary>
        /// 启动后事件
        /// </summary>
        /// <param name="arg"></param>
        /// <returns></returns>
        private Task _mqttServer_StartedAsync(EventArgs arg)
        {
            Console.WriteLine($"StartedAsync：MQTT服务已启动……");
            return Task.CompletedTask;
        }
 
        /// <summary>
        /// 客户端取消订阅事件
        /// </summary>
        /// <param name="arg"></param>
        /// <returns></returns>
        private Task _mqttServer_ClientUnsubscribedTopicAsync(ClientUnsubscribedTopicEventArgs arg)
        {
            Console.WriteLine($"ClientUnsubscribedTopicAsync：客户端ID=【{arg.ClientId}】已取消订阅的主题=【{arg.TopicFilter}】  ");
            return Task.CompletedTask;
        }
 
        private Task _mqttServer_ApplicationMessageNotConsumedAsync(ApplicationMessageNotConsumedEventArgs arg)
        {
            Console.WriteLine($"ApplicationMessageNotConsumedAsync：发送端ID=【{arg.SenderId}】 Topic主题=【{arg.ApplicationMessage.Topic}】 消息=【{Encoding.UTF8.GetString(arg.ApplicationMessage.Payload)}】 qos等级=【{arg.ApplicationMessage.QualityOfServiceLevel}】");
            return Task.CompletedTask;
 
        }
 
        /// <summary>
        /// 客户端断开时候触发
        /// </summary>
        /// <param name="arg"></param>
        /// <returns></returns>
        /// <exception cref="NotImplementedException"></exception>
        private Task _mqttServer_ClientDisconnectedAsync(ClientDisconnectedEventArgs arg)
        {
            Console.WriteLine($"ClientDisconnectedAsync：客户端ID=【{arg.ClientId}】已断开, 地址=【{arg.Endpoint}】  ");
            return Task.CompletedTask;
 
        }
 
        /// <summary>
        /// 客户端连接时候触发
        /// </summary>
        /// <param name="arg"></param>
        /// <returns></returns>
        private Task _mqttServer_ClientConnectedAsync(ClientConnectedEventArgs arg)
        {
            Console.WriteLine($"ClientConnectedAsync：客户端ID=【{arg.ClientId}】已连接, 用户名=【{arg.UserName}】地址=【{arg.Endpoint}】  ");
            return Task.CompletedTask;
        }
 
        public Task StopAsync(CancellationToken cancellationToken)
        {
            return Task.CompletedTask;
        }
    }
}

在Startup注入代码

Snipaste_2024-07-16_11-57-51.png

 services.AddHostedService<MqttHostService>();

增加一个服务器发送消息的方法

Snipaste_2024-07-16_11-58-40.png

  public IActionResult Test(string data)
  {
      MqttService.PublishData(data);
      return Ok();
  }

客户端代码

建立一个.net5  web api项目

nuget引用mqttnet

Snipaste_2024-07-16_12-00-44.png

MqttClientService代码

using MQTTnet;
using MQTTnet.Client;
using MQTTnet.Protocol;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
 
namespace mqttClient
{
    public class MqttClientService
    {
        public static IMqttClient _mqttClient;
        public void MqttClientStart()
        {
            var optionsBuilder = new MqttClientOptionsBuilder()
                .WithTcpServer("127.0.0.1", 10086) // 要访问的mqtt服务端的 ip 和 端口号
                .WithCredentials("admin", "123456") // 要访问的mqtt服务端的用户名和密码
                .WithClientId("testclient02") // 设置客户端id
                .WithCleanSession()
                .WithTls(new MqttClientOptionsBuilderTlsParameters
                {
                    UseTls = false  // 是否使用 tls加密
                });
 
            var clientOptions = optionsBuilder.Build();
            _mqttClient = new MqttFactory().CreateMqttClient();
 
            _mqttClient.ConnectedAsync += _mqttClient_ConnectedAsync; // 客户端连接成功事件
            _mqttClient.DisconnectedAsync += _mqttClient_DisconnectedAsync; // 客户端连接关闭事件
            _mqttClient.ApplicationMessageReceivedAsync += _mqttClient_ApplicationMessageReceivedAsync; // 收到消息事件
 
            _mqttClient.ConnectAsync(clientOptions);
 
 
        }
 
        /// <summary>
        /// 客户端连接关闭事件
        /// </summary>
        /// <param name="arg"></param>
        /// <returns></returns>
        private Task _mqttClient_DisconnectedAsync(MqttClientDisconnectedEventArgs arg)
        {
            Console.WriteLine($"客户端已断开与服务端的连接……");
            return Task.CompletedTask;
        }
 
        /// <summary>
        /// 客户端连接成功事件
        /// </summary>
        /// <param name="arg"></param>
        /// <returns></returns>
        private Task _mqttClient_ConnectedAsync(MqttClientConnectedEventArgs arg)
        {
            Console.WriteLine($"客户端已连接服务端……");
 
            // 订阅消息主题
            // MqttQualityOfServiceLevel: （QoS）:  0 最多一次，接收者不确认收到消息，并且消息不被发送者存储和重新发送提供与底层 TCP 协议相同的保证。
            // 1: 保证一条消息至少有一次会传递给接收方。发送方存储消息，直到它从接收方收到确认收到消息的数据包。一条消息可以多次发送或传递。
            // 2: 保证每条消息仅由预期的收件人接收一次。级别2是最安全和最慢的服务质量级别，保证由发送方和接收方之间的至少两个请求/响应（四次握手）。
            _mqttClient.SubscribeAsync("topic1", MqttQualityOfServiceLevel.AtLeastOnce); //topic_02
 
            return Task.CompletedTask;
        }
 
        /// <summary>
        /// 收到消息事件
        /// </summary>
        /// <param name="arg"></param>
        /// <returns></returns>
        private Task _mqttClient_ApplicationMessageReceivedAsync(MqttApplicationMessageReceivedEventArgs arg)
        {
            Console.WriteLine($"ApplicationMessageReceivedAsync：客户端ID=【{arg.ClientId}】接收到消息。 Topic主题=【{arg.ApplicationMessage.Topic}】 消息=【{Encoding.UTF8.GetString(arg.ApplicationMessage.Payload)}】 qos等级=【{arg.ApplicationMessage.QualityOfServiceLevel}】");
            return Task.CompletedTask;
        }
 
        public void Publish(string data)
        {
            var message = new MqttApplicationMessage
            {
                Topic = "topic2",
                Payload = Encoding.Default.GetBytes(data),
                QualityOfServiceLevel = MqttQualityOfServiceLevel.AtLeastOnce,
                Retain = true  // 服务端是否保留消息。true为保留，如果有新的订阅者连接，就会立马收到该消息。
            };
            _mqttClient.PublishAsync(message);
        }
    }
}

启动的时候，实例化调用，并且发送消息456

Snipaste_2024-07-16_12-01-48.png

效果

同时启动2个项目

Snipaste_2024-07-16_12-02-41.png

在服务端的api中输入发送的消息

Snipaste_2024-07-16_12-03-21.png可以看到B客户端收到了消息

Snipaste_2024-07-16_12-03-36.png

我在B客户端中点击回车，就把刚才说的456发送给了A服务端。

Snipaste_2024-07-16_12-03-49.png

注意：一个发布，一个订阅，发布的Topic和订阅的Topic一定要是同一个，就如同打电话一样，双方要在同一条线上。

如果运行不了，就换一台电脑，可能是系统组件的问题，我就遇到了这个问题。

拓展

实际上，服务器的代码可以不用写，我们只用写客户端的代码就行了。客户端控制发布和订阅的Topic即可，服务器只是一个转换的作用。

服务器安装

目前MQTT代理的主流平台有下面几个


我们选择第一个进行下载，直接安装

Snipaste_2024-07-16_12-07-19.png安装完成后，进入文件夹，按住Shift，右键鼠标点击空白处，然后打开Powershell，正常打开一个终端软件即可。

Snipaste_2024-07-16_12-07-29.png

输入./mosquitto.exe -h 可以查看相应的帮助；

输入./mosquitto.exe -p 10086，就开启了MQTT服务，监听的地址是127.0.0.1，端口是10086；

Snipaste_2024-07-16_12-07-38.png

客户端依然使用上面的代码

因为我们要进行2个客户端直接通信，那么就要修改对应发布和订阅的Topic 

客户端1

Snipaste_2024-07-16_12-15-52.png

修改3个地方，生产一个EXE 

客户端2 

Snipaste_2024-07-16_12-19-17.png

修改3个地方，生产一个EXE 。

分别生产了2个文件夹的EXE

Snipaste_2024-07-16_12-20-28.png

分别运行，效果，可以看到2个客户端互相给对方发送了456的消息。

Snipaste_2024-07-16_12-20-37.png

免责声明：本文来自故里2130，不代表白茶清欢的个人技术分享的观点和立场，如有侵权请联系本平台处理。

相关阅读

  • C#使用MQTT通信
  • 菱声音响文字转语音开发ls20

    • 发表评论

    快捷回复： 表情：
    AddoilApplauseBadlaughBombCoffeeFabulousFacepalmFecesFrownHeyhaInsidiousKeepFightingNoProbPigHeadShockedSinistersmileSlapSocialSweatTolaughWatermelonWittyWowYeahYellowdog
    评论列表 （暂无评论，710人围观）

    还没有评论，来说两句吧...