简介

MQTT（Message Queuing Telemetry Transport，消息队列遥测传输协议），是一种基于发布/订阅（publish/subscribe）模式的“轻量级”通讯协议，该协议构建于TCP/IP协议上;

主流的MQTT是基于TCP连接进行数据推送的，但是同样有基于UDP的版本，叫做MQTT-SN。这两种版本由于基于不同的连接方式，优缺点自然也就各有不同了。除标准版外，还有一个简化版MQTT-SN，该协议主要针对嵌入式 (opens new window)设备，这些设备一般工作于TCP/IP网络，如：ZigBee。

MQTT最大优点: 小型传输，开销很小（固定长度的头部是2字节[byte1(8位),byte2(8位)]）协议交换最小化，以降低网络流量。可以以极少的代码和有限的带宽，为连接远程设备提供实时可靠的消息服务。

MQTT 5.0 协议新增介绍

MQTT 5.0 协议相比 MQTT 3.1.1 协议新增了许多内容， 比如说属性，AUTH 包，还有对一些字段做了修改，比如将 Clean Session 修改成 Clean Start 配合 Session Expiry Internal 去实现更灵活的会话控制。

这里就简单罗列一下 5.0 协议新增的内容。

• 增强了扩展性
• 改善了错误报告的方式
• 定型了一些通用范式，例如能力发现和请求、响应
• 扩展机制包括用户属性(user properties)
• 性能改善，并且添加了对小客户端（small clients） 的支持

原因码: MQTT v3.1.1 只有寥寥 6 个返回码，用来表示网络连接时可能会出现的异常行为，在引入属性后的 MQTT 5.0 协议中，仅仅这 6 个返回码显然已经不足以用来描述各种异常行为，因此MQTT 5.0 协议中将返回码改成了原因码，用来实现改善错误报告的目的。

基本概念

1. MQTT 客户端

   一个使用 MQTT 协议的设备、应用程序等，它总是建立到服务器的网络连接。

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

2. MQTT 服务器

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

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

3. 主题（Topic）

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

   • 要订阅的主题。一个主题可以有多个级别，级别之间用斜杠字符分隔。例如，/world 和 emq/emqtt/emqx 是有效的主题。
   • 订阅者的Topic name支持通配符#和+ ：
     • # 支持一个主题内任意级别话题
     • +只匹配一个主题级别的通配符
   • 客户端成功订阅某个主题后，代理会返回一条 SUBACK 消息，其中包含一个或多个 returnCode 参数

4. 主题筛选器（Topic Filter）

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

5. QoS（消息传递的服务质量水平）

   服务质量，标志表明此主题范围内的消息传送到客户端所需的一致程度。

   • 值 0：不可靠，消息基本上仅传送一次，如果当时客户端不可用，则会丢失该消息。
   • 值 1：消息应传送至少 1 次。
   • 值 2：消息仅传送一次。

6. 会话（Session）

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

7. 订阅（Subscription）

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

   • 客户端在成功建立TCP连接之后，发送CONNECT消息，在得到服务器端授权允许建立彼此连接的CONNACK消息之后，客户端会发送SUBSCRIBE消息，订阅感兴趣的Topic主题列表（至少一个主题）
   • 订阅的主题名称采用UTF-8编码，然后紧跟着对应的QoS值

8. 发布（publish）

   控制报文是指从客户端向服务端或者服务端向客户端传输一个应用消息，MQTT 客户端发送消息请求，发送完成后返回应用程序线程

   • 比如安卓的推送服务，还有一些即时通信软件如微信等也是采用的推送技术。

9. 负载（Payload）

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

MQTT协议的工作方式

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

MQTT传输的消息分为：主题（Topic）和负载（payload）两部分：

（1）Topic，可以理解为消息的类型，订阅者订阅（Subscribe）后，就会收到该主题的消息内容（payload）；

（2）payload，可以理解为消息的内容，是指订阅者具体要使用的内容。

除了发布者和订阅者之间传递普通消息，代理还可以为发布者处理保留消息和遗愿消息**，并可以更改服务质量（QoS）等级。**

Last Will：即遗言机制，用于通知同一主题下的其他设备发送遗言的设备已经断开了连接。用于判断是否设备在线

MQTT协议数据包结构

主流的MQTT协议工作在TCP之上，端和代理之间通过交换预先定义的控制报文来完成通信。

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

**（1）固定头（Fixed header）**存在于所有MQTT数据包中，表示数据包类型及数据包的分组类标识。

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

**（3）消息体（Payload）**存在于部分MQTT数据包中，表示客户端收到的具体内容。

