基于 RocketMQ 的 MQTT 服务架构在小米的实践

作者：阿里巴巴云原生

2022 年 9 月 27 日
浙江
本文字数：4783 字
阅读完需：约 16 分钟

1.jpeg

本文作者：房成进 - 小米高级研发工程师

小米 MQTT 应用场景

2.png

小米之家门店的支付通知是小米 MQTT 落地的重要场景之一，流程如上图所示。店员通过终端发送下单请求到后端服务，后端在接收到下单请求后，调用支付服务，等待用户付款。门店终端如何知道本次付款是否成功呢？

我们采用 MQTT 协议来实现支付消息的通知。支付服务将本次订单的支付结果发布到 MQTT 服务的一个 Topic 中，门店终端与服务保持长连接，订阅 Topic 来实时获取支付结果，从而进行下一步操作如打印发票等。得益于 TCP 长连接和 MQTT 协议的轻量化，门店终端系统的支付响应能力从 200 毫秒下降至 10 毫秒内，MQTT 服务发布端到订阅端的平均延时为 2.6ms。

2.png

手机智能制造工厂是小米 MQTT 落地的另一个核心场景。MQTT 主要应用于设备状态数据采集以及设备控制指令下发。上图右侧为小米智能制造工厂架构图。

上行链路流程为：手机生产线上的众多工业设备会将操作日志、设备参数、环境参数等通过工业控制层发布到 MQTT 服务，MQTT 服务的存储层通过数据集成任务将数据打入大数据系统，进行数据的分析、建模和处理等，最后实现最上层应用工业 BI 和数字孪生的需求。

下行链路流程为：工厂的工作人员通过云端服务将控制指令下发到 MQTT 集群，生产线上的设备与 MQTT 服务集群保持长连接，以接受来自云端的控制指令并执行相应动作。这两个链路对时效性要求很高。目前， MQTT 服务能保证上行和下行链路延时在 20ms 以内，服务可用性为 99.95%。

小米 MQTT 服务架构演进过程

3.png

早期，小米主要基于 RocketMQ 社区在 18 年开源的 RocketMQ-IoT-Bridge 来构建自己的 MQTT 服务。RocketMQ-IoT-Bridge 为单机架构，一是不支持水平扩展，总连接数存在瓶颈，自然无法保证高可用。二是数据无法持久化，只提供内存存储，一旦重启服务，必然导致消息丢失。三是只支持 MQTT 协议 QoS0，消息存在丢失风险，无法满足小米的业务要求。如图所示，服务整体为单机服务架构，发布端和订阅端连接到同一个进程。

4.png

小米基于单机的架构进行了一系列的拓展。高可用方面，从单机变为分布式的可扩展架构，连接数从单机的 5 万变为可横向扩展的模式；可靠性方面，在 QoS0 的基础上实现了 MQTT 协议规定的 QoS 1 和 QoS 2；消费模式方面，除了默认的广播消费，支持了 MQTT5.0 新增的共享消费模式，同时还支持了离线消息。

5.png

上图右侧是小米基于 RocketMQ 的分布式 MQTT 架构图。最上层为客户端，发布者和订阅者连接到负载均衡器，这里使用四层的负载均衡 LVS, 主要目的是将请求均摊到各个 MQTT Bridge。MQTT Bridge 即 MQTT 的服务节点，负责连接、订阅、解析协议和消息转发。RocketMQ 作为存储层，负责持久化消息。类似于存算分离设计，MQTT Bridge 和 RocketMQ 均可独立水平扩展。

得益于 RocketMQ 从 2020 年开始在小米大规模落地，我们采用 RocketMQ 来持久化 MQTT 消息。整个发布订阅的过程演变成消息从 Bridge 发送到 RocketMQ，再从 RocketMQ 消费数据然后推送到订阅端。每一个 MQTT Bridge 内嵌 RocketMQ SDK ，充当 RocketMQ 的客户端，既作为生产者也作为消费者。

此外，持久化层支持了小米自研的消息队列 Talos，提供了可插拔模式。根据业务数据的下游使用场景，部署时可灵活选择任意一个消息队列作为持久化层。

6.png

MQTT 协议的消息结构和 RocketMQ 的消息结构互相独立，因此如果将 MQTT 协议的消息持久化到 RocketMQ 中，必然需要做一定的匹配。MQTT Topic 有多级，如图中 T1/T2/T3 所示，为多级树形结构。将 T1 看作一级 Topic，对应 RocketMQ 中的 Topic T1，则所有发往以 T1 开头的 MQTT Topic 的消息都会持久化到 RocketMQ 的 T1 Topic 中。

