MQTTNet 介绍

大家好，欢迎来到IT知识分享网。

MQTTnet 是一个高性能 .NET 库，用于基于 MQTT 的通信。 它提供了一个 MQTT Client和一个 MQTT Server（代理）。

MQTT是一个基于客户端-服务器的消息发布/订阅传输协议。MQTT协议是轻量、简单、开放和易于实现的，这些特点使它适用范围非常广泛。

从上图可以看出, MQTT主要包含两部分, 服务端负责接受客户端的订阅, 以及向客户端推送消息, 而客户端只需要负责连接服务器以及订阅消息。

MQTTNet主要包含两个部分, 一个是MQTTServer, 另一个则是MQTTClient。

MQTTServer

通过 .NET CLI 安装 MQTTnet

dotnet add package MQTTnet --version 3.1.2

典型的声明一个MQTTNet Server只需要简单的几行代码即可, 以下代码显示了使用 TCP 端点创建新 MQTT Server的最简单方法，该端点在默认端口 1883 上进行侦听。

var mqttServer = new MqttFactory().CreateMqttServer();
await mqttServer.StartAsync(new MqttServerOptions());

也可以通过 MqttServerOptionsBuilder 配置指定的端口以及其他设置, 例如客户端连接验证, 是否接受该连接:

 			var optionBuilder = new MqttServerOptionsBuilder()  
                .WithDefaultEndpointPort(1884)  
                .WithConnectionValidator(c =>
                 {
                    if (c.ClientId.Length < 10)
    				{
        				c.ReasonCode = MqttConnectReasonCode.ClientIdentifierNotValid;
        				return;
    				}

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

    				if (c.Password != "mySecretPassword")
    				{
       					 c.ReasonCode = MqttConnectReasonCode.BadUserNameOrPassword;
       				     return;
   				  }
    			  c.ReasonCode = MqttConnectReasonCode.Success;
                });
 
            server = new MqttFactory().CreateMqttServer();  
            await server.StartAsync(optionBuilder.Build());

拦截应用程序消息

通过配置 WithApplicationMessageInterceptor可以拦截由客户端发向服务器的消息, 在这里, 可以处理消息以及拦截消息。

var optionsBuilder = new MqttServerOptionsBuilder()
    .WithApplicationMessageInterceptor(context =>
    { 
        if (context.ClientId != "Someone")
        {
            context.AcceptPublish = false;
            return;
        } 
        //...
    })
    .Build();

拦截订阅

使用订阅拦截器可以设置自定义拦截器来控制 MQTT 客户端可以订阅哪些主题。

var optionsBuilder = new MqttServerOptionsBuilder()
    .WithSubscriptionInterceptor(context =>
    {
        if (context.TopicFilter.Topic.StartsWith("admin/foo/bar") && context.ClientId != "theAdmin")
        {
            context.AcceptSubscription = false;
        }

        if (context.TopicFilter.Topic.StartsWith("the/secret/stuff") && context.ClientId != "Imperator")
        {
            context.AcceptSubscription = false;
            context.CloseConnection = true;
        }
    })
    .Build();

向客户端发送消息

			 server.PublishAsync(new MqttApplicationMessage()
                {
                    Topic = "mytopic",
                    QualityOfServiceLevel = 			MQTTnet.Protocol.MqttQualityOfServiceLevel.ExactlyOnce,
                    Retain = false,
                    Payload = Encoding.UTF8.GetBytes("这是服务器消息！")
                }); ;

MQTTClient

典型的声明一个MQTTNet Client同样只需要简单的几行代码即可。

通常创建一个MQTTClient需要5个设置, 指定服务器的IP以及端口, 客户端ID 以及用户名密码。

            var options = new MqttClientOptionsBuilder()
                .WithTcpServer("127.0.0.1", 1883)
                .WithClientId(Guid.NewGuid().ToString())
                .WithCredentials("username", "password").Build();
            client = new MqttFactory().CreateMqttClient();  
            await client.ConnectAsync(options);

订阅主题消息

通俗的来讲, 该功能主要用于订阅服务端的消息, 当服务器有消息时, 客户端将允许被接收, 不同的主题消息格式内容也可能不同, 这取决于服务器的具体定义。

await client.SubscribeAsync("mytopic");

接收消息事件

接收消息则是单独的方法进行接收, 通过配置 UseApplicationMessageReceivedHandler

client.UseApplicationMessageReceivedHandler(c =>
            { 
                string message = Encoding.UTF8.GetString(c.ApplicationMessage.Payload);
                //...
            })

客户端发送消息

连接成功后, 客户端可以向服务器发送指定的主题消息

			var msg = new MqttApplicationMessageBuilder()
                .WithTopic("Test")
                .WithPayload("Hello World")
                .WithExactlyOnceQoS()
                .WithRetainFlag().Build();
            client.PublishAsync(msg);

通常来讲, Topic(主题)是由服务端来决定, 对于每个消息都有几个级别, 通过枚举MqttQualityOfServiceLevel 定义。

ManagedClient

无论是对于Client还是Server层, 它们都有各自的其它配置, 包括一些基本的设置, 如连接成功事件、断开事件等等。

• 对于Client而言, MQTTNet还提供了一个标准的封装库, 它包含了一些常用的功能, 开箱即用。

• 托管客户端启动一次，将自动保持连接，包括重新连接等。

• 所有 MQTT 应用程序消息都被添加到内部队列中，并在服务器可用时进行处理。
  可以存储所有 MQTT 应用程序消息，以支持在应用程序重启后发送它们

• 所有订阅都通过服务器连接进行管理。 与服务器断开连接后无需手动订阅。

以下代码显示了如何设置和启动托管 MQTT 客户端。

var options = new ManagedMqttClientOptionsBuilder()
    .WithAutoReconnectDelay(TimeSpan.FromSeconds(5))
    .WithClientOptions(new MqttClientOptionsBuilder()
        .WithClientId("clientId")
        .WithTcpServer("127.0.0.1")
        .WithTls().Build())
    .Build();

var mqttClient = new MqttFactory().CreateManagedMqttClient();
await mqttClient.SubscribeAsync(new MqttTopicFilterBuilder().WithTopic("my/topic").Build());
await mqttClient.StartAsync(options);