MQTT报文有3个部分组成，并按下表顺序出现：

固定报头（fixed header）	可变报头（variable header）	荷载（payload）
所有报文都包含	部分报文包含	部分报文包含

所有的MQTT控制报文都有一个固定报头，期格式如下：

固定头**固定长度的头部是2字节[byte1(8位),byte2(8位)]**存在于所有MQTT数据包中，其结构如下：

解释说明：

MQTT数据包类型:位置：Byte 1中bits 7-4。相于一个4位的无符号值

标识位

位置：Byte 1中bits 3-0。

在不使用标识位的消息类型中，标识位被作为保留位。如果收到无效的标志时，接收端必须关闭网络连接：

（1）DUP：发布消息的副本。用来在保证消息的可靠传输，如果设置为1，则在下面的变长中增加MessageId，并且需要回复确认，以保证消息传输完成，但不能用于检测消息重复发送。

（2）QoS：发布消息的服务质量，即：保证消息传递的次数

​ Ø00：最多一次，即：<=1

​ Ø01：至少一次，即：>=1

​ Ø10：一次，即：=1

​ Ø11：预留

（3）RETAIN： 发布保留标识，表示服务器要保留这次推送的信息，如果有新的订阅者出现，就把这消息推送给它，如果设有那么推送至当前订阅者后释放。

剩余长度（Remaining Length）

地址：Byte 2。

固定头的第二字节用来保存变长头部和消息体的总大小的，但不是直接保存的。这一字节是可以扩展，其保存机制，前7位用于保存长度，后一位用做标识。当最后一位为1时，表示长度不足，需要使用二个字节继续保存

MQTT可变头

MQTT数据包中包含一个可变头，它驻位于固定的头和负载之间。可变头的内容因数据包类型而不同，较常的应用是作为包的标识：

很多类型数据包中都包括一个2字节的数据包标识字段，这些类型的包有：

PUBLISH (QoS > 0)、PUBACK、PUBREC、PUBREL、PUBCOMP、SUBSCRIBE、SUBACK、UNSUBSCRIBE、UNSUBACK。

MQTT传输的消息

在通讯过程中，MQTT协议中有三种身份：发布者（Publish）、代理（Broker）（服务器）、订阅者（Subscribe）。其中，消息的发布者和订阅者都是客户端，消息代理是服务器，消息发布者可以同时是订阅者。

MQTT传输的消息分为：主题（Topic）和负载（payload）两部分；

Topic

匹配规则

主题(Topic)通过 / 分割层级，支持 +， # 通配符: 注: 单层通配符和多层通配符只能用于订阅(subscribe)消息而不能用于发布(publish)消息，层级分隔符两种情况下均可使用。

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

Payload消息体(负载)

MQTT传输的消息

Payload消息体位MQTT数据包的第三部分，包含CONNECT、SUBSCRIBE、SUBACK、UNSUBSCRIBE等类型的消息：

（1）CONNECT，消息体内容主要是：客户端的ClientID、订阅的Topic、Message以及用户名和密码。

（2）SUBSCRIBE，消息体内容是一系列的要订阅的主题以及QoS。

（3）SUBACK，消息体内容是服务器对于SUBSCRIBE所申请的主题及QoS进行确认和回复。

（4）UNSUBSCRIBE，消息体内容是要订阅的主题。

控制报文Control Packet

协议版本3定义了14种MQTT报文，用于建立/断开连接、发布消息、订阅消息和维护连接。

固定报头的第一字节的4-7位的值指定了报文类型，其取值如下表。0和15为系统保留值；0-3位为标志位，依照报文类型有不同的含义，

事实上，除了PUBLISH报文以外，其他报文的标志位均为系统保留。如果收到报文的标志位无效，代理应断开连接。

通过网络连接发送的信息数据包,也是通讯连接过程。MQTT规范定义了十四种不同类型的控制报文，其中一个（PUBLISH报文）用于传输应用消息; 常用的有五种(加粗处理):

报文类型	值	描述
CONNECT	1	客户端向代理发起连接请求
CONNACK	2	连接确认
PUBLISH	3	发布消息
PUBACK	4	发布确认
PUBREC	5	发布收到（QoS2）
PUBREL	6	发布释放（QoS2）
PUBCOMP	7	发布完成（QoS2）
SUBSCRIBE	8	客户端向代理发起订阅请求
SUBACK	9	订阅确认
UNSUBSCRIBE	10	取消订阅
UNSUBACK	11	取消订阅确认
PINGREQ	12	PING请求
PINGRESP	13	PING响应
DISCONNECT	14	断开连接

