本文作者：房成进 – 小米高级研发工程师
小米 MQTT 利用场景
小米之家门店的领取告诉是小米 MQTT 落地的重要场景之一，流程如上图所示。店员通过终端发送下单申请到后端服务，后端在接管到下单申请后，调用领取服务，期待用户付款。门店终端如何晓得本次付款是否胜利呢？
咱们采纳 MQTT 协定来实现领取音讯的告诉。领取服务将本次订单的领取后果公布到 MQTT 服务的一个 Topic 中，门店终端与服务放弃长连贯，订阅 Topic 来实时获取领取后果，从而进行下一步操作如打印发票等。得益于 TCP 长连贯和 MQTT 协定的轻量化，门店终端零碎的领取响应能力从 200 毫秒降落至 10 毫秒内，MQTT 服务公布端到订阅端的均匀延时为 2.6ms。
手机智能制作工厂是小米 MQTT 落地的另一个外围场景。MQTT 次要利用于设施状态数据采集以及设施控制指令下发。上图右侧为小米智能制作工厂架构图。
上行链路流程为：手机生产线上的泛滥工业设施会将操作日志、设施参数、环境参数等通过工业管制层公布到 MQTT 服务，MQTT 服务的存储层通过数据集成工作将数据打入大数据系统，进行数据的剖析、建模和解决等，最初实现最上层利用工业 BI 和数字孪生的需要。
上行链路流程为：工厂的工作人员通过云端服务将控制指令下发到 MQTT 集群，生产线上的设施与 MQTT 服务集群放弃长连贯，以承受来自云端的控制指令并执行相应动作。这两个链路对时效性要求很高。目前，MQTT 服务能保障上行和上行链路延时在 20ms 以内，服务可用性为 99.95%。
小米 MQTT 服务架构演进过程
晚期，小米次要基于 RocketMQ 社区在 18 年开源的 RocketMQ-IoT-Bridge 来构建本人的 MQTT 服务。RocketMQ-IoT-Bridge 为单机架构，一是不反对程度扩大，总连接数存在瓶颈，天然无奈保障高可用。二是数据无奈长久化，只提供内存存储，一旦重启服务，必然导致音讯失落。三是只反对 MQTT 协定 QoS0，音讯存在失落危险，无奈满足小米的业务要求。如图所示，服务整体为单机服务架构，公布端和订阅端连贯到同一个过程。
小米基于单机的架构进行了一系列的拓展。高可用方面，从单机变为分布式的可扩大架构，连接数从单机的 5 万变为可横向扩大的模式；可靠性方面，在 QoS0 的根底上实现了 MQTT 协定规定的 QoS 1 和 QoS 2；生产模式方面，除了默认的播送生产，反对了 MQTT5.0 新增的共享生产模式，同时还反对了离线音讯。
上图右侧是小米基于 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，提供了可插拔模式。依据业务数据的上游应用场景，部署时可灵便抉择任意一个音讯队列作为长久化层。
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 音讯的映射。
除音讯数据之外，元数据是 MQTT 服务十分重要的一部分。MQTT Bridge 中保留了两类元数据，别离是客户端元数据和订阅元数据。客户端元数据保留了客户端的连贯信息、连接时间、客户端 ID、Netty channel 等信息，咱们实现了可视化的控制台，反对查问 MQTT 服务的连接数，反对通过连贯 ID 和客户端 ID 查问客户端的信息。此外，实现了客户端高低线告诉，用户能够通过订阅 MQTT Topic 实时获取到某个客户端的上线和下线事件。订阅元数据保留了客户端和 MQTT 的映射关系，次要通过 Trie 树来保留订阅关系，能够满足通配符的形式订阅，实现疾速匹配。Bridge 通过订阅 Topic 找到客户端，将音讯定向推送。
MQTT 协定次要有三个服务质量等级 QoS 0、QoS 1 和 QoS 2。QoS 0 示意音讯最多发一次，可能存在失落音讯的状况，性能最好，对于数据可靠性要求不高的业务较为实用。QoS 1 为音讯保障能至多达到一次，可能会反复，性能绝对差一些。QoS 2 为音讯不丢不重，但性能最差。
上图为 QoS0 的实现流程。QoS 指发送端和接收端之间的音讯传输品质。公布音讯时，MQTT Bridge 作为音讯的接收端，IoT 设施作为公布端。订阅音讯时，MQTT Bridge 作为公布端，IoT 设施作为接收端。公布和订阅是两个独立的 QoS 过程，整条链路的 QoS 是这两局部 QoS 的最低值，比方公布是 QoS 1，订阅是 QoS 0，则整条链路的 QoS 等级就是 0。左侧是 QoS 0 公布的过程。公布端 IoT 将音讯推送给 MQTT Bridge，Bridge 将音讯异步推送到 RocketMQ，无需期待响应。图中两个箭头的申请都可能失败，可能会丢数据，可靠性很低。但因为链路短，因而性能较高。
上图为 QoS 1 的实现流程。IoT 终端公布音讯之前，会先将其长久化到本地内存里，Bridge 收到音讯之后，将音讯异步推送到 RocketMQ，期待音讯长久化胜利的后果后，再返回 pubAck 包给 IoT，IoT 将内存里的这条音讯删除。发送的申请可能会失败，发送端内存里存储了音讯，因而能够通过重试来实现音讯至多被发一次，但也导致音讯可能会反复发送。订阅端同理。
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 的重用。
推拉模型是 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 没有订阅时再将线程停掉。这样存在的问题是如果长时间没有音讯公布，但订阅关系始终存在，会导致线程空转，存在很大的资源节约。
社区在往年 3 月份开源新版 MQTT 架构，架构中引入了 notify 组件。作用为告诉所有 MQTT Bridge 一级 Topic 中是否有新的音讯产生。每一个 Bridge 中都内置 notify 组件，负责启动针对 RocketMQ 一级 Topic 的集群模式消费者，一旦一级 Topic 中有音讯产生时，notify 组件可能感知到音讯的产生，同时将音讯作为事件播送给所有 Bridge。其余 Bridge 收到音讯事件的告诉后，会为连贯在这台 Bridge 上的每个终端开启独立拉取工作。拉取时不是拉取一级 Topic 中的所有数据，而是通过生产 4.9.3 版本新引入的 LMQ，防止拉取一级 Topic 中其余没有被以后客户端订阅的音讯，以此防止了读放大。另外，每个终端独立的拉取工作能够为每个终端保护独立的 offset 进度，不便实现离线音讯。
因而，只有新的音讯事件到来时，才会为终端开启拉取工作。Topic 没有音讯或没有任何订阅关系之后，拉取工作将进行。降级后的推拉模型可能反对离线音讯，大幅升高了延时，正当的启停机制无效防止线程资源的节约。
共享订阅是 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 多订阅端的共享生产。
将来工作
将来，小米 MQTT 的工作将从以下四个方面持续深刻摸索：
- 架构：推拉模型持续降级欠缺；
- 性能：离线音讯、保留音讯、遗嘱音讯等性能的欠缺；
- 社区：拥抱开源社区，追随社区降级 RocketMQ 端云一体化的架构；
- 业务：小米汽车等 IoT 的场景推广和利用。
退出 Apache RocketMQ 社区
十年铸剑，Apache RocketMQ 的成长离不开寰球靠近 500 位开发者的积极参与奉献，置信在下个版本你就是 Apache RocketMQ 的贡献者，在社区不仅能够结识社区大牛，晋升技术水平，也能够晋升集体影响力，促成本身成长。
社区 5.0 版本正在进行着热火朝天的开发，另外还有靠近 30 个 SIG（兴趣小组）等你退出，欢送立志打造世界级分布式系统的同学退出社区，增加社区开发者微信：rocketmq666 即可进群，参加奉献，打造下一代音讯、事件、流交融解决平台。
微信扫码增加小火箭进群
另外还能够退出钉钉群与 RocketMQ 爱好者一起宽泛探讨：
钉钉扫码加群
关注【Apache RocketMQ】公众号，获取更多技术干货！