本文作者：房成进 – 小米高档研发工程师
小米 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】大众号，获取更多技术干货！