固定报头的第二字节起表示报文的剩余长度。最大4个字节，每字节可以编码至127，并含有一位继续位，如继续位非0，则下一字节依然为剩余长度。由此，理论上一个控制报文最长可以到256MB。

一些报文在固定报头和荷载之间可以有一个可变报头。可变报头的内容根据报文类型不同而不同。最常见的可变报头是报文标识符（PacketIdentifier）。

一些报文可以在最后携带一个荷载。不同的报文可以无荷载，可选荷载，或必须带有荷载。

MQTT协议中的方法

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

（1）Connect。等待与服务器建立连接。

（2）Disconnect。等待MQTT客户端完成所做的工作，并与服务器断开TCP/IP会话。

（3）Subscribe。等待完成订阅。

（4）UnSubscribe。等待服务器取消客户端的一个或多个topics订阅。

（5）Publish。MQTT客户端发送消息请求，发送完成后返回应用程序线程。

三种消息发布服务质量

• QoS 0：最多收到一次：数据包被发送，就是这样。没有确认是否已收到。

• QoS 1：至少接收一次：只要客户端未收到服务器的确认，就会发送和存储数据包。MQTT确保它将被接收，但可能存在重复。

• QoS 2：只收到一次：与QoS 1相同，但没有重复。

关于数据消耗，显然，QoS 2> QoS 1> QoS 0，如果这是您关心的问题。

QoS Level 0：至多一次

这个我理解是如果接收方离线了就不能收到消息，可以用在音视频聊天请求，因为当接收方离线后就不用收到请求了，就算是接收方在线但是没有收到消息也可以通过发送方超时来重发请求。

QoS Level 1：至少一次，有可能重复

这个可以用在普通文本聊天， 接收方离线后，服务器自动缓存消息，等接收方上线时服务器马上把消息推送给他，就算是接收方重复收包也没关系因为可以通过消息里包含的时间来过滤掉。这个级别的消息服务器得注意限制发送方的消息大小和数量，免得服务器内存被爆掉。

QoS Level 2：只有一次，确保消息只到达一次

这个用在消息重要性比较严格的场合。IM在一般情况下用不着，或者在用户发生金钱消费有关的情况下可以使用

MQTT中间broker

MQTT服务器(Broker)非常多，如apache的ActiveMQ，emtqqd，HiveMQ，Emitter，Mosquitto，Moquette, Mosquitto,RabbitMQ,RocketMQ,等等。轻量级的mosquitto开源项目是MQTT服务器。

Mosquitto

部署

Docker镜像非常小,及使用的人非常多;推荐;[1.6]

docker pull eclipse-mosquitto:1.6
docker pull eclipse-mosquitto

启动

Three directories have been created in the image to be used for configuration, persistent storage and logs.
/mosquitto/config
/mosquitto/data
/mosquitto/log
cd /mosquitto
docker run -it -p 1883:1883 -p 9001:9001 \
    -v /config/mosquitto.conf:/mosquitto/config/mosquitto.conf \
    -v /mosquitto/data  \
    -v /mosquitto/log  \
    --name mosquitto \
    eclipse-mosquitto
docker run -it -p 1883:1883 -p 9001:9001 -v /mosquitto/data -v /mosquitto/log --name mosquitto eclipse-mosquitto
docker run --restart=always -p 1883:1883 -p 9001:9001 -v /mosquitto/data -v /mosquitto/log --name mosquitto eclipse-mosquitto

#最后使用;start.sh
sudo docker run -d --restart=always  -p 1883:1883 -p 9001:9001 -v /home/samy/mosquitto/mosquitto.conf:/mosquitto/config/mosquitto.conf -v /home/samy/mosquitto/data:/mosquitto/data -v /home/samy/mosquitto/log:/mosquitto/log eclipse-mosquitto

mosquitto.conf:

persistence true
persistence_location /mosquitto/data/
log_dest file /mosquitto/log/mosquitto.log

参照：

https://mosquitto.org/man/mosquitto-8.html

https://github.com/toke/docker-mosquitto/blob/master/config/mosquitto.conf.example

# Place your local configuration in /mqtt/config/conf.d/
pid_file /var/run/mosquitto.pid

persistence true
persistence_location /mqtt/data/

user mosquitto
# Port to use for the default listener.
port 1883

log_dest file /mqtt/log/mosquitto.log
log_dest stdout

include_dir /mqtt/config/conf.d

other case:

