MQTT通讯协议之MQTTnet

一、MQTT的概念

• MQTT 协议的全称是 Message Queuing Telemetry Transport&＃xff0c;翻译为消息队列传输探测&＃xff0c;它是 ISO 标准下的一种基于发布 - 订阅模式的消息协议&＃xff0c;它是基于 TCP/IP 协议簇的&＃xff0c;它是为了改善网络设备硬件的性能和网络的性能来设计的。

二、MQTT应用场景

• MQTT 一般多用于 IoT 即物联网上&＃xff0c;广泛应用于工业级别的应用场景&＃xff0c;比如汽车、制造、石油、天然气等。

三、MQTT的通信过程

• 角色
  • 客户端&＃xff08;Client&＃xff09;
    • 使用 MQTT 的程序或设备。客户端总是通过网络连接到服务端。它可以
      • 发布应用消息给其它相关的客户端。
      • 订阅以请求接受相关的应用消息
      • 取消订阅以移除接受应用消息的请求。
      • 从服务端断开连接。
  • 服务端&＃xff08;Server&＃xff09;
    • 一个程序或设备&＃xff0c;作为发送消息的客户端和请求订阅的客户端之间的中介。服务端
      • 接受来自客户端的网络连接
      • 接受客户端发布的应用消息
      • 处理客户端的订阅和取消订阅请求。
      • 转发应用消息给符合条件的客户端订阅。
  • 订阅&＃xff08;Subscription&＃xff09;
    • 订阅包含一个主题过滤器&＃xff08;Topic Filter&＃xff09;和一个最大的服务质量&＃xff08;QoS&＃xff09;等级。订阅与单个会话
      &＃xff08;Session&＃xff09;关联。会话可以包含多于一个的订阅。会话的每个订阅都有一个不同的主题过滤器
• 角色相关的术语
  • 主题名&＃xff08;Topic Name&＃xff09;&＃xff1a;
    附加在应用消息上的一个标签&＃xff0c;服务端已知且与订阅匹配。服务端发送应用消息的一个副本给每一个匹配的客户端订阅。
  • 主题过滤器&＃xff08;Topic Filter:&＃xff09;&＃xff1a;
    订阅中包含的一个表达式&＃xff0c;用于表示相关的一个或多个主题。主题过滤器可以使用通配符。
  • 会话&＃xff08;Session&＃xff09;&＃xff1a;
    客户端和服务端之间的状态交互。一些会话持续时长与网络连接一样&＃xff0c;另一些可以在客户端和服务端的多个连续网络连接间扩展。
  • 控制报文&＃xff08;MQTT Control Packet&＃xff09;&＃xff1a;
    通过网络连接发送的信息数据包。MQTT 规范定义了十四种不同类型的控制报文&＃xff0c;其中一个&＃xff08;PUBLISH 报 文&＃xff09;用于传输应用消息。

四、MQTT代码实现

• 条件

  • JDK

  • MQTT服务端

    • 下载地址
      链接&＃xff1a;https://pan.baidu.com/s/1uRBLZjkbS0OVbyM6lA9LnQ
      提取码&＃xff1a;umt3

  • 客户端代码

• 服务端安装

  • 执行命令

    #进入bin文件&＃xff0c;执行命令apollo.cmd create [文件夹名称]#进入新建文件夹下bin文件夹&＃xff0c;在执行命令apollo-broker.cmd run


• 客户端代码实现

  • nuget 安装包

    MQTTnet


  • 代码实现
    • 发布

      Console.WriteLine("---------------MQTT客户端 通信开始------------");//建立连接var conn &＃61; new MqttFactory().CreateManagedMqttClient();IMqttClientOptions value &＃61; new MqttClientOptionsBuilder().WithClientId("MQTTClie").WithTcpServer("127.0.0.1",1883).WithCredentials("admin","123456").Build(); IManagedMqttClientOptions options &＃61; new ManagedMqttClientOptionsBuilder().WithClientOptions(value).Build(); await conn.StartAsync(options); //客户端发布 MqttApplicationMessage mqttApplicationMessage &＃61; new MqttApplicationMessageBuilder().WithTopic("server").WithPayload(Encoding.UTF8.GetBytes("Good Luck&＃xff01;")).Build();await conn.PublishAsync(mqttApplicationMessage);Console.WriteLine("---------------MQTT客户端 通信结束------------");Console.ReadLine();


    • 订阅

      Console.WriteLine("---------------MQTT客户端 通信开始------------");//建立连接var conn &＃61; new MqttFactory().CreateManagedMqttClient();IMqttClientOptions value &＃61; new MqttClientOptionsBuilder().WithClientId("MQTTClie").WithTcpServer("127.0.0.1",1883).WithCredentials("admin","123456").Build(); //响应订阅主题conn.ApplicationMessageReceivedHandler&＃61; new MqttApplicationMessageReceivedHandlerDelegate(e &＃61;> {Console.WriteLine($"收到消息{e.ApplicationMessage.ConvertPayloadToString()},来自clientID{e.ClientId}");});IManagedMqttClientOptions options &＃61; new ManagedMqttClientOptionsBuilder().WithClientOptions(value).Build(); await conn.StartAsync(options);//订阅主题过滤器 MqttTopicFilter topicFilter &＃61; new MqttTopicFilterBuilder()//设置主题.WithTopic("server")//设置服务质量.WithQualityOfServiceLevel(MQTTnet.Protocol.MqttQualityOfServiceLevel.AtMostOnce).Build(); await conn.SubscribeAsync(topicFilter); Console.WriteLine("---------------MQTT客户端 通信结束------------");Console.ReadLine();


    • 既是发布也是订阅

      Console.WriteLine("---------------MQTT客户端 通信开始------------");//建立连接var conn &＃61; new MqttFactory().CreateManagedMqttClient();//响应订阅主题conn.ApplicationMessageReceivedHandler&＃61; new MqttApplicationMessageReceivedHandlerDelegate(e &＃61;> {Console.WriteLine($"收到消息{e.ApplicationMessage.ConvertPayloadToString()},来自clientID{e.ClientId}");});await Task.Delay(1000);IMqttClientOptions value &＃61; new MqttClientOptionsBuilder().WithClientId("MQTTClie").WithTcpServer("127.0.0.1",1883).WithCredentials("admin","123456").Build(); IManagedMqttClientOptions options &＃61; new ManagedMqttClientOptionsBuilder().WithClientOptions(value).Build(); await conn.StartAsync(options);//订阅主题MqttTopicFilter topicFilter &＃61; new MqttTopicFilterBuilder()//设置主题.WithTopic("server")//设置服务质量.WithQualityOfServiceLevel(MQTTnet.Protocol.MqttQualityOfServiceLevel.AtMostOnce).Build();await conn.SubscribeAsync(topicFilter);//客户端发布 MqttApplicationMessage mqttApplicationMessage &＃61; new MqttApplicationMessageBuilder().WithTopic("server").WithPayload(Encoding.UTF8.GetBytes("Good Luck&＃xff01;")).WithRetainFlag (true) .Build();await conn.PublishAsync(mqttApplicationMessage);Console.WriteLine("---------------MQTT客户端 通信结束------------");Console.ReadLine();