基于 RocketMQ 的云原生 MQTT 消息引擎设计

本文主要是介绍基于 RocketMQ 的云原生 MQTT 消息引擎设计，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

作者：沁君

概述

随着智能家居、工业互联网和车联网的迅猛发展，面向 IoT（物联网）设备类的消息通讯需求正在经历前所未有的增长。在这样的背景下，高效和可靠的消息传输标准成为了枢纽。MQTT 协议作为新一代物联网场景中得到广泛认可的协议，正逐渐成为行业标准。

本次我们将介绍搭建在 RocketMQ 基础上实现的 MQTT 核心设计，本文重点分析 RocketMQ 如何适应这些变化，通过优化存储和计算架构、推送模型及服务器架构设计，推动 IoT 场景下消息处理的高效性和可扩展性以实现 MQTT 协议。

此外，阿里云 MQTT 以 RocketMQ-MQTT 为基础，不断进行迭代创新。阿里云是开源 RocketMQ-MQTT 的主要贡献者和使用者之一。面对设备通信峰谷时段差异性的挑战，本文将介绍阿里云如何将 Serverless 架构应用于消息队列，有效降低运营成本，同时利用云原生环境的特性，为 IoT 设备提供快速响应和灵活伸缩的通讯能力。

进一步地，我们将探讨介绍在云端生态体系中整合 MQTT 的实践，介绍基于统一存储的数据生态集成方案，展示其强大的技术能力和灵活的数据流转能力。

loT 消息场景

消息场景对比

物联网技术，作为当代科技领域的璀璨明星，其迅猛发展态势已成共识。据权威预测，至 2025 年，全球物联网设备安装基数有望突破 200 亿大关，这一数字无疑昭示着一个万物互联时代的到来。

更进一步，物联网数据量正以惊人的年增长率约 28% 蓬勃膨胀，预示着未来数据生态中超过 90% 的实时数据将源自物联网。这一趋势深刻改变了数据处理的格局，将实时流数据处理推向以物联网数据为核心的新阶段。

边缘计算的崛起，则是对这一变革的积极响应。预计未来，高达 75% 的数据处理任务将在远离传统数据中心或云端的边缘侧完成。鉴于物联网数据的海量特性，依赖云端进行全部数据处理不仅成本高昂，且难以满足低延迟要求。因此，有效利用边缘计算资源，就地进行数据初步处理，仅将提炼后的关键信息上传云端，成为了提升效率、优化用户体验的关键策略。

在此背景下，消息传递机制在物联网场景中的核心价值愈发凸显：

桥梁作用： 消息系统充当了物联网世界中的“神经网络”，无缝衔接设备与设备、设备与云端应用间的沟通渠道，构筑起云-边-端一体化的应用框架，确保了信息交流的即时与高效。
数据加工引擎： 面对物联网持续涌动的数据洪流，基于消息队列（MQ）的事件流存储与流计算技术成为了解锁实时数据分析潜能的钥匙。这一机制不仅能够实时捕捉、存储数据流，还支持在数据产生的瞬间执行计算操作，为物联网应用提供了强大的数据处理基础架构，助力实现数据的即时洞察与决策响应。

总之，消息技术不仅是物联网架构的粘合剂，更是驱动数据流动与智能决策的核心动力，其在物联网领域的应用深度与广度，正随着技术迭代与市场需求的双重驱动而不断拓展。

同时传统消息场景和物联网消息场景有很多的不同，包含以下几个特点：

1）硬件资源差异

传统消息场景依托于高性能、高可靠性的服务集群，运算资源充沛，客户端部署环境多为容器、虚拟机乃至物理服务器，强调集中式计算能力。相比之下，物联网消息场景的客户端直接嵌入至网络边缘的微型设备中，如传感器、智能家电等，这些设备往往受限于极低的计算与存储资源，对能效比有着极高要求。

2）网络环境挑战

在经典的内部数据中心（IDC）环境中，消息处理享有稳定的网络条件和可控的带宽、延迟指标。而物联网环境则拓展至公共网络，面对的是复杂多变的网络状况，尤其在偏远地区或网络覆盖弱的区域，不稳定的连接成为常态，对消息传输的健壮性和效率提出了更高挑战。

3）客户端规模的量级跃迁

传统消息系统处理的日均消息量通常维持在百万级，适用于集中度较高的消息分发。物联网的兴起促使设备数量呈爆炸性增长，动辄涉及亿万级别的终端节点，这对系统的扩展性、消息路由的高效性提出了全新的要求。

