MQTT系列---入门介绍

糖爸的架构师之路 2021-06-24

1989

写在前面:

最近参与了公司的物联网项目开发，因此花了几天时间研究了一下MQTT协议相关的内容，也查阅了相关的资料和开源中间件的源码。在学习过程中，我发现目前网上这部分内容的资料不多并且内容较为零散。所以就有了写这个系列的想法，一方面想系统的梳理下MQTT的相关知识和设计思想，方便日后查看，另一方面如果有正在开发相关业务的小伙伴，也可以作为参考，不至于一头雾水。目前这个系列会涵盖

MQTT协议内容的基本概述以及入门使用
Spring-intergration整合Eclipse.paho作为MQTTClient实现消费订阅消息功能
Eclipse.paho内部消息流转的执行流程及源码解析
使用过程中遇到的问题汇总

这个是我写的第一篇公众号文章，关于内容有很多也是我自己通过阅读官方文档和查看源代码得到的理解。肯定会有存在有偏差的问题出现,欢迎小伙伴们随时留言指正。下面我们进入正题

一. 简述

目前市面上流行的消息协议很多，例如基于Java平台的JMS协议，基于应用层标准的AMQP高级消息队列协议，我们熟悉的RabbitMQ消息服务中间件就是基于AMQP协议的ERlang实现（之后我们搭建的MQTT消息服务也会基于RabbitMQ来实现，这个后面详细说）。那么MQTT协议到底是什么呢？下面我们来看一下。

MQTT（Message Queuing Telemetry Transport，消息队列遥测传输协议）是一种基于发布/订阅（publish/subscribe）模式的"轻量级"通讯协议，该协议构建于TCP/IP协议上，由IBM在1999年发布。MQTT最大优点在于，可以以极少的代码和有限的带宽，为连接远程设备提供实时可靠的消息服务。作为一种低开销、低带宽占用的即时通讯协议，使其在物联网、小型设备、移动应用等方面有较广泛的应用。

MQTT消息模型

通过上面的图示，我们可以很清晰的了解到MQTT采用的也是发布/订阅这种方式。与AMQP协议类似，它本身支持消息通过生产端发起，推送到消息服务，消费端通过订阅的方式统一消费的模式。消费端可以根据自己的需要通过订阅不同的主题（topic）来消费对应的消息。这种类似“推模式”的方式既可以保证消息的及时性，也可以减小消费端和Broker之间的通信压力。

二、协议原理

2.1 协议的实现方式

实现MQTT协议需要客户端和服务器端通讯完成。在通讯过程中，MQTT协议中有三种身份

发布者（Publish）
代理（Broker）（服务器）
订阅者（Subscribe）

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

MQTT传输的消息分为两个部分

主题（Topic）
负载（payload）

2.2 MQTT客户端

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

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

2.3 MQTT服务器

MQTT服务器也称为"消息代理"（Broker），可以是一个应用程序或一台设备。它是位于消息发布者和订阅者之间，可以实现下面的功能

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

MQTT在通信时会构建底层网络传输，它将建立客户端到服务器的连接，提供两者之间的一个有序的、无损的、基于字节流的双向传输。当应用数据通过MQTT网络发送时，MQTT会把与之相关的服务质量（QOS）和主题名（Topic）相关联。

2.4 MQTT协议中的订阅、主题、会话

订阅（Subscription）

订阅包含主题筛选器（Topic Filter）和最大服务质量（QOS）。订阅会与一个会话（Session）关联。一个会话可以包含多个订阅。每个会话中的每个订阅都有一个不同的主题筛选器

2. 会话（Session）

每个客户端与服务器建立连接后就是一个会话，客户端和服务器之间有状态交互。会话存在于一个网络之间，也可能在客户端和服务器之间跨越多个连续的网络连接。

3. 主题名（Topic Name）

连接到一个应用程序消息的标签，该标签与服务器的订阅相匹配。服务器会将消息发送给订阅所匹配标签的每个客户端。

4. 主题筛选器（Topic Filter）

一个对主题名通配符筛选器，在订阅表达式中使用，表示订阅所匹配到的多个主题。

5. 负载

