mqtt_manul

MQTT物联网协议的学习笔记

一、MQTT基础知识

主要优势

1. 发布订阅模式，一对多消息发布
2. 基于 TCP/IP 网络连接
3. 消息Qos支持，可靠传输保证（QoS机制保证可靠传输）
4. 灵活的消息传输，不关心 Payload 数据格式（考虑json格式）
5. 持续的会话感知能力，时刻知道设备是否在线

主流物联网协议选择：MQTT、CoAP 还是 LwM2M

主题topic

发布订阅模式区别于传统的客户端-服务器模式，它使发送消息的客户端（发布者）与接收消息的客户端（订阅者）分离，发布者与订阅者不需要建立直接联系。

MQTT 协议使用发布订阅模式，并且根据主题而不是消息内容来路由消息，每个消息都包含一个主题，代理无需解析用户数据，这为实现一个通用的、与业务无关的 MQTT 代理提供了可能。

MQTT 主题 (Topic) 类似 URL 路径，例如:

sensor/10/temperature
sensor/+/temperature

# $SYS 主题路径以 $SYS/brokers/{node}/ 开头。{node} 是指产生该 事件 / 消息 所在的节点名称
$SYS/brokers/emqx@127.0.0.1/version
$SYS/brokers/emqx@127.0.0.1/uptime

主题 (Topic) 通过’/‘分割层级，支持’+', '#'通配符:

'+': 表示通配一个层级，例如 a/+，匹配 a/x, a/y

'#': 表示通配多个层级，例如 a/#，匹配 a/x, a/b/c/d

排它订阅

排它订阅允许对主题进行互斥订阅，一个主题同一时刻仅被允许存在一个订阅者，在当前订阅者未取消订阅前，其他订阅者都将无法订阅对应主题。

示例	前缀	真实主题名
$exclusive/t/1	$exclusive/	t/1

延迟发布

延迟发布可以实现按照用户配置的时间间隔延迟发布消息，当客户端使用特殊主题前缀 $delayed/{DelayInteval} 发布消息时，将触发延迟发布功能。

$delayed/{DelayInterval}/{TopicName}

# 15秒后将MQTT消息发布到主题x/y
$delayed/15/x/y

# 1 分钟后将 MQTT 消息发布到 a/b
$delayed/60/a/b

控制报文

PINGREQ —— 心跳请求

客户端发送 PINGREQ 报文给服务端的。用于：

1. 在没有任何其它控制报文从客户端发给服务的时，告知服务端客户端还活着。
2. 请求服务端发送 响应确认它还活着。
3. 使用网络以确认网络连接没有断开。

保持连接（Keep Alive）处理中用到这个报文

服务质量定义

Quality of Service

Qos值	描述
0	最多分发一次
1	至少分发一次
2	只分发一次

MQTT QoS 0, 1, 2 介绍

使用 QoS 0 可能丢失消息，使用 QoS 1 可以保证收到消息，但消息可能重复，使用 QoS 2 可以保证消息既不丢失也不重复。QoS 等级从低到高，不仅意味着消息可靠性的提升，也意味着传输复杂程度的提升。

在一个完整的从发布者到订阅者的消息投递流程中，QoS 等级是由发布者在 PUBLISH 报文中指定的，大部分情况下 Broker 向订阅者转发消息时都会维持原始的 QoS 不变。不过也有一些例外的情况，根据订阅者的订阅要求，消息的 QoS 等级可能会在转发的时候发生降级。例如，订阅者在订阅时要求 Broker 可以向其转发的消息的最大 QoS 等级为 QoS 1，那么后续所有 QoS 2 消息都会降级至 QoS 1 转发给此订阅者，而所有 QoS 0 和 QoS 1 消息则会保持原始的 QoS 等级转发。

QoS0 - 最多交付一次

QoS 0 是最低的 QoS 等级。QoS 0 消息即发即弃，不需要等待确认，不需要存储和重传

为什么Qos0消息会丢失？

当我们使用 QoS 0 传递消息时，消息的可靠性完全依赖于底层的 TCP 协议。

