基于MQTTnet 3.0.12实现MQTT服务器和客户端「建议收藏」

2022-09-12 20:26:09 浏览数 (1)

大家好，又见面了，我是你们的朋友全栈君。

基于MQTTnet 3.0.12实现MQTT服务器和客户端)

概述-可看可不看的废话

现在物联网的概念很火，辣条从业7年来一直号称是“自动化工程师”，但其实到目前为止，所处的行业还是比较局限在自动化产线改造。就工业智能互联来说，水平方向上如何把流水线式的孤岛式机台联动起来，我们比较有经验，但是垂直方向上，如何做数采，或者说如何高效灵活的做数采，需要补课的东西还有很多。MQTT是IBM很早以前就提出来的协议，但很可惜一直没有接触过，新公司的项目上引用了MQTTnet的开源库，但是这个GitHub作者跳版跳得太狠了点，随着新版本的更新，改了很多旧版的东西，导致百度排在前几的文章都没法直接借鉴了，无奈之下，疯狂百度 阅读Git上源码，总算搞出来一个小的Demo，特此记录如下。

MQTT是啥

MQTT 是物联网 （IoT） 的 OASIS 标准消息传递协议。它设计为极其轻量级的发布/订阅消息传输，非常适合连接具有小代码占用空间和最小网络带宽的远程设备。如今，MQTT 广泛应用于汽车、制造、电信、石油和天然气等行业。

好了，这段话是我抄的MQTT官网的，英语原文，浏览器直译，傲娇。 官网指路：https://mqtt.org/

MQTTnet

MQTTnet是基于MQTT通信的高性能.NET库，它提供了一个MQTT客户端和一个MQTT服务器（代理）。截止目前，最新版本为3.0.12.0，支持.net core，支持MQTT 3.X和5.0版本。

https://github.com/chkr1011/MQTTnet MQTTnet的Git路径。

正文

本Demo设计为一个Winform窗体程序。基于MQTTnet，实现了一个MQTT Server或者说Broker的创建，同时在窗体上提供了MQTT Client的创建功能，MQTT Client跟Server连接之后，通过点击按钮，实现主题订阅、发布的基础功能。

MQTT Server/Broker实现

创建MQTT Server的思路还是蛮清晰的，主要是MqttServer各个事件的实现，一开始很懵逼，因为3.0.12的写法变化跟2.X完全不一样。 这里贴一段MQTTnet的代码，方便理解类的实现——

MQTT Server/Broker创建

  • 先定义好MqttServerOptions，这是启动Mqtt服务时候的传参，定义服务启动时的各种参数：IP、Port口，账密等等
代码语言：javascript复制
var optionsBuilder = new MqttServerOptionsBuilder();
  • 实例化MqttServer以及委托实现MqttServer的各个事件
代码语言：javascript复制
mqttServer = new MqttFactory().CreateMqttServer() as MqttServer;
mqttServer.StartedHandler = new MqttServerStartedHandlerDelegate(OnMqttServerStarted);
mqttServer.StoppedHandler = new MqttServerStoppedHandlerDelegate(OnMqttServerStopped);

下面是完成的实现代码：

代码语言：javascript复制
public MqttServer mqttServer = null;
public async void StartMqttServer()
{ 
   
    try
    { 
   
        if (mqttServer == null)
        { 
   
            var config = ReadConfiguration();
            var optionsBuilder = new MqttServerOptionsBuilder()
            .WithDefaultEndpoint().WithDefaultEndpointPort(int.Parse(config["Port"].ToString())).WithConnectionValidator(
                c =>
                        { 
   
                            var currentUser = config["Users"][0]["UserName"].ToString();
                            var currentPWD = config["Users"][0]["Password"].ToString();

                            if (currentUser == null || currentPWD == null)
                            { 
   
                                c.ReasonCode = MqttConnectReasonCode.BadUserNameOrPassword;
                                return;
                            }

                            if (c.Username != currentUser)
                            { 
   
                                c.ReasonCode = MqttConnectReasonCode.BadUserNameOrPassword;
                                return;
                            }

                            if (c.Password != currentPWD)
                            { 
   
                                c.ReasonCode = MqttConnectReasonCode.BadUserNameOrPassword;
                                return;
                            }

                            c.ReasonCode = MqttConnectReasonCode.Success;
                        }).WithSubscriptionInterceptor(
                c =>
                { 
   
                    c.AcceptSubscription = true;
                }).WithApplicationMessageInterceptor(
                c =>
                { 
   
                    c.AcceptPublish = true;
                });

            mqttServer = new MqttFactory().CreateMqttServer() as MqttServer;
            mqttServer.StartedHandler = new MqttServerStartedHandlerDelegate(OnMqttServerStarted);
            mqttServer.StoppedHandler = new MqttServerStoppedHandlerDelegate(OnMqttServerStopped);

            mqttServer.ClientConnectedHandler = new MqttServerClientConnectedHandlerDelegate(OnMqttServerClientConnected);
            mqttServer.ClientDisconnectedHandler = new MqttServerClientDisconnectedHandlerDelegate(OnMqttServerClientDisconnected);
            mqttServer.ClientSubscribedTopicHandler = new MqttServerClientSubscribedHandlerDelegate(OnMqttServerClientSubscribedTopic);
            mqttServer.ClientUnsubscribedTopicHandler = new MqttServerClientUnsubscribedTopicHandlerDelegate(OnMqttServerClientUnsubscribedTopic);
            mqttServer.ApplicationMessageReceivedHandler = new MqttApplicationMessageReceivedHandlerDelegate(OnMqttServer_ApplicationMessageReceived);


            await mqttServer.StartAsync(optionsBuilder.Build());
            lbxMonitor.BeginInvoke(_updateMonitorAction,
                Logger.TraceLog(Logger.Level.Info, "MQTT Server is started."));

        }
    }
    catch (Exception ex)
    { 
   
        lbxMonitor.BeginInvoke(_updateMonitorAction,
                    Logger.TraceLog(Logger.Level.Fatal, $"MQTT Server start fail.>{ex.Message}"));
    }
}