mkdir -p /srv/mqtt/config/
mkdir -p /srv/mqtt/data/
mkdir -p /srv/mqtt/log/
# place your mosquitto.conf in /srv/mqtt/config/
# NOTE: You have to change the permissions of the directories
# to allow the user to read/write to data and log and read from
# config directory
# For TESTING purposes you can use chmod -R 777 /srv/mqtt/*
# Better use "-u" with a valid user id on your docker host

# Copy the files from the config directory of this project
# into /src/mqtt/config. Change them as needed for your
# particular needs.

docker run -ti -p 1883:1883 -p 9001:9001 \
-v /srv/mqtt/config:/mqtt/config:ro \
-v /srv/mqtt/log:/mqtt/log \
-v /srv/mqtt/data/:/mqtt/data/ \
--name mqtt toke/mosquitto

参数介绍

安装成功后，在终端运行mosquitto_sub --help 或 mosquitto_pub —help获取脚本参数说明：

mosquitto_pub参数说明：

-d  打印debug信息
-f  将指定文件的内容作为发送消息的内容
-h  指定要连接的域名  默认为localhost
-i  指定客户端clientid，默认为附加进程ID的mosquitto_pub_
-I  指定clientId前缀
-m  消息内容
-n  发送一个空（null）消息
-p  连接端口号
-q  指定QoS的值（0,1,2）
-t  指定topic
-u  用户名
-P  用户密码    
-V  指定MQTT协议版本
--will-payload  指定一个消息，该消息当客户端与broker意外断开连接时发出。该参数需要与--will-topic一起使用
--will-qos  Will的QoS值。该参数需要与--will-topic一起使用
--will-retain 指定Will消息被当做一个retain消息（即消息被广播后，该消息被保留起来）。该参数需要与--will-topic一起使 用
--will-topic  用户发送Will消息的topic

---

mosquitto_sub参数说明

-c  指定客户端clean_session是否保存。
-d  打印debug信息
-h  指定要连接的域名  默认为localhost
-i   指定客户端clientid
-I   指定clientId前缀
-k  keepalive 每隔一段时间，发PING消息通知broker，仍处于连接状态。 默认为60秒.
-q 指定希望接收到QoS为什么的消息  默认QoS为0
-R 不显示陈旧的消息
-t 订阅topic
-v 打印消息
--will-payload  指定一个消息，该消息当客户端与broker意外断开连接时发出。该参数需要与--will-topic一起使用
--will-qos  Will的QoS值。该参数需要与--will-topic一起使用
--will-retain 指定Will消息被当做一个retain消息（即消息被广播后，该消息被保留起来）。该参数需要与--will-topic一起使>用
--will-topic  用户发送Will消息的topic

常用命令

#订阅主题
mosquitto_sub -v -t sensor
【-t】指定主题，此处为sensor
【-v】打印更多的调试信息
#发布内容
mosquitto_pub -t sensor -m 12
【-t】指定主题
【-m】指定消息内容
command line oper:
mosquito
mosquitto_sub -t 'test/topic' -v
mosquitto_pub -t 'test/topic' -m 'hello world'

sudo apt install mosquitto-clients 
#mosquitto [-c config file] [ -d | --daemon ] [-p port number] [-v]
mosquitto -c mosquitto.conf //启动
//1）订阅主题 -v:打印更多调式信息 -t：指定主题，此处为topicTest01
mosquitto_sub -t topicTest01 -v
//2）发布内容 -t:指定主题 -m：指定消息内容
mosquitto_pub -t topicTest01 -m TestMessage

#mosquitto重启
ps -aux | grep mosquitto //查看mosquitto的进程并kill掉；
mosquitto -c /etc/mosquitto/mosquitto.conf -d //启动

初始化设置

//egg-emqtt默认连接配置
const mqttClient = mqtt.connect(config.host, {
    clientId: config.clientId,
    username: config.username,
    password: config.password,
    keepalive: 60,
    protocolId: 'MQTT',
    protocolVersion: 4,
    clean: true,
    reconnectPeriod: 1000,
    connectTimeout: 30 * 1000,
    rejectUnauthorized: false,
    ...config.options,
});
//mqtt && emqx
config.mqtt = {
    address: 'mqtt://127.0.0.1:11883',
    queueSize: 4,
    topic: '$queue/#',
    option: {
        username: 'super_user_client',
        password: '_this_is_secrect_',
    },
}
config.emqx = {
    mgmtBaseUrl: 'http://127.0.0.1:8080/v3',
    appId: 'egg_iot_with_mqtt',
    appSecret: 'Mjg1OTEzMzAyMTI2MzA0NTUzMTkwMjcyMjIzMDg5Nzg2ODI',
}

消息封装