而 TCP 只能保证在连接稳定不关闭的情况下消息的可靠到达，一旦出现连接关闭、重置，仍有可能丢失当前处于网络链路或操作系统底层缓冲区中的消息。这也是 QoS 0 消息最主要的丢失场景。

QoS1 - 至少交付一次

为了保证消息到达，QoS 1 加入了应答与重传机制，发送方只有在收到接收方的 PUBACK 报文以后，才能认为消息投递成功，在此之前，发送方需要存储该 PUBLISH 报文以便下次重传。

1. Publisher在本地存储消息
2. 向Broker发布消息
3. Broker存储消息
4. Broker向Subscribe发布消息
5. Broker向Pushlisher返回PUBACK报文（发布确认）
6. Publisher删除本地消息
7. Subscriber向Broker返回发布确认报文
8. Broker删除本地消息

QoS 1 需要在 PUBLISH 报文中设置 Packet ID，而作为响应的 PUBACK 报文，则会使用与 PUBLISH 报文相同的 Packet ID，以便发送方收到后删除正确的 PUBLISH 报文缓存。

为什么QoS1消息会重复？

对于发送方来说，没收到 PUBACK 报文分为以下两种情况：

1. PUBLISH 未到达接收方
   • 发送方虽然重传了 PUBLISH 报文，但是对于接收方来说，实际上仍然仅收到了一次消息。
2. PUBLISH 已经到达接收方，接收方的 PUBACK 报文还未到达发送方
   • 在发送方重传时，接收方已经收到过了这个 PUBLISH 报文，这就导致接收方将收到重复的消息。

PUBLISH QoS=1, Packet ID=1024, DUP=0,Payload=I'am Alice

PUBLISH QoS=1, Packet ID=1024, DUP=1,Payload=I'am Alice

虽然重传时 PUBLISH 报文中的 DUP 标志会被设置为 1，用以表示这是一个重传的报文。但是接收方并不能因此假定自己曾经接收过这个消息(Case 1)，仍然需要将其视作一个全新的消息。

对于接收方来说，可能存在以下两种情况：

第一种情况，发送方由于没有收到 PUBACK 报文而重传了 PUBLISH 报文。此时，接收方收到的前后两个 PUBLISH 报文使用了相同的 Packet ID，并且第二个 PUBLISH 报文的 DUP 标志为 1，此时它确实是一个重复的消息。

第二种情况，第一个 PUBLISH 报文已经完成了投递，1024 这个 Packet ID 重新变为可用状态。发送方使用这个 Packet ID 发送了一个全新的 PUBLISH 报文，但这一次报文未能到达对端，所以发送方后续重传了这个 PUBLISH 报文。这就使得虽然接收方收到的第二个 PUBLISH 报文同样是相同的 Packet ID，并且 DUP 为 1，但确实是一个全新的消息。

由于我们无法区分这两种情况，所以只能让接收方将这些 PUBLISH 报文都当作全新的消息来处理。因此当我们使用 QoS 1 时，消息的重复在协议层面上是无法避免的。

eg.

如果我们不对 QoS 1 进行去重处理，我们可能会遭遇这种情况，发布方以 1、2 的顺序发布消息，但最终订阅方接收到的消息顺序可能是 1、2、1、2。如果 1 表示开灯指令，2 表示关灯指令，我想大部分用户都不会接受自己仅仅进行了开灯然后关灯的操作，结果灯在开和关的状态来回变化。

如何为 QoS 1 消息去重？

一个比较常用且简单的方法是，在每个 PUBLISH 报文的 Payload 中都带上一个时间戳或者一个单调递增的计数，这样上层业务就可以根据当前收到消息中的时间戳或计数是否大于自己上一次接收的消息中的时间戳或计数来判断这是否是一个新消息。

QoS2 - 只交付一次

QoS 2 解决了 QoS 0、1 消息可能丢失或者重复的问题，但相应地，它也带来了最复杂的交互流程和最高的开销。每一次的 QoS 2 消息投递，都要求发送方与接收方进行至少两次请求/响应流程。

