MQTT协议通信过程及使用介绍-京东云开发者社区

### 1 MQTT的诞生
众所周知，物联网是在传统互联网基础上延伸和扩展而出的概念，用户端从传统的计算机延伸和扩展到了任何物品与物品之间，而物品则通过各种传感器进行信息采集，然后通过计算设备进行网络信息交换与通信。物联网并不仅仅是一种网络，而是一个新的生态环境，它描述的本质是越来越多的使用物品通过网络连接在一起并可使用单个或者多个的终端设备对它们进行各种控制和使用。思科曾经预测过，物联网市场的规模可能在14万亿美金左右，前景极为光明。但是当前移动互联网正处于起步阶段，很多时候无法提供可靠的网络保障，因此，在1999年IBM主导并提出了MQTT协议，致力于解决这一方面的问题。

**MQTT（Message Queuing Telemetry Transport，消息队列遥测传输协议）**，是一种基于发布/订阅模式的轻量级通讯协议，该协议构建于TCP/IP协议上，属于应用层协议，因此只要是支持TCP/IP协议栈的地方，都可以使用MQTT。

### 2 MQTT的特点
MQTT是物联网连接协议。它被设计为一个极其轻量级的发布/订阅消息传输协议。对于需要较小代码占用空间或网络带宽非常宝贵的远程连接非常有用，是专为受限设备和低带宽、高延迟或不可靠的网络而设计。它体积小，功耗低，数据包最小，并且可以有效地将信息分配给一个或多个接收器。

**MQTT主要有以下特点：**
- 开放消息协议，简单易实现
- 发布订阅模式，一对多消息发布
- 基于TCP/IP网络连接,提供有序，无损，双向连接。
- 1字节固定报头，2字节心跳报文，最小化传输开销和协议交换，有效减少网络流量。
- 消息QoS支持，可靠传输保证