消息订阅者所具体接收的内容。

2.5 MQTT协议中的方法

MQTT协议中定义了一些方法（也被称为动作），用于表示对确定资源所进行操作。这个资源可以代表预先存在的数据或动态生成数据，这取决于服务器的实现。通常来说，资源指服务器上的文件或输出。主要方法有：

Connect：等待与服务器建立连接。
Disconnect：等待MQTT客户端完成所做的工作，并与服务器断开TCP/IP会话。
Subscribe：等待完成订阅。
UnSubscribe：等待服务器取消客户端的一个或多个topics订阅。
Publish：MQTT客户端发送消息请求，发送完成后返回应用程序线程。

三、MQTT协议数据包结构

在MQTT协议中，一个MQTT数据包由以下三部分内容组成

固定头（Fixed header）：存在于所有MQTT数据包中，表示数据包类型及数据包的分组类标识。
可变头（Variable header）：存在于部分MQTT数据包中，数据包类型决定了可变头是否存在及其具体内容。
消息体（Payload）：存在于部分MQTT数据包中，表示客户端收到的具体内容。

3. 1 MQTT固定头

每个MQTT控制报文都包含一个固定报头，固定头部占用两个字节。其内存结构大概如下图所示

固定头内存结构

3.1.1 控制报文类型

用来描述报文的类型，通常用于客户端和服务端进行双向通信时对报文进行分类，不同的报文类型处理的方式也不同。

3.1.2 重发表示 DUP FLAG：

发送消息的副本。如果DUP标志被设置为0，表示这是客户端或者服务端第一次请求发送这个PUBLISH报文。如果DUP标志被设置为1，表示这可能是一个早期报文请求的重发。但不能用于检测消息重复发送

3.1.3 消息质量QOS（Quality of Service）

发送者和接收者之间，对于消息传递的可靠程度的协商。目前有三种消息发布的服务质量

QOS= 0："至多一次"，对于client而言，有且仅发一次publish包，对于broker而言，有且仅发一次publish。简而言之，就是仅发一次包，是否收到完全不管，适合那些不是很重要的数据。

QOS= 1：确保消息到达，但消息重复可能会发生。当broker收到client的publisher或者subscriber收到broker的publisher时，都会产生一条puback用来确认到达。当client没收到broker的puback或者broker没有收到client的puback，name就会一直发送publisher

QOS= 2：确保消息到达一次。publisher和broker分别进行了缓存，其中publisher缓存了message和messageId，而broker缓存了messageId，两方都做了记录。所以可以保证消息不重复，但同时因为增加了两次通信交互，对网络带宽和服务器负载也会产生一些压力

注意：对于QOS2的情况，增加了一个PUBREL及PUBCOM的过程，同时broker端做了如下的特殊处理。

如果Publisher没收到Broker的PUBRECV,Publisher会重发，但是对于之前一条message，Broker有两种处理方式：

message存在本地，先不给publish和subscriber
保存messageId，把message publish给他的subscriber

对于第一种处理方式，如果Publisher继续重发，且被收到（ID相同），那在Broker端只算一条message，继续等Publisher发PUBREL。这样broker就保证只publish了一条message，而不是多条。

对于第二种处理方式，如果sender继续重发，且被收到，sever会检查它的message ID，如果重发过来的message ID是之前存过的，broker就不会publish给他的subscriber，因为之前已经publish了，直接删除。不同的消息服务针对QOS2的情况会选择不同的处理方式。针对RabbitMQ而言，本身不支持QOS2的处理方式，会自动降级为QOS1进行处理。所以在这里针对这个问题我们不展开讨论。

3.1.4 retain（保留消息）

发布保留标识，表示服务器要保留这次推送的信息，如果有新的订阅者出现，就把这消息推送给它，如果不设那么推送至当前订阅的就释放了。这里需要和消息积压处理区分开，通常如果publisher生产一条消息到broker，如果订阅的消费者此时与broker断开连接，在cleanSession设置为true的情况下（这个属性后面会详细讲），消费者重新连接后是无法收到这些在自己断开时被生产的消息的。这与我们经常使用的AMQP有很大的不同。而保留消息的意思是生产端如果生产一条保留消息，这条消息会一直缓存在broker端，每次消费端断开重连，都会收到这条消息。但是保留消息在broker有且仅有一条。也就是说，后发送的保留消息会覆盖掉先发送的保留消息。消费端连接成功后收到的保留消息，永远是最后一条发送给broker的保留消息。