public async void StopMqttServer()
{ 
   
    if (mqttServer == null) return;
    try
    { 
   
        await mqttServer?.StopAsync();
        mqttServer = null;
        lbxMonitor.BeginInvoke(_updateMonitorAction,
                Logger.TraceLog(Logger.Level.Info, "MQTT Server is stopped."));
    }
    catch (Exception ex)
    { 
   
        lbxMonitor.BeginInvoke(_updateMonitorAction,
                    Logger.TraceLog(Logger.Level.Fatal, $"MQTT Server stop fail.>{ex.Message}"));
    }
}

MQTT Server/Broker发布消息

-从MQTT的设计来看，服务端是代理的角色，订阅者和发布者是客户端，所以通常来说，消息的订阅与发布应当都是客户端干的事。但是，服务端自然也是可以参与一下发布者的角色的。我还没想到这样用的实际场景，先功能实现一下——

  • 很粗暴，先实例化一个MqttApplicationMessage对象，然后作为传参调用MqttServer.PublishAsync进行消息发布。
代码语言：javascript复制
public async void ServerPublishMqttTopic(string topic, string payload)
{ 
   
    var message = new MqttApplicationMessage()
    { 
   
        Topic = topic,
        Payload = Encoding.UTF8.GetBytes(payload)
    };
    await mqttServer.PublishAsync(message);
    lbxMonitor.BeginInvoke(_updateMonitorAction,
            Logger.TraceLog(Logger.Level.Info, string.Format("MQTT Broker发布主题[{0}]成功！", topic)));
}

当然啦，MqttApplicationMessage还有很多属性值可以传，QoS(MqttQualityOfServiceLevel)、Retain等等，这里只是为了先实现功能，就只传了最简单的Topic和Payload。

MQTT Client实现

MQTT Client创建

整体的实现思路跟Server端如出一辙，声明一个MqttClientOptions，赋值各种连接Server端需要的参数，最后作为MqttClient.ConnectAsync的传参，连接Server。 这里我在MqttClientOptions里面尝试了一下WillMessage，其实就是一个MqttApplicationMessage对象，WillMessage作为遗言机制，用于Client跟Server端挂点时的广播通知。 再之后就是实现MqttClient的各个事件，用来客制Connected，Disconnected，MessageReceived的各种逻辑，跟Server端实现没有什么区别，不再赘述。 上代码——

