IOT设备通讯，MQTT物联网协议，MQTTnet

一.IOT设备的特性

• 硬件能力差(存储能力基本只有几MB,CPU频率低连使用HTTP请求都很奢侈)
• 系统千差万别(Brillo,mbedOS,RIOT等)
• 如使用电池供电,电量消耗敏感
• 如果是小设备，设备基数大需要维持大量在线链接
• 网络情况不稳定,移动网络网络资费贵,需要尽量减少开销和稳定

在以上这样苛刻的场景下很多技术上常用在智能设备方案都望而却步,总结一下我们主要面对下面三个问题:

• socket.io,websocket? 不同的系统可能无法使用HTTP,设备资源可能使用HTTP都奢侈 。
• TCP/IP自定协议? 虽然不用在意系统,自定义报文怎么解决网络开销问题?
• 自主研发成本高,使用第三方IOT平台容易被技术或硬件绑定 。

二. MQTT为什么适合IOT场景

• MQTT（Message Queuing Telemetry Transport，消息队列遥测传输）是IBM开发的一个即时通讯协议，使用方式比较类似于队列软件比如RabbitMQ，使用发布/订阅的方式提供互相之间的通讯。
• MQTT是为在计算能力有限，且工作在低带宽、不可靠的网络的远程传感器和控制设备通讯而设计的协议。

MQTT主要特性：

• 该协议支持所有平台，几乎可以把所有联网物品和外部连接起来
• 有三种消息发布服务质量
  - “至多一次”，消息发布完全依赖底层 TCP/IP 网络。会发生消息丢失或重复。这一级别可用于如下情况，环境传感器数据，丢失一次读记录无所谓，因为不久后还会有第二次发送。
  - “至少一次”，确保消息到达，但消息重复可能会发生。
  - “只有一次”，确保消息到达一次。这一级别可用于如下情况，在计费系统中，消息重复或丢失会导致不正确的结果。
• 小型传输，开销很小（固定长度的头部是 2 字节），协议交换最小化，以降低网络流量；
• 使用 Last Will 和 Testament 特性通知有关各方客户端异常中断的机制；

除了MQTT的协议特性外还有一些客观原因:

• 对语言友好主流语言的客户端都有
• 大部分硬件方案天生支持
• 数十个MQTT服务器端程序可供选择
• 社区成熟解决方案被广泛运用遇到问题方便寻求帮助

以上基本是我们为什么也会选择MQTT作为IOT协议的原因,需要更多的了解或者查看客户端支不支持和服务端实现可以参考官方github:

• MQTT官方 : https://github.com/mqtt/mqtt.github.io
• MQTT协议中文版：https://github.com/mcxiaoke/mqtt
• 服务中间件列表: https://github.com/mqtt/mqtt.github.io/wiki/servers
• 客户端列表: https://github.com/mqtt/mqtt.github.io/wiki/libraries
• 客户端简单Demo可以见官方文档：https://github.com/chkr1011/MQTTnet/wiki/Client

三、MQTTnet

MQTTnet 是一个基于 MQTT 通信的高性能 .NET 开源库，它同时支持 MQTT 服务器端和客户端。而且作者也保持更新，目前支持新版的.NET core，这也是选择 MQTTnet 的原因。

MQTTnet 在 Github 还有 MqttDotNet、nMQTT、M2MQTT 等。

在解决方案在右键单击-选择“管理解决方案的 NuGet 程序包”-在“浏览”选项卡下面搜索 MQTTnet，为服务端项目和客户端项目都安装上 MQTTnet 库。

四、 服务端

MQTT 服务端主要用于与多个客户端保持连接，并处理客户端的发布和订阅等逻辑。一般很少直接从服务端发送消息给客户端（可以使用 mqttServer.Publish(appMsg); 直接发送消息），多数情况下服务端都是转发主题匹配的客户端消息，在系统中起到一个中介的作用。

1、 创建服务端并启动

IMqttServer  mqttServer = new MqttFactory().CreateMqttServer();

