MQTT从入门到放弃_mqttcallbackextended

MQTT，是一种基于发布/订阅模式的"轻量级"通讯协议，该协议构建于TCP/IP协议上，属于应用层协议，。

基于TCP协议、发布/订阅协议，属于应用层协议。使用C/S架构，本质是一个消息转发协议。所有的客户端往服务器发送消息，然后服务端根据过滤规则，把消息再转发给符合条件的客户端。消息的传输是有序的、可靠的、双向的。

一、概述

1.1 参考文档

• 官方文档 （推荐） http://docs.oasis-open.org/mqtt/mqtt/v3.1.1/os/mqtt-v3.1.1-os.html
• 官方文档（中文） http://mqtt.p2hp.com/mqtt311

1.2 MQTT优点

MQTT最大优点在于，可以以 极少的代码和有限的带宽 ，为远程连接设备提过实时可靠的消息服务，作为一种低开销、低带宽占用的即时通讯协议，使其在物联网、小型设备、移动应用等方面有较广泛的应用

• 保持长连接，具有一定实时性
• 适应高延时，偶尔断网
• 支持高并发
• 单次数据量小
• 传输可靠
• 提供不同QoS（服务优先级）
• 设置遗嘱消息

1.3 MQTT应用领域

MQTT是基于二进制消息的发布/订阅编程模式的消息协议，非常适合

需要 低功耗 和 网络带宽有限 的IoT场景

比如： 遥感数据、汽车、 智能家居、智慧城市、医疗医护、智慧农业 …

二、MQTT协议原理

实现MQTT协议需要客户端和服务器端通讯完成，在通讯过程中，MQTT协议中有三种身份：发布者（Publish）、代理（Broker）（服务器）、订阅者（Subscribe）。

注意：消息的发布者和订阅者都是客户端，消息代理是服务器，消息发布者可以同时是订阅者。

2.1 MQTT客户端

一个使用MQTT协议的应用程序或者设备，它总是建立到服务器的网络连接。客户端可以：

• 发布其他客户端可能会订阅的信息
• 订阅其它客户端发布的消息
• 退订或删除应用程序的消息
• 断开与服务器连接

2.2 MQTT服务端

QTT服务器以称为“消息代理”（Broker），可以是一个应用程序或一台设备。它是位于消息发布者和订阅者之间，它可以：

• 接受来自客户的网络连接
• 接受客户发布的应用信息
• 处理来自客户端的订阅和退订请求
• 向订阅的客户转发应用程序消息
  
  

2.3 消息结构

每条MQTT命令消息的消息头都包含一个固定的报头，有些消息会携带一个可变报文头和一个负荷。消息格式如下：

固定报文头 | 可变报文头 | 负载

2.3.1 固定报文头

存在于所有MQTT数据包中，表示数据包类型及数据包的分组类标识。

MQTT固定报文头最少有两个字节，第一字节包含消息类型（Message Type）和QoS级别等标志位。第二字节开始是剩余长度字段，该长度是后面的可变报文头加消息负载的总长度，该字段最多允许四个字节。

2.3.2 可变报文头

存在于部分MQTT数据包中，数据包类型决定了可变头是否存在及其具体内容。

可变报文头主要包含协议名、协议版本、连接标志（Connect Flags）、心跳间隔时间（Keep Alive timer）、连接返回码（Connect Return Code）、主题名（Topic Name）等。

2.3.3 负载

Payload直译为负荷，消息的内容。存在于部分MQTT数据包中，表示客户端收到的具体内容。

2.4 MQTT特点

2.4.1 MQTT的消息类型

固定报文头中的第一个字节包含连接标志（Connect Flags），连接标志用来区分MQTT的消息类型。MQTT协议拥有14种不同的消息类型（见下表），可简单分为连接及终止、发布和订阅、QoS 2消息的机制以及各种确认ACK。至于每一个消息类型会携带什么内容，这里不多阐述

2.4.2 服务质量（QOS）

2.4.2.1 QOS分类

服务质量水平(QoS)是一个消息的发送者和限定递送保证用于特定消息的消息的接收器之间的协议。MQTT 中有 3 个 QoS 级别：

• QoS0：发送就不管了，最多一次；
• QoS1：发送之后依赖MQTT规范，是否启动重传消息，所以至少一次；
• QoS2：发送之后依赖MQTT消息机制，确保只有一次。

QoS0 代表，Sender 发送的一条消息，Receiver 最多能收到一次，也就是说 Sender 尽力向 Receiver发送消息，如果发送失败，也就算了；这是完全依赖TCP重传机制，如果网络不好，TCP的重传也不是100%可靠，加上MQTT是Publisher 发出去的消息是依赖代理服务器完成转发，所以消息最多一次。