回顾一下 QoS 1 消息无法避免重复的原因：

当我们使用 QoS 1 消息时，对接收方来说，回复完 PUBACK 这个响应报文以后 Packet ID 就重新可用了，也不管响应是否确实已经到达了发送方。所以就无法得知之后到达的，携带了相同 Packet ID 的 PUBLISH 报文，到底是发送方因为没有收到响应而重传的，还是发送方因为收到了响应所以重新使用了这个 Packet ID 发送了一个全新的消息。

所以，消息去重的关键就在于，通信双方如何正确地同步释放 Packet ID，换句话说，不管发送方是重传消息还是发布新消息，一定是和对端达成共识了的。

PUBREC - 发布收到（QoS 2，第一步）

PUBREL - 发布释放（QoS 2，第二步）

PUBCMP - 发布完成（QoS 2，第三步）

QoS 2 规定，发送方只有在收到 PUBREC 报文之前可以重传 PUBLISH 报文。一旦收到 PUBREC 报文并发出 PUBREL 报文，发送方就进入了 Packet ID 释放流程，不可以再使用当前 Packet ID 重传 PUBLISH 报文。同时，在收到对端回复的 PUBCOMP 报文确认双方都完成 Packet ID 释放之前，也不可以使用当前 Packet ID 发送新的消息。

因此，对于接收方来说，能够以 PUBREL 报文为界限，凡是在 PUBREL 报文之前到达的 PUBLISH 报文，都必然是重复的消息；而凡是在 PUBREL 报文之后到达的 PUBLISH 报文，都必然是全新的消息。

何时向后分发QoS2消息？

我们已经了解到，QoS 2 的流程是非常长的，为了不影响消息的实时性，我们可以在第一次收到 PUBLISH 报文时，就启动消息的向后分发。当然一旦开始向后分发，后续收到在 PUBREL 报文之前到达的 PUBLISH 报文，都不能再重复分发操作，以免消息重复。

选择

QoS 1 可以保证消息到达，所以适合传输一些较为重要的数据，比如下达关键指令、更新重要的有实时性要求的状态等。但因为 QoS 1 还可能会导致消息重复，所以当我们选择使用 QoS 1 时，还需要能够处理消息的重复，或者能够允许消息的重复

---

保留消息(Retained Messages)

含义：MQTT 客户端向服务器发布消息时，可以设置保留消息，标志在当前主题设置一条持久消息，消息将被保存在服务器上，新的订阅者订阅主题时将接收到该消息。

例如 MQTT X CLI 发布一条保留消息到主题 ‘a/b/c’:

mqttx pub -r -q 1 -t a/b/c -m 'hello' 

之后连接上来的 MQTT 客户端订阅主题 ‘a/b/c’ 时候，仍可收到该消息:

$ mqttx sub -t a/b/c -q 1
payload:  hello

删除保留消息

1. 客户端向有保留消息的主题发布一个空消息（客户端往某个主题发送一个 Payload 为空的保留消息，服务端就会删除这个主题下的保留消息）：

    mqttx pub -r -q 1 -t a/b/c -m ''
   

2. 超过 EMQX 设置的最大保留消息数。

3. 通过 EMQX 保留消息 REST API 删除。

4. 设置了消息过期间隔，到期后保留消息将被删除，适用于 MQTT 5.0。

MQTT 服务器会为每个主题存储最新一条保留消息，以方便消息发布后才上线的客户端在订阅主题时仍可以接收到该消息。

何时使用保留消息

发布订阅模式虽然能让消息的发布者与订阅者充分解耦，但也存在一个缺点，即订阅者无法主动向发布者请求消息。

借助保留消息，新的订阅者能够立即获取最近的状态：

• 智能家居设备的状态只有在变更时才会上报，但是控制端需要在上线后就能获取到设备的状态；
• 传感器上报数据的间隔太长，但是订阅者需要在订阅后立即获取到最新的数据；
• 传感器的版本号、序列号等不会经常变更的属性，可在上线后发布一条保留消息告知后续的所有订阅者。

---

遗嘱