4）生产与消费模式的演变

传统场景倾向于采用集中式同步生产模式，强调消息的一致性与有序处理。而物联网消息生成则体现出分散化的特性，每个设备根据其感知环境独立产生少量消息，这对消息收集与整合机制设计提出了新的思考。消费模式上，物联网场景经常涉及大规模广播或组播，单条消息可能触达数百万计的接收者，要求系统具备高效的广播能力和灵活的订阅管理机制。

RocketMQ 的融合架构设计

融合架构设计图

我们看到，物联网所需要的消息技术与经典的消息设计有很大的不同。接下来我们来看看基于 RocketMQ 的融合架构 MQTT 设计为了解决物联网的消息场景有哪些特点。

1）融入 MQTT 协议，适应物联网环境特性

RocketMQ 5.0 通过整合 RocketMQ-MQTT，紧密贴合了物联网领域广泛采用的MQTT协议标准。此协议针对低功耗、网络条件不稳定的情况优化，以其轻量级特性和丰富的功能集脱颖而出，支持不同的消息传递保障级别，满足了从“最多一次”到“仅一次”的多样化需求。协议的领域模型与 RocketMQ 的核心组件高度协调，促进了消息、主题、发布订阅模式的自然融合，为建立一个从设备到云端的无缝消息传递体系打下了稳固的基础。

2）灵活的存储与计算解耦架构

为了应对物联网场景下对高并发连接和大规模数据处理的需求，RocketMQ 5.0 采取了存储与计算分离的架构设计。RocketMQ Broker 作为核心存储组件，确保了数据的持久化与可靠性，而 MQTT 相关的逻辑操作则在专门的代理层实施，这不仅优化了对大量连接、复杂订阅关系的管理，也增强了实时消息推送的能力。这种架构允许根据业务负载动态调整代理层资源，通过增加代理节点来平滑应对设备连接数的增加，体现了系统良好的弹性和扩展潜力。

3）促进端云数据协同的整合架构

RocketMQ-MQTT 通过其整合的架构设计，促进了物联网设备与云端应用之间的高效数据共享。基于统一的消息存储策略，每条消息在系统内只需存储一次，即可供两端消费，减少了数据冗余，提高了数据流通的效率。此外，RocketMQ 作为数据流的存储中枢，自然而然地与流计算技术结合，为实时分析物联网生成的海量数据提供了便利，加速了数据价值的发掘过程。

存储设计

首先要解决的问题是物联网消息的存储模型。在发布订阅业务模型中，常用的存储模型有两种，写放大和读放大，我们将依次分析两种模型。

写放大模型

读放大模型

写放大模型：

在此模型下，每位消费者拥有专属的消息队列，每条消息需要复制并分布到所有目标消费者的队列中，消费者仅关注并处理自己队列中的消息。以三级主题/Topic/subTopic/test 为例，若该主题吸引了大量客户端订阅，采取“一客一队列”的策略，即每个符合订阅规则的客户端或通配符订阅均维护一份消息副本，将导致消息的存储需求随订阅者数量线性增长。

尽管这种模式在某些传统消息场景，比如遵循 AMQP 协议的应用中表现得游刃有余，因为它确保了消息传递的隔离性和可靠性。但在物联网场景下，特别是当单个消息需被数以百万计的设备消费时，“写放大”策略将引发严重的存储资源消耗问题，迅速成为不可承受之重。

读放大的考量与挑战：

鉴于物联网场景的特殊需求，直接应用传统的“写放大”模型显然是不可行的。为解决这一难题，RocketMQ-MQTT 采取了更为高效与灵活的存储策略，旨在减少存储冗余，提高系统整体的可扩展性和资源利用率：

