MqttNet作为.Net平台下的一个基于MQTT通信的开源库, 地址: https://github.com/chkr1011/MQTTnet。 本章内容主要介绍MQTT服务器与客户端之间如何通信, 创建实例等下相关内容。

添加MQTT包

- 创建基于NetCore平台的WPF项目,通过NuGet安装Mqttnet
    1.方式一(Package Manager): Install-Package MQTTnet -Version 3.0.8
    2.方式二(Net CLI): dotnet add package MQTTnet --version 3.0.8
    3...
    等: 手动方式添加NuGet包, 搜索:MqttNet进行安装。

创建MQTT服务端:

    IMqttServer server;
    List<UserInstance> instances;
    /// <summary>
        /// 创建MQTT服务端
        /// </summary>
        private async void Init()
        {
            var optionBuilder = new MqttServerOptionsBuilder().
                //设置默认IP: 127.0.0.1 ，端口1883，启用连接验证器
                WithDefaultEndpoint().WithDefaultEndpointPort(1883).WithConnectionValidator(
                c =>
                {
                    //此处忽略验证,做简单为空验证处理进行测试
                    var flag = (c.Username != "" && c.Password != "") ? true : false;

                    if (!flag)
                    {
                        //返回错误码
                        c.ReasonCode = MQTTnet.Protocol.MqttConnectReasonCode.BadUserNameOrPassword;
                        return;
                    }
                    //设置代码为 Success
                    c.ReasonCode = MQTTnet.Protocol.MqttConnectReasonCode.Success;
                    instances.Add(new UserInstance()  //缓存到内存的List集合当中
                    {
                        clientId = c.ClientId,
                        userName = c.Username,
                        passWord = c.Password
                    });
                    Showlog($"{DateTime.Now}:账号:{c.Username}已订阅!\r\n");
                }).WithSubscriptionInterceptor(c => 
                {
                    //消息拦截器
                    if (c == null) return;
                    c.AcceptSubscription = true;
                    Showlog($"{DateTime.Now}:订阅者{c.ClientId}\r\n");
                }).WithApplicationMessageInterceptor(c =>
                {
                    //应用程序消息拦截器
                    if (c == null) return;
                    c.AcceptPublish = true;
                    string str = c.ApplicationMessage?.Payload == null ? null : Encoding.UTF8.GetString(c.ApplicationMessage?.Payload) + "\r\n";
                    Showlog($"{DateTime.Now}:{str}\r\n");
                })

                ;
            //创建MQTT服务端
            server = new MqttFactory().CreateMqttServer();
            //用户断开通知
            server.UseClientDisconnectedHandler(c =>
            {
                var use = instances.FirstOrDefault(t => t.clientId == c.ClientId);
                if (use != null)
                {
                    instances.Remove(use);
                    Showlog($"{DateTime.Now}:订阅者{use.userName}已退出\r\n");
                }
            });
            //启动服务
            await server.StartAsync(optionBuilder.Build());
        }

- 下代码为实体类对象, 用于缓存用户信息

    public class UserInstance
    {
        public string clientId { get; set; }
        public string userName { get; set; }
        public string passWord { get; set; }
    }

- 服务端发送消息至客户端方法

                server.PublishAsync(new MqttApplicationMessage()
                {
                    Topic = arg.clientId,  //client id
                    QualityOfServiceLevel = MQTTnet.Protocol.MqttQualityOfServiceLevel.ExactlyOnce, //只发送一次
                    Retain = false,
                    Payload = Encoding.UTF8.GetBytes($"{DateTime.Now}:服务器:明天都不要来上班了！")  //测试内容
                }); ;

创建MQTT客户端

- 定义clienId 用于模拟缓存用户的clientId,

         IMqttClient client;
         string clientId = "";
          /// <summary>
        /// 启动MQTT客户端
        /// </summary>
        private async void Init()
        {
            //模拟唯一ID
            clientId = Guid.NewGuid().ToString();
            //本地TCP连接, 127.0.0.1 端口 1883 , 测试情况下账号密码固定 admin  123
            var options = new MqttClientOptionsBuilder().WithTcpServer("127.0.0.1", 1883) 
                .WithClientId(clientId).WithCredentials("admin", "123").Build();
            //创建MQTT实例
            client = new MqttFactory().CreateMqttClient();
            //注册连接成功时间
            client.UseConnectedHandler(async c =>
            {
                //订阅服务端消息
                await client.SubscribeAsync(new TopicFilterBuilder().WithTopic(clientId).Build());
            }).UseDisconnectedHandler(c => //注册断开事件
            {
                Showlog(c.Exception.Message);
            }).UseApplicationMessageReceivedHandler(c => //接受消息事件
            {
                string str = Encoding.UTF8.GetString(c.ApplicationMessage.Payload);
                Showlog(str);
            });
            //启动客户端
            await client.ConnectAsync(options);
        }

- 客户端给服务端发送消息

     var msg = new MqttApplicationMessageBuilder().
                WithTopic(clientId). // client Id
                WithPayload($"{DateTime.Now}:Hello World") //测试消息内容
                .WithExactlyOnceQoS(). //消息模式
                WithRetainFlag().
                Build();;
            client.PublishAsync(msg); //发送消息

        private void Showlog(string str)
        {
            this.Dispatcher.Invoke(() =>
            {
                txtResult.Text += str;
            });
        }

程序运行效果

MqttNet 官方文档以及介绍

Demo示例 :https://github.com/chkr1011/MQTTnet/wiki
接口文档: https://github.com/chkr1011/MQTTnet/wiki/Server-and-client-method-documentation

WPF Core—MqttNet开源库通讯实践