3.2 可变报头

MQTT数据包中包含一个可变头，它驻位于固定的头和负载之间。可变报头的结构会根据数据包类型的不同而不同。内容也会有相应的变化。这里要注意下，网上大部分资料通常会把CONN类型或者Publish类型的消息可变头作为通用的可变头结构理解。其实不存在通用可变头结构的概念，不同类型数据包，携带的信息都是不一样的。这里因为篇幅关系，我就简单介绍一下可变头中比较重要的信息属性。

3.2.1 Clean Session

这里就是刚才在介绍保留消息时提到的属性。包含在CONN类型可变头中，表示如果订阅的客户机断线了，那么要保存其要推送的消息，如果其重新连接时，则将这些消息推送。1表示消除，表示客户机是第一次连接，消息所以以前的连接信息。

3.2.2 遗嘱消息

当client发送一条CONN类型的报文给服务器时，可以在可变报头中包含遗嘱消息。当client与broker断开时，broker会主动向消费订阅端推送这条遗嘱消息。遗嘱消息的发送条件包括

服务端检测到了一个I/O错误或者网络故障。
客户端在保持连接（Keep Alive）的时间内未能通讯。
客户端没有先发送DISCONNECT报文直接关闭了网络连接。
由于协议错误服务端关闭了网络连接。

遗嘱消息本质也是一条保留消息，也有自己的消息质量（QOS）。也就是说，如果遗嘱消息被发送时，会作为最新的一条保留消息替换掉之前的保留消息。遗嘱消息和保留消息配合使用可以作为设备端通知控制端在线或者离线的通信手段，网上有很多解决方案，这里不再赘述。

3.2.3 主题名（Topic Name）

用于识别有效载荷数据应该被发布到哪一个信息通道。存在于PUBLISH（QOS>0时）， PUBACK，PUBREC，PUBREL，PUBCOMP，SUBSCRIBE,SUBACK，UNSUBSCIBE，UNSUBACK类型的报文中。

这里顺便说明一下Topic的过滤器，也就是所谓的主题过滤器的匹配规则。

我们先来简单回顾下，使用过RabbitMQ消息服务的小伙伴应该都了解，在AMQP协议下，当使用publisher/subscriber模式时，可以使用topic模式模糊监听消息队列。例如，producter发送一条“hello world”的消息到指定路由，routingKey为mqtt.abc.def。此时有两个consumer同时在消费，consumerA监听队列queue1，bindingkey为mqtt.#,consumerB监听队列queue2,bindingKey为mqtt.*，此时收到消息的应该是comsumerA,因为consumerA的bindingKey格式为mqtt.#，即以mqtt开头的routingkey无论有几个分段标识符（即多层分隔符），都会被监听到。而consumerB只能监听到单层分隔符。这是一些最基本的知识，但是我为什么要先说这个呢？原因就在于mqtt的主题过滤器与amqp协议下的主题过滤器非常类似，区别就在于把 "." 更换成了 “#”。

举个栗子：如果客户端订阅主题 “sport/tennis/player1/#”，它会收到使用下列主题名发布的消息：

sport/tennis/player1
sport/tennis/player1/ranking
sport/tennis/player1/score/wimbledon

值得注意的是：“sport/#”也匹配单独的 “sport” ，因为 # 包括它的父级。并且#只能放在订阅主题名称的/后，并且在末尾的位置，否则匹配无效。例如

sport/tennis#
sport/tennis/#/ranking

上面两种监听方式都是无效的！！

对应AMQP的单层通配符 “*”，MQTT协议使用的是 “+”，并且+只能用于单层通配符。在主题过滤器的任意层级都可以使用单层通配符，包括第一个和最后一个层级。然而它必须占据过滤器的整个层级。可以在主题过滤器中的多个层级中使用它，也可以和多层通配符一起使用。

再举个栗子

