一、 概述

　　MQTT(Message Queuing Telemetry Transport，消息队列遥测传输协议)，是一种基于发布/订阅模式的"轻量级"通讯协议，它和 Modbus TCP 一样都是基于 TCP/IP 之上的应用层协议。

　　1.1 MQTT 具备以下几个特点

　　简单容易实现、支持设置消息服务质量等级(QoS)、轻量且省带宽。

　　1.2 为什么是 MQTT

　　物联网设备通信有一些问题需要针对性的解决：网络环境复杂而不可靠、硬件内存和闪存容量小、处理器能力有限。MQTT 协议就是为了解决遇到的这些问题而创建的。

　　1.3 MQTT 与 HTTP 协议对比

　　MQTT 最小报文仅为 2 个字节，比 HTTP 占用更少网络开销。

　　http需要指定请求头、请求方式、请求协议版本等比MQTT请求更多的内容，也代表着它要比MQTT的网络开销要更多。

　　MQTT 基于发布订阅模式，HTTP 基于请求响应。

　　

 

　　MQTT 可实时推送消息，HTTP 需要通过轮询获取数据。

　　MQTT 是有状态的，HTTP 是无状态的。

　　MQTT客户端启动之后与Broker进行连接之后后续请求不用再带自身认证信息，而对于http来说客户端需要每次请求都带着自身认证信息表明自身身份。

　　1.4 MQTT 与消息队列对比

　　MQTT 和消息队列的很多行为和特性非常接近， 比如都采用发布/订阅模式，但是他们面向的场景却 有着显著的不同。消息队列主要用于服务端应用之间的消息存储与转发，这类场景往往数据量大但客户端 数量少。MQTT 是一种消息传输协议，主要用于物联网设备之间的消息传递，这类场景的特点是海量的设备接入、管理与消息传输。

　　1.5 MQTT消息服务质量等级

　　MQTT协议可以对消息设置三种不同的服务质量等级，分别是：

　　QoS=0：最多发送一次，订阅方不会发送接收到消息的响应，发布方也不会重新发送消息，消息可能会丢失。

 

　　QoS=1：最少发送一次，订阅方接收到消息之后发送响应给发布方，如果发布方没收到响应会重新发送消息，直到接收到响应为止，消息可能会重复。

 

 

　　QoS=2：保证只有一次，订阅方不会重复接收到消息，也不会丢失消息。发布方和订阅方之间需要进行两次请求响应确认消息传输，保证相同消息只接收一次，QoS为2时消耗网络资源也最多。

 

　　1.6 MQTT 主题通配符

　　MQTT 通配符包含单层通配符 `+` 和多层通配符`#`,主要用于客户端一次订阅多个主题。在发布的时候不能用通配符下发到多个主题中。

　　单层通配符适用于单个主题层级匹配的通配符,单层通配符必须占据整个层级才能生效。

 

　　多层通配符适用于主题匹配任意层级的通配符，多层通配符表示它的父级和任意数量的子层级，多层通配符必须单独占据一个层级才能生效。

 

　　二、 创建 MQTT 链接

　　下面我们通过 Java 代码创建一个 MQTT 链接，模拟设备发送数据，通过代码模拟订阅者接收数据。

　　2.1添加依赖

　　1.<dependency>  

       2.    <groupId>org.eclipse.paho</groupId>  

       3.    <artifactId>org.eclipse.paho.client.MQTTv3</artifactId>  

       4.    <version>1.2.5</version>  

       5.</dependency>

　　2.2编写发布者代码

　　1./**

　　2. *

　　3. * @param topic

　　4. * @param content

　　5. * @param qos 有3个值分别为:

　　6. * 0:消息最多传递一次,如果客户端没有接收到消息,那么消息会丢失

　　7. * 1:消息传递至少一次,发布者发送消息会等待订阅者的ACK,如果在规定时间内没有收到ACK则会重新发送,可能会出现消息重复

　　8. * 2:消息仅且只发送一次,保证消息到达对方并且只发送一次

　　9. */

　　10.public void sendMessage(MQTTClient client,String topic,String content,int qos){

　　11. MQTTMessage message = new MQTTMessage(content.getBytes());

　　12. message.setQos(qos);

　　13. try {

　　14. client.publish(topic,message);

　　15. } catch (MQTTException e) {

　　16. //todo publish fail

　　17. }

　　18.}