通过上述方式创建了一个 IMqttServer 对象后，调用其 StartAsync 方法即可启动 MQTT 服务。值得注意的是：之前版本采用的是 Start 方法，作者也是紧跟 C# 语言新特性，能使用异步的地方也都改为异步方式。

Task.Run(async () => { await mqttServer.StartAsync(optionsBuilder.Build()); });

2、 配置设置、验证客户端

WithDefaultEndpointPort是设置使用的端口，协议里默认是用1883，不过调试我改成8222了。
WithConnectionValidator是用于连接验证，验证client id，用户名，密码什么的。示例没用数据库，随便写死了两个值。

还有其他配置选项，比如加密协议，可以在官方文档里看看，示例就是先简单能用。

var optionsBuilder = new MqttServerOptionsBuilder()
    .WithConnectionBacklog(100)
    .WithDefaultEndpointPort(8222)
    .WithConnectionValidator(ValidatingMqttClients())
    ;

private static Action<MqttConnectionValidatorContext> ValidatingMqttClients()
        {
            // Setup client validator.    
            var options =new MqttServerOptions();
            options.ConnectionValidator = c =>
            {
                Dictionary<string, string> c_u = new Dictionary<string, string>();
                c_u.Add("client001", "username001");
                c_u.Add("client002", "username002");
                Dictionary<string, string> u_psw = new Dictionary<string, string>();
                u_psw.Add("username001", "psw001");
                u_psw.Add("username002", "psw002");

                if (c_u.ContainsKey(c.ClientId) && c_u[c.ClientId] == c.Username)
                {
                    if (u_psw.ContainsKey(c.Username) && u_psw[c.Username] == c.Password)
                    {
                        c.ReturnCode = MqttConnectReturnCode.ConnectionAccepted;
                    }
                    else
                    {
                        c.ReturnCode = MqttConnectReturnCode.ConnectionRefusedBadUsernameOrPassword;
                    }
                }
                else
                {
                    c.ReturnCode = MqttConnectReturnCode.ConnectionRefusedIdentifierRejected;
                }
            };
            return options.ConnectionValidator;
        }

3、 相关事件

服务端支持 ClientConnected、ClientDisconnected 和 ApplicationMessageReceived 事件，分别用来检查客户端连接、客户端断开以及接收客户端发来的消息。

其中 ClientConnected 和 ClientDisconnected 事件的事件参数一个客户端连接对象 ConnectedMqttClient，通过该对象可以获取客户端ID标识 ClientId 和 MQTT 版本 ProtocolVersion。

ApplicationMessageReceived 的事件参数包含了客户端ID标识 ClientId 和 MQTT 应用消息 MqttApplicationMessage 对象，通过该对象可以获取主题 Topic、QoS QualityOfServiceLevel 和消息内容 Payload 等信息。

mqttServer.ApplicationMessageReceived += MqttServer_ApplicationMessageReceived;
mqttServer.ClientConnected += MqttServer_ClientConnected;
mqttServer.ClientDisconnected += MqttServer_ClientDisconnected;

五、 客户端

MQTT 与 HTTP 不同，后者是基于请求/响应方式的，服务器端无法直接发送数据给客户端。而 MQTT 是基于发布/订阅模式的,所有的客户端均与服务端保持连接状态。

那么客户端之间是如何通信的呢？

具体逻辑是：某些客户端向服务端订阅它感兴趣（主题）的消息，另一些客户端向服务端发布（主题）消息，服务端将订阅和发布的主题进行匹配，并将消息转发给匹配通过的客户端。

1、 创建客户端并连接

使用 MQTTnet 创建 MQTT 也非常简单，只需要使用 MqttClientFactory 对象的 CreateMqttClient 方法即可。

var factory = new MqttFactory();
var mqttClient = factory.CreateMqttClient();

创建客户端对象后，调用其异步方法 ConnectAsync 来连接到服务端。

await mqttClient.ConnectAsync(options, CancellationToken.None); // Since 3.0.5 with CancellationToken