“+” 是有效的。
“+/tennis/#” 是有效的。
“sport+” 是无效的。
“sport/+/player1” 也是有效的。
“/finance” 匹配 “+/+” 和 “/+” ，但是不匹配 “+”。

另外：服务端不能将 $ 字符开头的主题名匹配通配符 (#或+) 开头的主题过滤器

订阅 “#” 的客户端不会收到任何发布到以 “$” 开头主题的消息。
订阅 “+/monitor/Clients” 的客户端不会收到任何发布到 “$SYS/monitor/Clients” 的消息。
订阅 “$SYS/#” 的客户端会收到发布到以 “$SYS/” 开头主题的消息。
订阅 “$SYS/monitor/+” 的客户端会收到发布到 “$SYS/monitor/Clients” 主题的消息。
如果客户端想同时接受以 “$SYS/” 开头主题的消息和不以 $ 开头主题的消息，它需要同时订阅 “#” 和 ““$SYS/#”。

除此之外：主题名和主题过滤器还必须符合下列规则

所有的主题名和主题过滤器必须至少包含一个字符 [MQTT-4.7.3-1]。
主题名和主题过滤器是区分大小写的。
主题名和主题过滤器可以包含空格。
主题名或主题过滤器以前置或后置斜杠 “/” 区分。
只包含斜杠 “/” 的主题名或主题过滤器是合法的。
主题名和主题过滤器不能包含空字符
主题名和主题过滤器是UTF-8编码字符串，它们不能超过65535字节

3.2.4 报文标识符 Packet Identifier

报文标识符用来区分报文，特别是在重发的报文中用来标识是否是同一个报文，并在需要应答的场景中用于确定是对哪个发送报文的应答。可变报头的报文标识符（Packet Identifier）字段存在于在多个类型的报文里。

客户端每次发送一个新的这些类型的报文时都必须分配一个当前未使用的报文标识符。如果一个客户端要重发这个特殊的控制报文，在随后重发那个报文时，它必须使用相同的标识符。当客户端处理完这个报文对应的确认后，这个报文标识符就释放可重用。QOS1的PUBLISH对应的是PUBACK，QOS2的PUBLISH对应的是PUBCOMP。QOS设置为0的PUBLISH报文不能包含报文标识符。

值得注意的是，假设客户端发送标识符为0x1234的Publish报文，它有可能会在收到那个报文的PUBACK之前，先收到服务端发送的另一个消息不同但是报文标识符相同的Publish报文。但只要保证在一条消息中Publish和PubACK有唯一的映射关系就可以

3.3 有效负载（payload）

Payload消息体是MQTT数据包的第三部分。包含CONNECT、SUBSCRIBE、SUBACK、UNSUBSCRIBE四种类型的消息。也就是说，这四种控制报文的消息体是必须要有的

CONNECT：消息体内容主要是：客户端的ClientID、订阅的Topic、Message以及用户名和密码。
SUBSCRIBE：消息体内容是一系列的要订阅的主题以及QOS。
SUBACK：消息体内容是服务器对于SUBSCRIBE所申请的主题及QOS进行确认和回复。
UNSUBSCRIBE：消息体内容是要订阅的主题。

而publish是消息体中则保存推送的消息，以二进制形式，当然这里的编辑可自定义。

以下是各个控制报文对消息体的规则

以上就是MQTT协议中的较为常用和核心的内容，当然还有一些其他边边角角的东西。例如各个控制报文之间数据结构的异同，连接方式，安全认证等。这些内容都可以在官方文档中找到，感兴趣的小伙伴可以直接参考官方文档

传送门：http://mqtt.org/

结束语

本章节主要介绍了MQTT协议的基本构成，通信方式，控制报文格式以及数据包的组成部分。下一章会从代码的角度基于Spring-intergration和Eclipse.Paho中间件和RabbitMQ消息服务搭建一套完整的消息发送接收功能。敬请期待~~~

数据库

文章转载自糖爸的架构师之路，如果涉嫌侵权，请发送邮件至：contact@modb.pro进行举报，并提供相关证据，一经查实，墨天轮将立刻删除相关内容。

MQTT系列---入门介绍

评论