### 3 MQTT的原理
![](//img1.jcloudcs.com/developer.jdcloud.com/1df20122-99a7-4e03-b3a0-4fb97cd2f22220220329155205.png)
MQTT通信中有三种身份：发布者（Publish）、代理服务器（Broker）、订阅者（Subscribe），其中代理服务器把消息发布者和订阅者都视为客户端，发布者也可以同时是订阅者。客户端会建立一个到服务端的TCP连接通道用于提供两者之间的有序、无损的数据传输。

MQTT传输的消息分为主题（Topic）和负载（Payload）两部分，主题可以理解为消息的类型，客户端订阅后就会收到该主题的消息内容；负载可以理解为消息的内容。

每个客户端和服务器连接之后就是一个会话（Session），一个会话可以包含多个订阅。

#### 3.1 MQTT vs 消息队列
从这些概念我们可以看到，MQTT和我们平时使用的消息队列（例如JMQ）很类似，那么他们有什么区别呢？

首先MQTT是专为物联网设计的，其设计之初就考虑到了终端设备的能力，所以MQTT功能精简，只包含一些必备的功能，可有可无的功能不添加，例如负载均衡，同时其传输消息很精简，本文第四节会详细介绍协议的消息格式。所以MQTT适合于处理海量客户端的接入，但是单个客户端处理的消息量很少，而我们平时使用的消息队列则是接入的客户端较少，但是单个客户端处理的消息量很大。

#### 3.2 MQTT vs HTTP
HTTP协议作为应用非常广泛的互联网协议，而且和MQTT同属于应用层协议，他们之间有什么区别呢？

HTTP最初的目的是提供一种发布和接收HTML页面的方法，主要用于Web。HTTP是典型的C/S通讯模式：请求从客户端发出，服务端只能被动接收，一条连接只能发送一次请求，获取响应后就断开连接。该协议最早是为了适用Web浏览器的上网浏览场景而设计的，目前在PC、手机、Pad等终端上都应用广泛。由于这样的通信特点，HTTP技术在物联网设备中很难实现设备的反向控制。

#### 3.3 MQTT vs TCP

MQTT是构建于TCP/IP协议上，所以理论上讲用MQTT能实现的功能也可以采用TCP/IP方式实现，MQTT针对物联网场景已经做了标准化的规定，如果自己在通过TCP/IP协议实现这些功能，会增加很大的开发难度。

### 4 MQTT的报文结构
在MQTT协议中，一个MQTT数据包由：固定包头（Fixed header）、可变包头（Variable header）、消息体（payload）三部分构成。

#### 4.1 固定包头
MQTT的固定包头长度为2～5字节，又可分为3部分：控制包类型、标志位和剩余长度：

##### 4.1.1 控制包类型
第一个字节的前4位（第7～4位）无符号值表示该控制包的类型：
![](//img1.jcloudcs.com/developer.jdcloud.com/3a6e8362-915a-4df1-83b9-2ca26ee77f1320220329155909.png)

##### 4.1.2 标志位
第一个字节的后4位（第3～0位），根据控制包类型的不同，标志位的值也不同，目前只有PUBLISH控制包的标志位在使用，其余控制包的标志位都为固定值：
![](//img1.jcloudcs.com/developer.jdcloud.com/dd0dae51-5391-4d74-94b4-4f1ff6e7f9d620220329155953.png)

DUP和QoS标志位用于MQTT在不同服务质量下的消息传输 RETAIN是发布保留标识，表示服务器要保留该消息包，若是有新的订阅者出现，就把该消息包推送给它，若是没有那么推送至当前订阅者后丢弃

**服务质量**

目前MQTT有3种服务质量，所以用bit2和bit1两位来表示：

**00：即QoS==0**，最多分发一次。此种质量下消息的分发依赖于底层网络的能力，发送者发送带有QoS=0，DUP=0的PUBLISH包，接收者不会发送响应，发送者也不会重试，消息可能送达一次也可能根本没送达。如果客户端断开了连接，或者服务端出现了故障，该消息可能就会因此丢失。这也是最快的传输模式。

**01：即QoS==1**，至少分发一次。此种质量下确保消息至少送达一次。
1. 发送方首先发送带有QoS=1，DUP=0的PUBLISH包，并且需要设置一个未使用的包标识（详细介绍参见可变包头）
2. 接收方接到PUBLISH包之后会回复一个PUBACK包，PUBACK包没有消息体，可变包头中的包标识和接收到的PUBLISH包的包标识一致
3. 发送方接收到这个PUBACK包就认为该条消息已被接收方接收，根据PUBACK包中包标识找到本地的存储的消息包并丢弃，然后释放这个包标识以供后续使用
4. 如果发送方在一定时间内未收到PUBLISH包对应的PUBACK包，服务端会尝试将该PUBLISH包的DUP标志置为1（表示这个包是重新发送的），然后重新发送这个PUBLISH包，所以接收方可能会接收到重复的消息

**10：即QoS==2**，恰好分发一次。此级别是 MQTT 中最高级别的服务，保证每条消息仅被收件人接收一次。
1. 发送方首先发送带有QoS=2，DUP=0的PUBLISH包，并且需要设置一个未使用的包标识
2. 接收方接到PUBLISH包之后，本地保存该PUBLISH包并回复一个PUBREC包，PUBREC包没有消息体，可变包头中的包标识和接收到的PUBLISH包的包标识一致
3. 发送方接收到这个PUBREC包认为该条消息已被接收方接收，根据PUBREC包中包标识找到本地的存储的消息包并丢弃，然后存储这个PUBREC包并回复一个PUBREL包，PUBREL包没有消息体，可变包头中的包标识和接收到的PUBREC包的包标识一致
4. 当接收方接到PUBREL包之后，根据包头中的包标识找到本地存储PUBLISH消息并将消息投递给上层处理，然后丢弃该PUBLISH包并回复一个PUBCOMP，PUBCOMP包没有消息体，可变包头中的包标识和接收到的PUBREL包的包标识一致
5. 当接收方接到PUBCOMP包之后认为传输已经完成，丢弃本地存储的对应PUBREC包并释放这个包标识以供后续使用

整个过程经历了发送方和接收方的两次数据交互，在数据交互中涉及PUBLISH、PUBREC、PUBREL、PUBCOMP4种类型的数据包发送接收，那么在这4个包交互中都可能存在发送失败或者接收失败的情况，针对这4种情况分别又是怎么处理的呢？
- 首先，PUBLISH、PUBREC这两个数据包发送失败，发送方在一段时间内没有收到PUBREC包的话都会将PUBLISH包的DUP标志置为1重新发送，同时接收方接到消息并没有立即投递上层处理而是先存储下来等待PUBREL包的指令，重复的PUBLISH包会丢弃
- 其次，PUBREL、PUBCOMP这两个数据包发送失败，发送方在一段时间内没有收到PUBCOMP包的话都会重新发送PUBREL，同时接收方接到PUBREL包将PUBLISH消息投递之后会删除本地存储的PUBLISH包，如果再次接收到PUBREL包，本地找不到PUBLISH消息会认为该消息已经处理从而直接回复PUBCOMP包

从这两方面，就可以保证消息有且只有一次被消费。
但是，这个过程经历两次交互，效率较低同时要求接收方有持久化的能力，而在物联网场景下有一部分的客户端构造很简单不具备这样的能力，所以这种服务质量用的较少

##### 4.1.3 剩余长度
从第二个字节开始，固定包头剩余部分代表了消息剩余长度，注意这个剩余长度指的是可变包头+消息体的长度，不包含固定包头。剩余长度每个字节的低7位用来编码数据，最高位用来表示是否还有后续字节，因此每个字节最大可以表示2^7=128，再加上一个标识位，剩余长度最多可以用四个字节来表示，于是剩余长度最大为128^4=256M，所以MQTT单个消息最大是256M。

#### 4.2 可变包头
它驻位于固定的头和消息体之间。可变头的内容因数据包类型而不同，较常的应用是做为包标识，很多类型数据包中都包括一个2字节的数据包标识字段：PUBLISH (QoS > 0)、PUBACK、PUBREC、PUBREL、PUBCOMP、SUBSCRIBE、SUBACK、UNSUBSCRIBE、UNSUBACK

#### 4.3 消息体
消息体位MQTT数据包的第三部分，CONNECT、SUBSCRIBE、SUBACK、UNSUBSCRIBE四种类型的消息有消息体：
- CONNECT：消息体内容主要是：客户端的ClientID、订阅的Topic、Message以及用户名和密码
- SUBSCRIBE：消息体内容是一系列的要订阅的主题以及QoS
- SUBACK：消息体内容是服务器对于SUBSCRIBE所申请的主题及QoS进行确认和回复
- UNSUBSCRIBE：消息体内容是要订阅的主题

### 5 MQTT的使用
咱们的JMQ4平台已经支持了MQTT协议，可以很方便的接入使用MQTT。
首先进入JMQ4国内站控制台，点击申请生产发布
![](//img1.jcloudcs.com/developer.jdcloud.com/7a6a18b1-9ffa-40a4-b83e-c32edad2114a20220329160322.png)

客户端类型选择MQTT。如果是要接入消费，那么点击申请消费订阅
![](//img1.jcloudcs.com/developer.jdcloud.com/6b537725-fa5b-4fc4-9a21-51fc4bdb8b0f20220329160356.png)

填写对应信息提交流程就可以了，流程审批通过之后就可以发送或者消费MQTT消息了
这里以java为例，引入客户端jar
```java
<dependency>
   <groupId>org.eclipse.paho</groupId>
   <artifactId>org.eclipse.paho.client.mqttv3</artifactId>
   <version>1.1.1</version>
</dependency>
```

这是一个简单的消费客户端的例子：
```java
public class MqttClientDemo {

private static String topic = "your topic";
    private static String address = "tcp://192.168.131.206:1883";
    private static String username = "your username";
    private static String password = "your password";
    private static String constantClientId = "your clientId";
    private static MqttClient client;
    private static MqttConnectOptions connectOptions = new MqttConnectOptions();

static {
        try {
            String clientId = constantClientId + "-" + UUID.randomUUID().toString().replace("-", "");
            client = new MqttClient(address, clientId, new MemoryPersistence());
            client.setTimeToWait(1000);
            client.setCallback(new MqttCallback() {

@Override
                public void connectionLost(Throwable cause) {
                    System.out.println("connectionLost-----------: <" + cause + ">");
                }

@Override
                public void deliveryComplete(IMqttDeliveryToken token) {
                    System.out.println("deliveryComplete-----------: <" + token + ">");
                }

@Override
                public void messageArrived(String topic, MqttMessage arg1)
                        throws Exception {
                    System.out.println("messageArrived-----------: <" + topic + ">, <" + arg1.toString() + ">");
                }
            });

connectOptions.setCleanSession(false);
            connectOptions.setUserName(username);
            connectOptions.setPassword(password.toCharArray());
            connectOptions.setConnectionTimeout(30);
            connectOptions.setKeepAliveInterval(60);
            connectOptions.setMqttVersion(MqttConnectOptions.MQTT_VERSION_3_1_1);
            connectOptions.setMaxInflight(100);

client.connect(connectOptions);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

public void publish(String messagebody) throws MqttException {
        MqttMessage message = new MqttMessage();
        message.setQos(1);
        message.setPayload(messagebody.getBytes());
        boolean connected = client.isConnected();
        client.publish(topic, message);
    }
}
```

### 6 MQTT的缺点
MQTT作为专为物联网而诞生的协议，目前已广泛应用在各个物联网相关场景。但是由于其极简的设计原则，所以MQTT还是存在着不少的缺点：
- 不支持文件和影音：在某些应用场景中，需要传输的信息可能不限于需要通过影音与文件通信的指令，例如语音和视频信号。
- 不支持与第三方HTTP集成：MQTT协议优于常规HTTP协议，但是基于传统HTTP协议的WEB服务器仍在主流市场中占主导地位。这些服务器应与MQTT协议互连，以降低升级成本，但目前MQTT不能支持。
- 不支持负载均衡：负载均衡服务器对于高并发性和防止恶意攻击也是必不可少的。
- 设备脱机后，不支持脱机消息来补偿从MQTT服务器到设备的控制信息丢失。
- 不支持点对点通信，并且使用标准的MQTT协议。从理论上讲，点对点通信可以通过相互订阅来实现，但是逻辑相对复杂并且涉及设备安全性。当设备B和设备C相同时-在主题的情况下，设备A无法知道消息是来自设备B还是来自设备C，并且消息很可能是被设备D窃听的。

------------
**自猿其说Tech-京东物流技术与数据智能部
作者：何红呈**