遗嘱标志（Will Flag）被设置为 1，表示如果连接请求被接受了，遗嘱（Will Message）消息必须被存储在服务端并且与这个网络连接关联。之后网络连接关闭时，服务端必须发布这个遗嘱消息，除非服务端收到 DISCONNECT 报文时删除了这个遗嘱消息

遗嘱消息发布的条件，包括但不限于：

• 服务端检测到了一个 I/O 错误或者网络故障。

• 客户端在保持连接（Keep Alive）的时间内未能通讯。

• 客户端没有先发送 DISCONNECT 报文直接关闭了网络连接。

• 由于协议错误服务端关闭了网络连接。

如果遗嘱标志被设置为 1，连接标志中的 Will QoS 和 Will Retain 字段会被服务端用到，同时有效载荷中必须包含 Will Topic 和 Will Message 字段 。

一旦被发布或者服务端收到了客户端发送的 DISCONNECT报文，遗嘱消息就必须从存储的会话状态中移除。

如果遗嘱标志被设置为 0，连接标志中的 Will QoS 和 Will Retain 字段必须设置为 0，并且有效载荷中不能 包含 Will Topic 和 Will Message 字段 。

如果遗嘱标志被设置为 0，网络连接断开时，不能发送遗嘱消息 。 服务端应该迅速发布遗嘱消息。在关机或故障的情况下，服务端可以推迟遗嘱消息的发布直到之后的重启。 如果发生了这种情况，在服务器故障和遗嘱消息被发布之间可能会有一个延迟。

---

持久会话与Clean Session

不稳定的网络及有限的硬件资源是物联网应用需要面对的两大难题，MQTT 客户端与服务器的连接可能随时会因为网络波动及资源限制而异常断开。为了解决网络连接断开对通信造成的影响，MQTT 协议提供了持久会话功能。

持久会话需要存储的数据

通过上文我们知道持久会话需要存储一些重要的数据，以使会话能被恢复。这些数据有的存储在客户端，有的则存储在服务端。

客户端中存储的会话数据：

• 已发送给服务端，但是还没有完成确认的 QoS 1 与 QoS 2 消息。
• 从服务端收到的，但是还没有完成确认的 QoS 2 消息。

服务端中存储的会话数据：

• 会话是否存在，即使会话状态其余部分为空。
• 已发送给客户端，但是还没有完成确认的 QoS 1 与 QoS 2 消息。
• 等待传输给客户端的 QoS 0 消息（可选），QoS 1 与 QoS 2 消息。
• 从客户端收到的，但是还没有完成确认的 QoS 2 消息，遗嘱消息和遗嘱延时间隔。

Clean Session 的使用

Clean Session 是用来控制会话状态生命周期的标志位

1. 为 true 时表示创建一个新的会话，在客户端断开连接时，会话将自动销毁

2. 为 false 时表示创建一个持久会话，在客户端断开连接后会话仍然保持，直到会话超时注销。（能成功接收到离线期间的消息。）

注意： 持久会话能被恢复的前提是客户端使用固定的 Client ID 再次连接，如果 Client ID 是动态的，那么连接成功后将会创建一个新的持久会话。

持久会话的作用

1. 实现离线消息的接收
2. 降低订阅开销

MQTT 5.0中会话的改进

MQTT 5.0 中将 Clean Session 拆分成了 Clean Start 与 Session Expiry Interval。Clean Start 用于指定连接时是创建一个全新的会话还是尝试复用一个已存在的会话，Session Expiry Interval 用于指定网络连接断开后会话的过期时间。

Clean Start 为 true 时表示必须丢弃任何已存在的会话，并创建一个全新的会话；为 false 时表示必须使用与 Client ID 关联的会话来恢复与客户端的通信（除非会话不存在）。

Session Expiry Interval 解决了 MQTT 3.1.1 中持久会话永久存在造成的服务器资源浪费问题。设置为 0 或未设置，表示断开连接时会话即到期；设置为大于 0 的数值，则表示会话在网络连接关闭后会保持多少秒；设置为 0xFFFFFFFF 表示会话永远不会过期。