QoS1 代表，Sender 发送的一条消息，Receiver 至少能收到一次，也就是说 Sender 向 Receiver发送消息，如果发送之后没有收到对应的PUBACK，就会继续重试，直到发送者Sender 接收到 Receiver 发送的 PUBACK为止，因为重传的原因，Receiver 有可能会收到重复的消息；

QoS2 代表，Sender 发送的一条消息，Receiver 确保能收到而且只收到一次，也就是说 Sender 尽力向 Receiver 发送消息，如果发送失败，会继续重试，直到 Receiver 收到消息为止，同时保证 Receiver 不会因为消息重传而收到重复的消息。（个人理解这一点有点像TCP三次握手的交互过程）

2.4.2.2 QOS特性

• QoS 是 MQTT 协议的一个关键特性。QoS 使客户端能够选择与其网络可靠性和应用程序逻辑相匹配的服务级别。因为 MQTT 管理消息的重新传输并保证交付（即使底层传输不可靠），QoS 使不可靠网络中的通信变得更加容易。
  
  
• QoS流，在发送端和接收端是两件不同的事情。当然发送端与接收端QoS的等级也可以不一样。在发送端与broker之间，发送端定义了QoS等级。当broker发送消息到接收端是，接收端决定了QoS的等级
  
  
• 发送（发布）消息的客户端和接收消息的客户端之间的 QoS 定义和级别是两件不同的事情。这两种交互的 QoS 级别也可以不同。向代理发送 PUBLISH 消息的客户端定义消息的 QoS。但是，当代理将消息传递给接收者（订阅者）时，代理使用接收者（订阅者）在订阅期间定义的 QoS。例如，客户端 A 是消息的发送者。客户端 B 是消息的接收者。如果客户端 B 以 QoS 1 订阅代理并且客户端 A 以 QoS 2 向代理发送消息，则代理以 QoS 1 将消息传递给客户端 B（接收者/订阅者）。

2.4.2.3 QOS应用场景

QoS 0

• 发送方和接收方之间建立了完全或大部分稳定的连接。
• 不介意偶尔丢失几条消息。如果数据不是那么重要或数据间隔很短，则某些消息的丢失是可以接受的
• 不需要消息队列。仅当断开连接的客户端具有 QoS 1 或 2 和持久会话时，消息才会排队

QoS 1

• 您需要获取每条消息，并且您的用例可以处理重复项。QoS 级别 1 是最常用的服务级别，因为它保证消息至少到达一次，但允许多次传递。当然，您的应用程序必须容忍重复并能够相应地处理它们。
• 无法承受 QoS 2 的开销。QoS 1 传递消息的速度比 QoS 2 快得多。

QoS 2

• 支付场景。一次接收所有消息对您的应用程序至关重要。如果重复交付可能损害应用程序用户或订阅客户端，则通常会出现这种情况。请注意开销以及 QoS 2 交互需要更多时间才能完成。

关于QOS的优秀连接：
https://blog.csdn.net/m0_50668851/article/details/112555171
https://blog.csdn.net/qq1623803207/article/details/89518318

2.4.2.3 发布、订阅qos不一致

向代理发布消息的客户端在向代理发送消息时定义了消息的 QoS 级别。代理使用每个订阅客户端在订阅过程中定义的 QoS 级别将此消息传输到订阅客户端。
对于发布和订阅消息的客户端，服务端会主动采用较低级别的QoS来实现消息传输。

MQTT 发布与订阅操作中的 QoS 代表了不同的含义，发布时的 QoS 表示消息发送到服务端时使用的 QoS，订阅时的 QoS 表示服务端向自己转发消息时可以使用的最大 QoS。

• 当客户端 A 的发布 QoS 大于客户端 B 的订阅 QoS 时，服务端向客户端 B 转发消息时使用的 QoS 为客户端 B 的订阅 QoS。
• 当客户端 A 的发布 QoS 小于客户端 B 的订阅 QoS 时，服务端向客户端 B 转发消息时使用的 QoS 为客户端 A 的发布 QoS。

不同情况下客户端收到的消息 QoS 可参考下表：

参考链接~~~

2.4.3 遗愿标志（Will Flag）

在可变报文头的连接标志位字段（Connect Flags）里有三个Will标志位：Will Flag、Will QoS和Will Retain Flag，这些Will字段用于监控客户端与服务器之间的连接状况。如果设置了Will Flag，就必须设置Will QoS和Will Retain标志位，消息主体中也必须有Will Topic和Will Message字段。

那遗愿消息是怎么回事呢？
服务器与客户端通信时，当遇到异常或客户端心跳超时的情况，MQTT服务器会替客户端发布一个Will消息。当然如果服务器收到来自客户端的DISCONNECT消息，则不会触发Will消息的发送。