async messagePublish(topic, payload, _option = {}) {
    const option = Object.assign({}, { qos: 1, retain: false }, _option)
    return await this.ctx.helper.emqxAPI().post('/mqtt/publish', {
      topic,
      payload,
      qos: option.qos,
      retain: option.retain,
    })
  }

设备上下线

• 1:正常的上下线通知都是由客户端主动发送，即：建立mqtt连接之后的第一条pub消息和断开mqtt连接之前的最后一条pub消息；

• 2:对于异常断开的连接使用MQTT的遗嘱机制让mqtt broker来自动发送异常下线消息;连接时就得先遗嘱事件；

• 3:再加:主动询问是否设备在线;

自带的监听事件，没有带相关信息,没有采用：mosquitto $SYS下topic ： $SYS/broker/clients/connected

实现：

mosquitto_sub --will-topic cn/kinfo/device/will --will-payload "mac:11223344" -t cn/kinfo/device/will  -i sub -h 39.108.58.xxx:1883
mosquitto_sub --will-topic cn/kinfo/device/will --will-payload "mac:11223344" -t cn/kinfo/device/will -h 39.108.58.xxx
//修改后,调整
 route('cn/kinfo/device/will', controller.devices.deviceWill)

//在线,下线,will时,存储到redis hashf类型记录下来;
 let { mac: macStr } = ctx.req.message
 let mac = ctx.helper.strToMac(macStr)
 await app.redis.hset('deviceList', mac, '{ "status": 1 }')
 await app.redis.hdel('deviceList', mac)

//获取设备列表查询下,再返回给接口;
  async index () {
    const { ctx, service, app } = this
    const res = await service.device.listByCtx()
    const devices = res.rows
    for (const device of devices) {//主动检查指定设备是否在线
      await app.emqtt.publish(`cn/kinfo/device/statusCheck/${device.mac}`, { qos: 0, retain: false })
        await app.emqtt.publish(`cn/kinfo/${device.mac}/status`, { qos: 0, retain: false })
    }
    ctx.success(res)
  }
//获取当前在线的设备列表:
 const deviceList = await app.redis.hgetall('deviceList')
    let arrList = []
    for (let [key, value] of Object.entries(deviceList)) {
      arrList.push({
        mac: key,
        info: JSON.parse(value)
      })
    }

**ps:**在kInfo项目中,由于设备没有15s就会息屏,下线断网,不好频繁操作上报,故现在暂时不做处理;最后还是修改了上下线功能；

微消息队列 MQTT

阿里云，微消息队列 MQTT 是专为物联网（IoT）、移动互联网（MI）领域设计的消息产品，覆盖互动直播、金融支付、智能餐饮、即时聊天、移动 Apps、智能设备、车联网等多种应用场景；通过对 MQTT、WebSocket 等协议的全面支持，连接端和云之间的双向通信，实现 C2C、C2B、B2C 等业务场景之间的消息通信，可支撑千万级设备与消息并发，真正做到万物互联。

RocketMQ

RcoketMQ 是一款低延迟、高可靠、可伸缩、易于使用的消息中间件。具有以下特性：

1. 支持发布/订阅（Pub/Sub）和点对点（P2P）消息模型
2. 在一个队列中可靠的先进先出（FIFO）和严格的顺序传递
3. 支持拉（pull）和推（push）两种消息模式
4. 单一队列百万消息的堆积能力
5. 支持多种消息协议，如 JMS、MQTT 等
6. 分布式高可用的部署架构,满足至少一次消息传递语义
7. 提供 docker 镜像用于隔离测试和云集群部署
8. 提供配置、指标和监控等功能丰富的 Dashboard

MQTT 与 RocketMQ 的应用场景对比

产品名	适用场景
微消息队列 MQTT	面向移动端场景，移动端场景一般都具备海量设备，单设备数据较少的特点。因此，微消息队列 MQTT 适用于拥有大量在线客户端（很多企业设备端过万，甚至上百万），但每个客户端消息较少的场景。
消息队列 RocketMQ	面向服务端的消息引擎，主要用于服务组件之间的解耦、异步通知、削峰填谷等，服务器规模较小（极少企业服务器规模过万），但需要大量的消息处理，吞吐量要求高。因此，消息队列 RocketMQ 适用于服务端进行大批量的数据处理和分析的场景。

分析调试工具

MQTT-Explorer

参考链接

http://mosquitto.org/

http://mosquitto.org/ChangeLog.txt

https://mosquitto.org/man/mosquitto-8.html

https://hub.docker.com/_/eclipse-mosquitto

https://github.com/thomasnordquist/MQTT-Explorer