　　19./**

　　20. * 连接MQTT客户端

　　21. * @return

　　22. */

　　23.public MQTTClient connectMQTT() {

　　24. MQTTClient client = null;

　　25. try {

　　26. //broker

　　27. String broker = "tcp://localhost:1883";

　　28. //用户名

　　29. String username = "admin";

　　30. //密码

　　31. String password = "123456";

　　32. //clientId

　　33. String clientId = System.currentTimeMillis() + "";

　　34. //创建MQTT客户端(指定broker、客户端id、消息持久策略)

　　35. client = new MQTTClient(broker, clientId, new MemoryPersistence());

　　36. //创建连接参数配置

　　37. MQTTConnectOptions options = new MQTTConnectOptions();

　　38. options.setUserName(username);

　　39. options.setPassword(password.toCharArray());

　　40. //是否清除会话

　　41. options.setCleanSession(true);

　　42. //心跳间隔

　　43. options.setKeepAliveInterval(20);

　　44. //连接超时时间

　　45. options.setConnectionTimeout(20);

　　46. //是否自动重连

　　47. options.setAutomaticReconnect(true);

　　48. //连接

　　49. client.connect(options);

　　50.

　　51. }catch (Exception e){

　　52. //todo connect fail

　　53. }

　　54. return client;

　　55.}

　　56.

　　57./**

　　58. * 关闭MQTT客户端

　　59. * @param client

　　60. */

　　61.public void closeClient(MQTTClient client){

　　62. try {

　　63. //关闭连接

　　64. client.disconnect();

　　65. //关闭客户端

　　66. client.close();

　　67. } catch (MQTTException e) {

　　68. //todo close fail

　　69. }

　　70.}

　　现在我们来模拟给订阅者发送一条消息，我们通过 MQTTX 客户端来模拟测试：

　　1.Demo demo = new Demo();

　　2.MQTTClient client = demo.connectMQTT();

　　3.String content = "{\n" +

　　4. " \"deviceNo\": \"x5a2aw\",\n" +

　　5. " \"val\": 19\n" +

　　6. "}";

　　7.//执行主题

　　8.String topic = "deviceUp";

　　9.int qos = 1;

　　10.if (null != client){

　　11. demo.sendMessage(client,topic,content,qos);

　　12.}

　　13.demo.closeClient(client);

　　2.3编写订阅者代码

　　1.String broker = "tcp://localhost:1883";

　　2.String topic = "/deviceUp";

　　3.String username = "admin";

　　4.String password = "123456";

　　5.String clientId = System.currentTimeMillis() + "";

　　6.int qos = 1;

　　7.

　　8.try {

　　9. MQTTClient client = new MQTTClient(broker, clientId, new MemoryPersistence());

　　10. // 连接参数

　　11. MQTTConnectOptions options = new MQTTConnectOptions();

　　12. options.setUserName(username);

　　13. options.setPassword(password.toCharArray());

　　14. options.setConnectionTimeout(60);

　　15. options.setKeepAliveInterval(60);

　　16. // 设置回调

　　17. client.setCallback(new MQTTCallback() {

　　18. public void connectionLost(Throwable cause) {

　　19. }

　　20. public void messageArrived(String topic, MQTTMessage message) {

　　21. System.out.println("topic为: " + topic);

　　22. System.out.println("topic为: " + message.getQos());

　　23. System.out.println("消息内容为: " + new String(message.getPayload()));

　　24. }

　　25. public void deliveryComplete(IMQTTDeliveryToken token) {

　　26. }

　　27. });

　　28. client.connect(options);

　　29. client.subscribe(topic, qos);

　　30.} catch (Exception e) {

　　31. //todo subscribe failed

　　32.}

　　启动订阅者之后再启动发布者代码，与此同时也可以启动自己本地的MQTTX接收消息。

　　订阅者收到消息为：

　　

 

　　MQTTX收到消息为：

 

　　三、 总结

　　以上就是对MQTT的简单介绍和通过Java代码创建一个 MQTT 链接，模拟设备发送数据，通过代码模拟订阅者接收数据，希望可以对你有所帮助。更多关于MQTT的细节可以留言获取或者通过阅读MQTT官方文档进一步学习。