因此，Will字段可以应用于设备掉线后需要通知用户的场景。

遗愿消息的触发条件：

1. broker 发现 I/O 异常或网络异常。
2. client 的 keep alive 超时。
3. client 的网络中断，但是没有发送 DISCONNECT 包。
4. protocol error。

2.4.4 连接保活心跳机制（Keep Alive Timer）

MQTT客户端可以设置一个心跳间隔时间（Keep Alive Timer），表示在每个心跳间隔时间内发送一条消息。如果在这个时间周期内，没有业务数据相关的消息，客户端会发一个PINGREQ消息，相应的，服务器会返回一个PINGRESP消息进行确认。如果服务器在一个半（1.5）心跳间隔时间周期内没有收到来自客户端的消息，就会断开与客户端的连接。心跳间隔时间最大值大约可以设置为18个小时，0值意味着客户端不断开。

2.4.5 MQTT vs MQ

MQTT：一种通信协议，类似人类交谈中的汉语、英语、俄语中的一种语言规范

MQ：一种通信通道，也叫消息队列，类似人类交谈中的用电话、email、微信的一种通信方式

市面上的MQ产品很多，如阿里自研并开源RocketMQ，还有类似RabbitMQ、ActiveMQ，他们不仅支持MQTT协议，还支持如AMQP、stomp协议等等，EMQ 使用的协议是mqtt。

2.4.6 协议对比

下图是各个协议间的对比：

MQTT协议（低带宽）

MQTT （Message Queuing Telemetry Transport ），消息队列遥测传输，由IBM开发的即时通讯协议，相比来说比较适合物联网场景的通讯协议。MQTT协议采用发布/订阅模式，所有的物联网终端都通过TCP连接到云端，云端通过主题的方式管理各个设备关注的通讯内容，负责将设备与设备之间消息的转发。

适用范围：在低带宽、不可靠的网络下提供基于云平台的远程设备的数据传输和监控。

MQTT协议一般适用于设备数据采集到端（Device-》Server，Device-》Gateway），集中星型网络架构（hub-and-spoke），不适用设备与设备之间通信，设备控制能力弱，另外实时性较差，一般都在秒级。

AMQP协议（互操作性）
AMQP（Advanced Message Queuing Protocol），先进消息队列协议，这是OASIS组织提出的，该组织曾提出OSLC（Open Source Lifecyle）标准，用于业务系统例如PLM，ERP，MES等进行数据交换。

适用范围：最早应用于金融系统之间的交易消息传递，在物联网应用中，主要适用于移动手持设备与后台数据中心的通信和分析。

XMPP协议（即时通信）

XMPP（Extensible Messaging and Presence Protocol）可扩展通讯和表示协议，XMPP的前身是Jabber，一个开源形式组织产生的网络即时通信协议。XMPP目前被IETF国际标准组织完成了标准化工作。

适用范围：即时通信的应用程序，还能用在网络管理、内容供稿、协同工具、档案共享、游戏、远端系统监控等。

JMS （Java Message Service）

Java消息服务（Java Message Service）应用程序接口，是一个Java平台中关于面向消息中间件（MOM）的API，用于在两个应用程序之间，或分布式系统中发送消息，进行异步通信。Java消息服务是一个与具体平台无关的API，绝大多数MOM提供商都对JMS提供支持
JMS是协议同时也是 Java 消息服务规范的标准实现，同时也是 Java 企业版（JEE）规范的一部分。

优秀连接 https://blog.csdn.net/gyshun/article/details/83036987

2.5 消息持久化

需要满足以下三个条件：
1、cleanSession = false
2、clientId不为空
3、客户端subscribe时的Qos=1，发布端publish时的Qos=1

// 接受离线消息  告诉代理客户端是否要建立持久会话   false为建立持久会话
 mqttConnectOptions.setCleanSession(Boolean.FALSE);
1
2

2.6 实现方式!!!

参考链接：https://blog.51cto.com/u_15067242/2574302

MQTT客户端采用的是Spring Intergration和Eclipse.paho的方式实现的。当然，你也可以直接使用Eclipse.paho作为客户端连接。

2.6.1 Spring Intergration（推荐）

什么是Spring Intergration？

Spring Integration是一个功能强大的EIP (Enterprise Integration
Patterns)，即企业集成模式。它是Spring Messaging的扩展，提供了Spring编程模型的扩展，用来支持企业集成模式。它集成了众多功能，是一种便捷的事件驱动消息框架用来在系统之间做消息传递的。