在“读放大”模式下，每条消息实际上被存储一次，而为了支持通配符订阅的高效检索，系统在消息存储阶段会创建额外的索引信息——即 consume queue（消费队列）。对于如/Topic/subTopic/*这样的通配符订阅，系统会在每个匹配的通配符队列中生成相应的索引，使得订阅了不同通配符主题或具体主题的消费者，都能通过这些共享的存储实体找到并消费到消息。尽管这看似增加了“读”的复杂度，但实际上，每个 consume queue 作为索引，其体积远小于原始消息，显著降低了整体存储成本，同时提高了消息检索与分发的效率。

原子分发示意图

为此，我们设计了一种多维度分发的 Topic 队列模型，如上图所示，消息可以来自各个接入场景（如服务端的 MQ/AMQP、客户端的 MQTT），但只会写一份存到 commitlog 里面，然后分发出多个需求场景的队列索引（ConsumerQueue），如服务端场景（MQ/AMQP）可以按照一级 Topic 队列进行传统的服务端消费，客户端 MQTT 场景可以按照 MQTT 多级 Topic 以及通配符订阅进行消费消息。这样的一个队列模型就可以同时支持服务端和终端场景的接入和消息收发，达到一体化的目标。

实现这一模型，RocketMQ依托了两项关键技术特性：

轻型队列（Light Message Queue）

这一特性允许一条消息被灵活地写入多个 topic queue 中，确保了消息能够高效地适应各种复杂的订阅模式，包括但不限于通配符订阅。它为读放大模型的实现提供了必要的灵活性和效率基础。
百万队列能力

RocketMQ 通过集成 RocksDB 这一高性能键值存储引擎，充分利用其在顺序写入方面的优势，实现了百万级别的队列管理能力。特别是通过定制化配置，去除了 RocksDB 内部的日志预写（Write-Ahead Log, WAL），进一步优化了存储效率。RocksDB 不仅为 consume queue 提供了稳定高效的存储方案，还确保了即便在极端的队列数量下，系统依然能够保持高性能的索引处理能力。

轻型队列的实现

通过采用“读放大”模型，结合 RocketMQ 的轻型队列特性和百万队列的底层技术支持，我们不仅有效解决了物联网环境下消息存储与分发的挑战，还实现了存储成本与系统性能的双重优化。这种设计不仅减少了存储空间的占用，还通过高度优化的索引机制加快了消息检索速度，为大规模物联网设备的消息通信提供了一个既经济又高效的解决方案。

推送模型

RocketMQ-MQTT 推送模型

在介绍完底层队列存储模型之后，我们将重点探讨匹配查找和可靠送达的实现机制。在传统的消息队列 RocketMQ 中，经典的消费模式是消费者通过客户端直接发起长轮询请求，以精准地获取对应主题的队列消息。然而，在 MQTT 场景下，由于客户端数量众多且订阅关系复杂，长轮询模式显得不够有效，因此消费过程变得更加复杂。为此，我们采用了一种推拉结合的模型。

本模型的核心在于终端通过 MQTT 协议连接至代理节点，消息可以来源于多种场景（如MQ、AMQP、MQTT）。当消息存入主题队列后，通知逻辑模块将实时监测到新消息的到达，进而生成消息事件（即消息的主题名称），并推送至网关节点。网关节点根据连接终端的订阅状态进行内部匹配，识别能够接收这一消息的终端，随后触发拉取请求，以从存储层读取消息并推送至终端。

在这个流程中，一个关键问题是通知模块如何确定终端感兴趣的消息，以及哪些网关节点会对此类消息感兴趣。这实际上是一个核心的匹配搜索问题。常见的解决方案主要有两种：第一种是简单的事件广播，第二种是将线上订阅关系集中存储（例如图中的 Lookup 模块），然后进行匹配搜索，再执行精准推送。

虽然事件广播机制在扩展性上存在一定问题，但其性能表现仍然良好，因为我们推送的数据量相对较小，仅为 Topic 名称。此外，同一 Topic 的消息事件可以合并为一个事件，这是我们当前在生产环境中默认采用的方式。另一方面，将线上订阅关系集中存储在 RDS 或 Redis 中也是一种普遍的做法，但这需要保证数据的实性，匹配搜索的过程可能会对整体实时消息链的延迟产生影响。

在该模型中，还设计了一个缓存模块，以便在需要广播大量消息时，避免各个终端对存储层发起重复的读取请求，从而提高整体系统的效率。

阿里云 MQTT 在 Serverless 上的实践

随着云原生技术的不断发展，现代消息中间件逐渐以容器编排为基础，如何实现真正的无服务器架构及秒级弹性管理已成为一项重要的研究课题。

阿里云作为开源 RocketMQ-MQTT 的主要贡献者和使用者之一，在 MQTT 弹性设计上有很多优化方式和实践经验。我们将介绍阿里云在弹性上的设计思路，展示其如何实现高效、弹性强的 MQTT 消息中间件：

1）抽离网络连接层

阿里云 MQTT 采用类似 Sidecar 的模式，将网络连接层与核心业务逻辑进行分离，使用 Rust 语言来处理网络连接。与 Java 相比，Rust 在内存消耗和启动速度上具有显著优势，尤其在处理大规模 MQTT 连接时，能够有效降低内存占用。

2）秒级扩容

每个 Pod 的资源请求设置较低，同时预留部分 Pod 专门运行 Rust 进程。在扩容需求出现时，系统能够快速启动 MQTT Proxy 进程，省去 Pod 创建和资源挂载的时间，从而显著提升响应速度。

3）弹性预测与监控

利用连接数、TPS、内存、CPU 等白盒指标，以及 RT 等黑盒指标，阿里云 MQTT 依据指标联动规则，制定了合理的扩容策略。这使得系统能够提前预测负载变化并启动 Pod 扩容，确保长期平稳运行。

通过以上设计思路，阿里云能够构建一个高效、弹性强的基于 RocketMQ 的 MQTT 实现方案，充分利用 Rust 带来的性能优势，同时保持系统的稳定性与可扩展性。这种创新设计将在实际应用中显著提升用户体验，助力系统整体性能的优化。

阿里云 MQTT 在车联网中的实践架构

随着汽车出行领域新四化（电气化、智能化、网联化和共享化）的推进，各大汽车制造商正逐步构建以智能驾驶和智能网联为核心的车联网系统。这一新一代车联网系统对底层消息采集、传输和处理的平台架构提出了更高的要求。接下来，我们将介绍阿里云 MQTT 在车联网中的实践架构及其应用价值。

在架构图中，我们可以看到常见的车联网设备，包括车载终端、路测单元和手机端系统。这些设备确保了安全的连接与数据传输。车端的功能涵盖车机数据上报、POI 下发、文件推送、配置下发、消息推送等全新车联业务。这些操作将产生海量的消息 Topic，需要更加安全、稳定的接入与传输，以实现可靠的消息订阅与发布。路端则强调路侧 RSU 的安全接入，支持消息的采集、传输以及地图数据的实时更新。

接入端支持多种协议，包括 TCP、x509、TLS、WSS、WS、OpenAPI 和 AMQP，以满足不同应用场景的灵活需求。这种多协议支持确保了设备之间的无缝互联与高效通信。

在流转生态方面，物联网场景下，各种设备持续产生大量数据，业务方需要对这些数据进行深入分析与处理。采用 RocketMQ 作为存储层，系统能够只保存一份消息，并支持物联网设备和云端应用的共同消费。RocketMQ 的流存储特性使得流计算引擎能够无缝、实时地分析物联网数据，为关键决策提供及时支持。

借助阿里云 EventBridge，MQTT 物联网设备所生成的信息可以顺利流转至 Kafka、AMQP、FC、Flink 等其他中间件或数据处理平台，实现深度的数据分析与处理。事件总线 EventBridge 是阿里云提供的一款 Serverless 总线服务，支持阿里云服务、自定义应用、SaaS 应用以标准化、中心化的方式接入，并能够以标准化的 CloudEvents 1.0 协议在这些应用之间路由事件，帮助轻松构建松耦合、分布式的事件驱动架构。这种灵活的数据流转能力不仅提升了处理速度，还为未来智能化应用的创新和发展奠定了基础。

通过以上架构，可以清晰地看到阿里云 MQTT 在车联网领域的最佳实践，为实现未来智能出行提供了可靠的技术支撑。

结语

在物联网的蓬勃发展背景下，消息传递技术的不断演进已成为支撑智能家居、工业互联网以及车联网等领域的重要基石。通过对 RocketMQ 和 MQTT 协议的深度融合，我们不仅有效解决了物联网时代对高效、可靠消息传输的需求，也为设备通信带来了灵活的解决方案。

阿里云在这一领域的积极探索，通过引入 Serverless 架构，不断推进 MQTT 的技术迭代与创新。这样的设计能够在面对高并发连接和海量数据时，动态调整资源配置，降低成本并提升响应速度，确保了实时数据处理的高效性。

当前，社区正在推动 MQTT 5.0 协议方面已取得显著进展，新的协议特性如更丰富的错误码、更灵活的连接选项以及 will 消息、retain 消息、共享订阅功能都将进一步提升系统的灵活性和可靠性。与此同时，我们在致力于实现更快的弹性扩展能力，以便在面对突发流量时及时响应，提高系统的可用性和灵活性。

随着 IoT 生态体系的不断完善，面对日益复杂的消息场景，消息技术的价值愈发凸显。我们相信，未来通过不断优化的消息架构，能够推动更深层次的智能化应用，同时为构建万物互联的未来奠定坚实的基础。让我们期待 MQTT 在物联网技术的场景中展现其无限可能，同时也继续探索持续探索在确保安全、稳定、高效的消息中间件。