RocketMQ 之 IoT 消息解析：物联网需要什么样的消息技术?
前言：
自开源消息队列初代产品的兴起以来，在推动PC互联网与移动互联网领域的快速腾飞后，则至今日物联网（IoT）、云计算以及云原生等技术 began引领了新的技术趋势。经过超过30年的持续演进与创新
目前，在国内多个行业领域中，消息中间件已成为一项不可或缺的关键技术。伴随着数字转型进程的不断推进，在实际应用场景中客户往往需要同时应对多个复杂场景：一方面需要完成物联网消息处理与微服务消息管理的任务；另一方面还需要完成跨领域应用集成、数据整合与实时分析等多维度的工作需求。这样一来，在这一过程中企业就必须搭建并维护多套不同的消息系统架构，并因此而导致运营成本与学习投入显著增加。
基于此背景，在今年，RocketMQ 5.0 正式推出了版本5.0；相较于上一代产品 RocketMQ 4.0，“架构朝着云计算本源化的方向发展，并支持了更为广泛的业务应用场景。
物联网消息场景
我们先来了解一下物联网的场景是什么？消息在物联网里面有什么作用？
物联网无疑是最为近年来最热门的技术领域之一；许多研究机构及行业报告均指出物联网正以迅速的速度发展：
首先，物联网设备规模爆发式增长，预测会在 2025 年达到 200 多亿台。
其次，在物联网领域中，数据规模以显著速度增长，在过去几年中其增速已接近28%；此外，在未来相当长的时间段内（预计占大多数），实时数据将源自物联网场景。这预示着未来实时流数据处理中将面临大量来自物联网的数据类型。
最后
物联网的发展速度这么快，数据规模那么大，跟消息有什么关系呢？
我们通过这个图来看一下消息在物联网场景发挥的作用：
该系统的主要功能是建立网络连接并负责数据传输任务，在物联网架构中扮演关键角色。该模块不仅支持不同设备之间的数据交互以及云端系统与各端点的数据交互（如传感器节点向云端发送数据、边缘服务器接收并处理指令等），还能够无缝实现云-边-端之间的集成与协作
该系统主要功能体现在数据处理方面，在物联网环境下能够持续稳定地运行，并满足多样化的实时性要求。该系统涉及大量实时性要求的数据流处理场景包括设备日常维护活动异常温度监测等。该系统基于MQ平台具备高效的数据存储与计算能力能够支撑物联网系统中的数据架构构建
物联网消息技术
下面我们来看看在物联网场景里，对消息技术有什么诉求？
通过该表格展开对比分析，并深入探讨物联网消息技术和传统消息技术之间的差异。
经典消息的主要功能是支持服务端系统进行发布与订阅的操作；而物联网技术通过支持不同节点之间的消息发布与订阅来实现不同实体间的通信连接。
我们来分别看一下各自场景的特点：
- 经典消息场景：消息 Broker 和消息客户端被视为服务端系统的重要组成部分，在 IDC 或公共云环境中运行并配备高性能服务器配置。这些服务架构涵盖了容器化部署、虚拟机配置以及物理服务器等不同形式。无论是内网还是外网环境都具备高带宽和稳定的网络质量特性。从业务角度来看，在经典架构中每个客户端能够支持数百至数千台业务机器的同时还能实现数百至数千次每秒的消息生产频率（TPS）。在消费层面上则采用了集群化的模式一个消费者组会服务于多个应用集群共同分享一个消费者 ID 来分担负载压力。每条消息的订阅数量相对较少正常情况下不会超过10个。
- IoT 消息场景：与经典的场景相比存在显著差异与经典的场景相比存在显著差异IoT 消息客户端多为小型设备其计算和存储资源极为有限这些小型设备往往集中于边缘计算环境并依赖于简单化的服务器配置以满足低功耗需求同时确保快速响应能力。由于这些设备通常通过公网进行连接导致网络环境更加复杂且经常处于断网状态或弱网环境下通信质量较差且不稳定性较高在这种特殊环境下物联网应用中的 msg 实例数会达到亿级规模远超传统大型企业服务系统的 server 数量尽管单个设备的 TPS 较低但一条 msg 却可能被成百上千个物联网终端接收导致订阅比出现高度集中现象。
RocketMQ - MQTT
从上述分析可以看到，在物联网领域中所采用的消息技术与传统消息技术存在显著差异。进一步探讨下去的话，则会发现为了适应物联网领域的消息处理需求，在这一版本中 RocketMQ 5.0 做了哪些改进？
RocketMQ 5.0 我们发布了一个子产品，叫做 RocketMQ - MQTT。
它有三个技术特点：
第一部分：它基于 MQTT 协议，在物联网弱网环境下及低算力设备上进行了优化设计。该协议架构紧凑，并提供了丰富的功能模块来支持多种订阅模式以及多样的消息 QoS 服务类型，包括限定式传输（最多一次）、非限定式传输（最少一次）以及条件性传输（当且仅当一次）。其领域模型设计主要围绕"消息、主题、发布订阅"等核心概念展开，并与 RocketMQ 架构高度契合，在云端构建一体化的产品形态方面奠定了坚实基础。
第二部分采用了存算分离的设计架构。其中 RocketMQ Broker 负责存储功能,而 MQTT 相关功能模块则在 MQTT Proxy 层进行功能实现,以确保系统能够高效应对海量连接、实时消息订阅以及数据推送等场景。MQTT 消息处理系统具备根据物联网业务负载自动扩展的能力,例如在需要扩展时仅需增加新的 Proxy 节点即可完成扩展现有方案。
第三部分基于端云一体化架构的设计。其中采用 RocketMQ 作为消息存储层，在这种架构下每条消息仅需复制一份至本地存储层即可满足两种不同场景的需求：一方面该份消息既可被物联网设备调用也可被云端服务引用；此外该架构还具备天然的流数据存储特性因此能够使得流计算引擎无需额外配置即可实现对IoT数据的实时处理能力。
随后我们将从几个核心的技术点出发进行深入探讨
（一）IoT 消息存储模型
1. 读放大为主，写放大为辅
为了实现物联网消息的有效管理，在发布订阅的核心业务流程中, 通常采用两种不同的存储方案, 其中一种方案是基于读放大的机制, 即所有消费者共享同一个公共队列用于接收消息, 并且每个消费者都能访问该共享队列并维护个人化的消费指针; 另一种方案则是赋予每个用户独立化的消息处理能力, 系统会自动将每一条新消息分配至目标用户的专属队列中进行处理, 这样可以确保数据的安全性和用户体验得到优化
在物联网场景中的一条消息可能被数百上千个设备同时消费。这就不言而喻地说明了采用放大机制的模型能够显著降低存储成本并提升系统性能。
但是仅采用放大模式难以完全满足需求,MQTT 协议具有其特殊性,其Topic 具有层次化特点,具体表现为支持精准订阅机制以及通配符匹配机制两种不同的订阅方式,在实际应用中,通常会根据需求定义一个多级主题结构,例如在家居场景中,我们既可以实现对"家/浴室/温度"等完整多级主题的精准订阅,也可以通过使用通配符来关注特定的主题如"温度",此外还可以设置仅关注一级主题为"家"的消息匹配策略
对于那些可以直接订阅完整多层次主题的消费者而言，在获取对应多层次主题公共队列方面具备较大的灵活性；然而针对通过通配符进行订阅的方式而言，则不具备反向追踪Topic的能力；因此为了实现这一目标，在消息存储过程中需依据消费者的通配符订阅关系添加相应的队列。这种机制使得相关消费者能够基于其已有的通配符队列顺利获取到相应的信息；而这正是 RocketMQ 所采用的独特设计——通过以读放大的方式为主导，在数据处理上实现高效扩展——即通过将大量数据以较小空间进行优化存储，并通过高效的索引机制快速定位所需信息。
2. 端云一体化存储
基于前文的分析，我们设计了 RocketMQ 端云一体化的存储模型，见下图。
消息可源自多种接入渠道（包括但不限于服务端使用的RMQ/AMQP协议与设备端采用的MQTT协议）。但仅存入一份至Commitlog。随后并随后分配至多个需求场景对应的队列索引。例如，在服务层的应用中使用传统的MQ/AMQP模式时，则会按一级Topic队列进行消息消费；而设备层的应用则可采用MQTT多级Topic及通配符订阅等方式实现消息接收。这样我们就可以基于同一套存储引擎实现...
（二）队列规模问题
众所周知，在Kafka等消息队列系统中每个Topic都是独立存储为单独的文件。然而当Topic的数量不断增加时 消息文件的数量也随之增加 从而导致顺序追加操作逐渐演变为随机追加操作 这种操作方式显著降低了系统的性能表现 RocketMQ作为一种改进型的消息队列系统 在Kafka的基础上实现了性能上的提升 其主要改进措施包括引入Commitlog文件来记录所有消息内容 并通过CQ索引文件来表示每个Topic内的消息队列结构 由于CQ索引的数据量相对较小 因此在这种场景下 随机追加对磁盘IO的影响相对较小 从而能够支持高达数万级的Queue数量然而在实际应用中 这种规模的Queue数量仍然无法满足某些高性能场景的需求 我们因此引入了Rocksdb引擎来进行CQ索引的数据分发机制
面向 IoT 的百万级队列设计
Rocksdb是一个广泛应用的单机KV存储引擎，并具有高效率的顺序写性能。因为我们已经具备了消息序列流存储的能力，因此我们可以避免使用Rocksdb引擎中的WAL来处理事务逻辑。基于Rocksdb来保存CQ索引，并在分发过程中利用其强大的WriteBatch原子特性以提高数据一致性的保障能力。在分发时将当前的MaxPhyOffset注入进去，并根据这一值来进行恢复性检查点操作，在后续操作中还可以根据MaxPhyOffset来执行恢复性检查点操作以确保系统的稳定性与一致性。最后我们还实现了PhyOffset确认以清理已删除的数据并减少不必要的I/O开销从而提高系统的运行效率
（三）IoT 消息推送模型
在介绍完底层的队列存储模型之后，请详细阐述上层的消息实时推送机制（包括匹配查找和可靠触达）是如何运作的？
在 RocketMQ 的传统消费模式中,消费者通过长轮询机制从客户端发起请求,精准获取对应的 Topic 队列信息.然而,在 MQTT 智能终端场景下,由于客户端数量及订阅关系规模庞大,无法继续沿用传统的长轮询模式,导致实现路径更加复杂.因此,我们采用了推拉结合的模型以适应这一场景的需求.
如图所示为一个推拉模型框架，在物联网环境下，边缘设备通过MQ/AMQP/MQTT协议连接至代理节点进行通信。当服务端系统发送的消息经MQ/AMQP/MQTT协议传输至代理节点后，具备实时监控机制的模块持续关注该主题队列的变化状态。随后触发生成事件流程：首先记录新到达的消息主题名称，并将该事件推送至代理节点进行处理；接着根据代理节点与各终端设备之间的订阅关系进行匹配筛选；最后若匹配成功则向存储层发起pull请求以取消消息读取操作并将其重新发送给对应终端设备执行处理
核心关注点即是订阅关系的匹配查询。常见的两种方法包括：第一种是主要依靠简单的广播事件；第二种则是通过集中存储的方式，在线订阅关系（如图中所示的lookup模块）进行匹配查询，并实现精准推送。
事件广播机制似乎存在扩展性局限性, 但其性能表现尚可, 因为我们推送的数据量较小, 具体来说就是 Topic 名称这一部分信息, 并且相同 Topic 的消息事件可以批量推送. RocketMQ 5.0 是默认采用了这一做法. 集中式存储的消息订阅关系, 这种做法并非独有, 等技术手段还包括将数据存入 RDS、Redis 等等系统中. 但要保证消息处理的实时一致性同样面临挑战, 而这种实时性会影响消息处理的整体效率. 下图所示模型中可以看到, 在 Proxy 结点还会增加一个缓存模块用于消息队列缓存, 这是为了在广播场景下减少终端设备在向存储层发起读取数据需求上的负担.
总结
本文将深入分析 RocketMQ 5.0 在物联网消息技术领域的应用与优化问题。第一部分将介绍一个典型的物联网技术架构，并详细探讨消息队列在此架构中的核心功能。第二部分深入分析了物联网场景对消息技术的独特需求，并比较这些需求与服务端应用中相应的技术特性。第三部分则详细介绍了 RocketMQ 5.0 的 MQTT 子系统及其在应对物联网领域挑战中的作用，并旨在帮助读者全面理解消息队列在物联网环境下的关键功能及解决方案。
作者：林清山（隆基）
本文为阿里云原创内容，未经允许不得转载。