其实Spring Intergration就类似一个水电系统。总闸、各楼层的控制、分流、聚合、过滤、沉淀、消毒、排污，这里的每一个环节都类似一个系统服务，可能是MQTT，可能是Redis，可能是MongoDB，可能是Job，可能是我们系统服务的任何一个模块。那么Spring Intergration扮演的角色就是将这些功能能够连接起来组成一个完整的服务系统，实现企业系统的集成的解决方案。就像管道一样各个模块连接到起，管道能够连接到千家万户需要很多水表、分头管、水龙头，管道开关等等这些都是Spring Intergration的主要组件。

        <!-- mqtt依赖-->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-integration</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.integration</groupId>
            <artifactId>spring-integration-stream</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.integration</groupId>
            <artifactId>spring-integration-mqtt</artifactId>
        </dependency>
1
2
3
4
5
6
7
8
9
10
11
12
13

spring-integration-mqtt内部依赖了Eclipse.paho的包，所以不需要在单独引入

关于Spring Intergration的版本问题，其官方文档：官方文档连接，值得深看！

这里是对官方文档的部分誊写&理解：

       从4.1版本开始，编程方式改变适配器订阅的主题可以省略url，DefaultMqttPahoClientFactory属性serverURIs可以提供服务端URI，例如，这将使能连接至HA高可用簇。

       从4.2.2版本开始，当适配器成功订阅至主题后，发布MqttSubscribedEvent，当连接/订阅失败时，发布MqttConnectionFailedEvent。这些事件可以由实现ApplicationListener接口的实体类获取。
       新的属性recoveryInterval控制在故障之后适配器会尝试重新连接的时间间隔，默认为10000ms（10s）

       在4.2.3版本之前，当适配器停止后，客户端总是会解除订阅，这是不正确的。 ，因为如果客户端QoS大于0，我们需要保持订阅以便适配器停止时到达的消息在下一次开始时会传送。这也需要设置客户端工厂cleanSession属性为false，默认值为true。

       从4.2.3版本开始，如果cleanSession值为false，适配器不会解除订阅（默认）。可以重写该行为，通过设置工厂属性consumerCloseAction，可以有以下值：UNSUBSCRIBE_ALWAYS,UNSUBSCRIBE_NEVER以及UNSUBSCRIBE_CLEAN，后者（默认）会解除订阅仅当cleanSession属性值为true。回退至4.2.3之前的行为，使用UNSUBSCRIBE_ALWAYS。

---

java实现代码见第三部分

---

2.6.2 Eclipse.paho (mqtt3，看看)

Eclipse.paho是基于IMqttClient和IMqttAnsycClient接口实现的MQTT客户端中间件。其内部实现了完整的消息发布与订阅、socket长连接、心跳机制、断线重连以及消息本地缓存等一系列功能。是目前比较主流的MQTT客户端中间件。

客户端：

package com;

import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;

public class MqttClientTest {
   
    //订阅的主题
    public static final String TOPIC = "mqtt-fabu";

    public static void main(String[] args) {
   
        MyClient myClient = new MyClient();
        myClient.subscribe(TOPIC, 1);
    }
}

/**
 * 订阅方
 */
class MyClient {
   
    //mqtt服务器默认的地址和端口号
    //这里的tcp地址就是上述mqtt服务端地址(emqx地址,端口号就是1883，emqx服务器的端口号是18083)
    public static final String HOST = "tcp://172.16.9.205:1883";

    //连接MQTT的客户端ID，一般以唯一标识符表示
    private static final String CLIENTID = "client-1";
    //连接的用户名密码（非必需）
    private String userName = "admin";
    private String password = "public";

    private MqttClient mqttClient;

    public  MyClient() {
   
        try {
   
            mqttClient = new MqttClient(HOST, CLIENTID, new MemoryPersistence());
            MqttConnectOptions options = new MqttConnectOptions();
            options.setCleanSession(true);
            options.setConnectionTimeout(60);
            options.setKeepAliveInterval(10);
            options.setUserName(userName);
            options.setPassword(password.toCharArray());
            //定义回调函数
            mqttClient.setCallback(new PushCallBack());
            mqttClient.connect(options);
        } catch (MqttException e) {
   
            e.printStackTrace();
        }
    }

    //订阅主题
    public void subscribe(String topic, int qos) {
   
        try {
   
            mqttClient.subscribe(topic, qos);
        } catch (MqttException e) {
   
            e.printStackTrace();
        }
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69

消费者：

package com;

import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;

public class MqttServerTest {
   
    //发布的主题
    public static final String TOPIC = "mqtt-fabu";

    public static void main(String[] args) throws InterruptedException {
   
        MqttServer mqttServer = new MqttServer();
        MqttMessage message = new MqttMessage();
        message.setQos(1);
        message.setPayload("第一次广播".getBytes());
        mqttServer.publish(TOPIC, message);
        Thread.sleep(1000);
        message.setPayload("第二次广播".getBytes());
        mqttServer.publish(TOPIC, message
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22