调用该方法时需要传递一个 MqttClientTcpOptions 对象（之前的版本是在创建对象时使用该选项），该选项包含了客户端ID标识 ClientId、服务端地址（可以使用IP地址或域名）Server、端口号 Port、用户名 UserName、密码 Password 等信息。

// Create TCP based options using the builder.
var options = new MqttClientOptionsBuilder()
    .WithClientId("Client1")
    .WithTcpServer("broker.hivemq.com")
    .WithCredentials("bud", "%spencer%")
    .WithTls()
    .WithCleanSession()
    .Build();

2、 相关事件

客户端支持 Connected、Disconnected 和 UseApplicationMessageReceivedHandler事件，用来处理客户端与服务端连接、客户端从服务端断开以及客户端收到消息的事情。

mqttClient.UseApplicationMessageReceivedHandler(e =>
{
    Console.WriteLine("### RECEIVED APPLICATION MESSAGE ###");
    Console.WriteLine($"+ Topic = {e.ApplicationMessage.Topic}");
    Console.WriteLine($"+ Payload = {Encoding.UTF8.GetString(e.ApplicationMessage.Payload)}");
    Console.WriteLine($"+ QoS = {e.ApplicationMessage.QualityOfServiceLevel}");
    Console.WriteLine($"+ Retain = {e.ApplicationMessage.Retain}");
    Console.WriteLine();

    Task.Run(() => mqttClient.PublishAsync("hello/world"));
});

3、 订阅消息

客户端连接到服务端之后，可以使用 SubscribeAsync 异步方法订阅消息，该方法可以传入一个可枚举或可变参数的主题过滤器 TopicFilter 参数，主题过滤器包含主题名和 QoS 等级。

mqttClient.UseConnectedHandler(async e =>
{
    // Subscribe to a topic
    await mqttClient.SubscribeAsync(new TopicFilterBuilder().WithTopic("家/客厅/空调/#").Build()); 
});

4、 发布消息

mqtt的消息包含topic和payload两部分。topic就是消息主题（类型），用于另外一端判断这个消息是干什么用的。payload就是实际想要发送的数据。

• WithTopic给一个topic。
• WithPayload给一个msg。
• WithAtMostOnceQoS设置QoS，至多1次。也可以设为别的。
• PublishAsync异步发送出去。
  

string topic = "topic/hello";
var message = new MqttApplicationMessageBuilder()
                .WithTopic(topic)
                .WithPayload(msg)
                .WithAtMostOnceQoS()
                .WithRetainFlag()
                .Build();
await mqttServer.PublishAsync(message);

六、跟踪消息

// Write all trace messages to the console window.
MqttNetGlobalLogger.LogMessagePublished += (s, e) =>
{
    var trace = $">> [{e.TraceMessage.Timestamp:O}] [{e.TraceMessage.ThreadId}] [{e.TraceMessage.Source}] [{e.TraceMessage.Level}]: {e.TraceMessage.Message}";
    if (e.TraceMessage.Exception != null)
    {
        trace += Environment.NewLine + e.TraceMessage.Exception.ToString();
    }

    Console.WriteLine(trace);
};

七、 运行效果

以下分别是服务端、客户端1和客户端2的运行效果，其中客户端1和客户端2只是同一个项目运行了两个实例。客户端1用于订阅传感器的“温度”数据，并模拟上位机（如 APP 等）发送开关控制命令；客户端2订阅上位机传来的“开关”控制命令，并模拟温度传感器上报温度数据。

1、 服务端

2、 客户端1

3、 客户端2

八、完整Demo

完整实例：https://github.com/landbroken/MQTTLearning

使用MQTTNet在WPF框架下搭建MQTT客户端： https://blog.csdn.net/lordwish/article/details/84970800

使用MQTTNet在WPF框架下创建MQTT服务端(broker)：https://blog.csdn.net/lordwish/article/details/85042476

使用MQTTNet+ASP.NET Core创建MQTT服务器(broker)：https://blog.csdn.net/lordwish/article/details/86708777