此时问题演变成如何区分一条消息属于哪个 MQTT Topic，我们选择将 MQTT Topic 设为消息的 tag，MQTT 消息中的一些可变 header 直接放在 RocketMQ 消息属性 KV 中，消息体可以直接映射到 RocketMQ 消息的 Payload 中，这样完成了 MQTT 消息到 RocketMQ 消息的映射。

7.png

除消息数据之外，元数据是 MQTT 服务非常重要的一部分。MQTT Bridge 中保存了两类元数据，分别是客户端元数据和订阅元数据。客户端元数据保存了客户端的连接信息、连接时间、客户端 ID、Netty channel 等信息，我们实现了可视化的控制台，支持查询 MQTT 服务的连接数，支持通过连接 ID 和客户端 ID 查询客户端的信息。此外，实现了客户端上下线通知，用户可以通过订阅 MQTT Topic 实时获取到某个客户端的上线和下线事件。订阅元数据保存了客户端和 MQTT 的映射关系，主要通过 Trie 树来保存订阅关系，可以满足通配符的方式订阅，实现快速匹配。Bridge 通过订阅 Topic 找到客户端，将消息定向推送。

8.png

MQTT 协议主要有三个服务质量等级 QoS 0、 QoS 1 和 QoS 2。QoS 0 表示消息最多发一次，可能存在丢失消息的情况，性能最好，对于数据可靠性要求不高的业务较为实用。QoS 1 为消息保证能至少到达一次，可能会重复，性能相对差一些。QoS 2 为消息不丢不重，但性能最差。

9.png

上图为 QoS0 的实现流程。QoS 指发送端和接收端之间的消息传输质量。发布消息时，MQTT Bridge 作为消息的接收端，IoT 设备作为发布端。订阅消息时，MQTT Bridge 作为发布端，IoT 设备作为接收端。发布和订阅是两个独立的 QoS 过程，整条链路的 QoS 是这两部分 QoS 的最低值，比如发布是 QoS 1，订阅是 QoS 0，则整条链路的 QoS 等级就是 0。左侧是 QoS 0 发布的过程。发布端 IoT 将消息推送给 MQTT Bridge，Bridge 将消息异步推送到 RocketMQ，无需等待响应。图中两个箭头的请求都可能失败，可能会丢数据，可靠性很低。但由于链路短，因此性能较高。

10.png

上图为 QoS 1 的实现流程。IoT 终端发布消息之前，会先将其持久化到本地内存里，Bridge 收到消息之后，将消息异步推送到 RocketMQ，等待消息持久化成功的结果后，再返回 pubAck 包给 IoT，IoT 将内存里的这条消息删除。发送的请求可能会失败，发送端内存里存储了消息，因此可以通过重试来实现消息至少被发一次，但也导致消息可能会重复发送。订阅端同理。

11.png

QoS 2 的实现流程如上图。在 QoS 1 时， Bridge 接受到消息后没有将消息持久化在自己的内存里，而是直接将消息推送到 RocketMQ 中。如果发送端一直没有收到 pubAck 包，则执行重发，重发之后 Bridge 无法获知收到的消息是新消息还是重发消息，会造成消息重复。QoS 2 基于 messageID 来实现消息去重。MQTT 协议要求 message ID 可以被重复使用，且有一定范围，不会一直递增。所以在利用 messageID 去重的同时，还要保证 messageID 在传输过程中不能有重复，用完后必须释放。

依据这两点前提，sender 在发送消息之前，会将消息持久化在自己的内存里，再推送给 receiver。receiver 收到消息之后也会放在本地内存里，返回 PubRec 包给 sender，通知其已经收到消息。如果 sender 一直没有收到 PubRec 包，会不断地重复发送消息。由于 receiver 内存里已经保存了消息，因此可以通过 messageID 来实现消息的去重。发送端在接收到 PubRec 包后发布 PubRel 包，通知 receiver 可以清理内存中的消息，也意味着 sender 已经知道消息已被 receiver 持久化，此时再由 receiver 将消息推给 RocketMQ 并等待持久化响应。receiver 发送 PubComp 包给 sender 通知其可将 PubRel 包删除。上图中步骤 3.3 可能失败，因此 sender 必须在内存中缓存 PubRel 包。上述流程存在两步确认机制，第一个是保证消息能到达 receiver ；第二个是保证将用过的 messageID 释放掉，能够实现 message ID 的重用。