代码语言：javascript复制
private MqttClient mqttClient = null;
private async Task ClientStart()
{ 
   
    try
    { 
   
        var tcpServer = txtIPAddr.Text;
        var tcpPort = int.Parse(txtPort.Text.Trim());
        var mqttUser = txtUserName.Text.Trim();
        var mqttPassword = txtPWD.Text.Trim();

        var mqttFactory = new MqttFactory();



        var options = new MqttClientOptions
        { 
   
            ClientId = txtClientID.Text.Trim(),
            ProtocolVersion = MQTTnet.Formatter.MqttProtocolVersion.V311,
            ChannelOptions = new MqttClientTcpOptions
                    { 
   
                        Server = tcpServer,
                        Port = tcpPort
                    },
            WillDelayInterval = 10,
            WillMessage = new MqttApplicationMessage()
            { 
   
                Topic = $"LastWill/{txtClientID.Text.Trim()}",
                Payload= Encoding.UTF8.GetBytes("I Lost the connection!"),
                QualityOfServiceLevel = MqttQualityOfServiceLevel.ExactlyOnce
            }
            
        };
        if (options.ChannelOptions == null)
        { 
   
            throw new InvalidOperationException();
        }

        if (!string.IsNullOrEmpty(mqttUser))
        { 
   
            options.Credentials = new MqttClientCredentials
            { 
   
                Username = mqttUser,
                Password = Encoding.UTF8.GetBytes(mqttPassword)
            };
        }

        options.CleanSession = true;
        options.KeepAlivePeriod = TimeSpan.FromSeconds(5);

        mqttClient = mqttFactory.CreateMqttClient() as MqttClient;
        mqttClient.ConnectedHandler = new MqttClientConnectedHandlerDelegate(OnMqttClientConnected);
        mqttClient.DisconnectedHandler = new MqttClientDisconnectedHandlerDelegate(OnMqttClientDisConnected);
        mqttClient.ApplicationMessageReceivedHandler = new MqttApplicationMessageReceivedHandlerDelegate(OnSubscriberMessageReceived);
        await mqttClient.ConnectAsync(options);
        lbxMonitor.BeginInvoke(_updateMonitorAction,
                Logger.TraceLog(Logger.Level.Info, $"客户端[{options.ClientId}]尝试连接..."));
    }
    catch (Exception ex)
    { 
   
        lbxMonitor.BeginInvoke(_updateMonitorAction,
                Logger.TraceLog(Logger.Level.Fatal, $"客户端尝试连接出错.>{ex.Message}"));
    }
}

private async Task ClientStop()
{ 
   
    try
    { 
   
        if (mqttClient == null) return;
        await mqttClient.DisconnectAsync();
        mqttClient = null;
    }
    catch (Exception ex)
    { 
   
        lbxMonitor.BeginInvoke(_updateMonitorAction,
                Logger.TraceLog(Logger.Level.Fatal, $"客户端尝试断开Server出错.>{ex.Message}"));
    }
}

MQTT Client发布消息

这里的实现逻辑跟写法和Server端的发布别无二致，我在这里的MqttApplicationMessage补上了QoS和Retain的设置，由Form页面的控件传参。 这里补一句关于Retain的用法：Retain意为保留，设为True表示这条消息发布的时候如果没有订阅者，则该消息保留在Server端，直到被人订阅时立刻发布出去并删除，设为False时则没有这样的效果。

代码语言：javascript复制
public async void ClientPublishMqttTopic(string topic, string payload)
{ 
   
    try
    { 
   
        var message = new MqttApplicationMessage()
        { 
   
            Topic = topic,
            Payload = Encoding.UTF8.GetBytes(payload),
            QualityOfServiceLevel = (MqttQualityOfServiceLevel)cmbQos.SelectedIndex,
            Retain = bool.Parse(cmbRetain.SelectedItem.ToString())
        };
        await mqttClient.PublishAsync(message);
        lbxMonitor.BeginInvoke(_updateMonitorAction,
                Logger.TraceLog(Logger.Level.Info, string.Format("客户端[{0}]发布主题[{1}]成功！", mqttClient.Options.ClientId, topic)));
    }
    catch (Exception ex)
    { 
   
        lbxMonitor.BeginInvoke(_updateMonitorAction,
                    Logger.TraceLog(Logger.Level.Fatal, string.Format("客户端[{0}]发布主题[{1}]异常！>{2}", mqttClient.Options.ClientId, topic,ex.Message)));
    }
}

MQTT Clien订阅消息

呼叫MqttClient.SubscribeAsync，传入消息主题即可。

代码语言：javascript复制
public async void ClientSubscribeTopic(string topic)
{ 
   
    await mqttClient.SubscribeAsync(topic);
    lbxMonitor.BeginInvoke(_updateMonitorAction,
            Logger.TraceLog(Logger.Level.Info, string.Format("客户端[{0}]订阅主题[{1}]成功！", mqttClient.Options.ClientId, topic)));
}

MQTT Clien取消订阅消息

呼叫MqttClient.UnsubscribeAsync，取消消息订阅。

代码语言：javascript复制
public async void ClientUnSubscribeTopic(string topic)
{ 
   
    await mqttClient.UnsubscribeAsync(topic);
    lbxMonitor.BeginInvoke(_updateMonitorAction,
            Logger.TraceLog(Logger.Level.Info, string.Format("客户端[{0}]取消主题[{1}]成功！", mqttClient.Options.ClientId, topic)));
}

至此，简单的基于MQTTnet的Server端以及Client端的功能就实现咯，完整代码已传至GitHub，欢迎大家指点讨论，共同进步，一起成长！

发布者：全栈程序员栈长，转载请注明出处：https://javaforall.cn/152881.html原文链接：https://javaforall.cn

mqtt .net https javascript node.js

0 人点赞