12.png

推拉模型是 MQTT Bridge 实现消息发布订阅的核心模型。假设以下场景：有四个订阅端，其中订阅端 IoT-1 和 IoT-2 分别订阅了 Topic1/a、Topic1/b，IoT-3 和 IoT-4 分别订阅了 Topic2/ a。第一、二台设备连接到第一个 Bridge，第三、四台设备连接到第二个 Bridge。当有新的订阅关系过来时，会检查订阅一级 Topic。上图中 Bridge1 维护的两个订阅关系分别是 Topic1/a、Topic1/b，它会启动 RocketMQ 的消费任务，从 RocketMQ 中消费 Topic1 中的数据。消费到的每条消息通过 tag 判断属于哪个 MQTT Topic，再通过匹配树将消息推送给客户端。每一个 RocketMQ Topic 对应一个拉取消息的任务，而一级 Topic 下面可能有很多 MQTT Topic，一旦 MQTT Topic 增多，推送给客户端的延时就会变高。此外，一级 Topic 下可能会存在很多终端，存在大量没有被订阅的无用消息。

Topic 级别的任务无法为每个客户端都维护独立的 offset 进度。只要 Bridge 接收到客户端订阅的请求就会开启消费线程，Topic 没有订阅时再将线程停掉。这样存在的问题是如果长时间没有消息发布，但订阅关系一直存在，会导致线程空转，存在很大的资源浪费。

13.png

社区在今年 3 月份开源新版 MQTT 架构，架构中引入了 notify 组件。作用为通知所有 MQTT Bridge 一级 Topic 中是否有新的消息产生。每一个 Bridge 中都内置 notify 组件，负责启动针对 RocketMQ 一级 Topic 的集群模式消费者，一旦一级 Topic 中有消息产生时，notify 组件能够感知到消息的产生，同时将消息作为事件广播给所有 Bridge。其他 Bridge 收到消息事件的通知后，会为连接在这台 Bridge 上的每个终端开启独立拉取任务。拉取时不是拉取一级 Topic 中的所有数据，而是通过消费 4.9.3 版本新引入的 LMQ，避免拉取一级 Topic 中其他没有被当前客户端订阅的消息，以此避免了读放大。另外，每个终端独立的拉取任务可以为每个终端维护独立的 offset 进度，方便实现离线消息。

因此，只有新的消息事件到来时，才会为终端开启拉取任务。Topic 没有消息或没有任何订阅关系之后，拉取任务将停止。升级后的推拉模型能够支持离线消息，大幅降低了延时，合理的启停机制有效避免线程资源的浪费。

14.png

15.png

共享订阅是 MQTT5.0 协议新增的订阅模式，可以理解为类似 RocketMQ 中的集群模式消费。上图左侧为简单的共享队列实例。IoT 发送几条消息到 Topic1/a 中，Topic1/a 有三个订阅端，每一条消息只会被推送给其中一个订阅端，比如 IoT-sub-1 会收到 message1 和 message4，IoT-sub-1 会收到 message2 和 message5，message 会收到 message 3 和 message 6。其实现原理为：

每个 MQTT Bridge 会启动一个针对 Topic 的拉取任务。RocketMQ 本身能够支持集群模式，MQTT Bridge 又作为 RocketMQ 的客户端，因此可以复用 RocketMQ 的共享订阅模型。订阅端订阅时以某种特殊方式带上消费者组名称，连接到某台 Bridge 后，该 Bridge 上就会用消费者组和订阅的一级 Topic 来启动一个 RocketMQ 的集群模式消费者。第二个订阅端连接了第二台 Bridge，该 Bridge 也会启动一个消费者。只要 Bridge 上有终端连接且他们处于一组内并订阅了同一个 RocketMQ 的一级 Topic，则所有符合要求的 Bridge 会组成集群模式的消费者集群。有新的消息到达 Topic1 之后，只会被其中一个 Bridge 消费，那么也只会被连接到该 Bridge 上的 IoT 订阅端消费到。如果有多个订阅端同时连到一个 Bridge 上，消息应该推给哪个客户端呢？我们在 MQTT Bridge 内置多种策略，默认选择轮询策略。一条消息发到 Bridge 后，Bridge 可以轮询发送给任意一个 IoT 订阅端，实现单 Bridge 多订阅端的共享消费。