RocketMQ 端云一体化设计与实践

Posted 2022-02-09 阿里云云栖号

简介：本次分享主要介绍面向设备端消息收发应用场景的架构模型设计，以及如何实现 基于RocketMQ的一体化消息平台。

作者：悟幻

一体化背景

不止于分发

我们都知道以 RocketMQ 为代表的消息（队列）起源于不同应用服务之间的异步解耦通信，与以 Dubbo 为代表的 RPC 类服务通信一同承载了分布式系统（服务）之间的通信场景，所以服务间的消息分发是消息的基础诉求。然而我们看到，在消息（队列）这个领域，近些年我们业界有个很重要的趋势，就是基于消息这份数据可以扩展到流批计算、事件驱动等不同场景，如RocketMQ-streams，Kafka-Streams、Rabbit-Streams等等。

不止于服务端

传统的消息队列MQ主要应用于服务（端）之间的消息通信，比如电商领域的交易消息、支付消息、物流消息等等。然而在消息这个大类下，还有一个非常重要且常见的消息领域，即终端消息。消息的本质就是发送和接受，终端和服务端并没有本质上的大区别。

一体化价值

如果可以有一个统一的消息系统（产品）来提供多场景计算（如stream、event）、多场景（IoT、APP）接入，其实是非常有价值的，因为消息也是一种重要数据，数据如果只存在一个系统内，可以最大地降低存储成本，同时可以有效地避免数据因在不同系统间同步带来的一致性难题。

终端消息分析

本文将主要描述的是终端消息和服务端消息一体化设计与实践问题，所以首先我们对面向终端的这一大类消息做一下基本分析。

场景介绍

近些年，我们看到随着智能家居、工业互联而兴起的面向IoT设备类的消息正在呈爆炸式增长，而已经发展十余年的移动互联网的手机APP端消息仍然是数量级庞大。面向终端设备的消息数量级比传统服务端的消息要大很多量级，并仍然在快速增长。

特性分析

尽管无论是终端消息还是服务端消息，其本质都是消息的发送和接受，但是终端场景还是有和服务端不太一样的特点，下面简要分析一下:

（1）轻量客户端，服务端一般都是使用很重的客户端SDK封装了很多功能和特性，然而终端因为运行环境受限且庞杂必须使用轻量简洁的客户端SDK。

（2) 服务端正是因为有了重量级客户端SDK，其封装了包括协议通信在内的全部功能，甚至可以弱化协议的存在，使用者无须感知，而终端场景因为要支持各类庞杂的设备和场景接入，必须要有个标准协议定义。

（3）P2P，服务端消息如果一台服务器处理失败可以由另外一台服务器处理成功即可，而终端消息必须明确发给具体终端，若该终端处理失败则必须一直重试发送该终端直到成功，这个和服务端很不一样。

（4）广播比，服务端消息比如交易系统发送了一条订单消息，可能有如营销、库存、物流等几个系统感兴趣，而终端场景比如群聊、直播可能成千上万的终端设备或用户需要收到。

（5）海量接入，终端场景接入的是终端设备，而服务端接入的就是服务器，前者在量级上肯定远大于后者。

架构与模型

消息基础分析

实现一体化前我们先从理论上分析一下问题和可行性。我们知道，无论是终端消息还是服务端消息，其实就是一种通信方式，从通信的层面看要解决的基础问题简单总结就是：协议、匹配、触达。

（1）协议

协议就是定义了一个沟通语言频道，通信双方能够听懂内容语义。在终端场景，目前业界广泛使用的是MQTT协议，起源于物联网IoT场景，OASIS联盟定义的标准的开放式协议。

MQTT协议定义了是一个Pub/Sub的通信模型，这个与RocketMQ类似的，不过其在订阅方式上比较灵活，可以支持多级Topic订阅（如 “/t/t1/t2”），可以支持通配符订阅（如 “/t/t1/+”）

（2）匹配

匹配就是发送一条消息后要找到所有的接受者，这个匹配查找过程是不可或缺的。

在RocketMQ里面实际上有这个类似的匹配过程，其通过将某个Queue通过rebalance方式分配到消费组内某台机器上，消息通过Queue就直接对应上了消费机器，再通过订阅过滤（Tag或SQL）进行精准匹配消费者。之所以通过Queue就可以匹配消费机器，是因为服务端场景消息并不需要明确指定某台消费机器，一条消息可以放到任意Queue里面，并且任意一台消费机器对应这个Queue都可以，消息不需要明确匹配消费机器。

而在终端场景下，一条消息必须明确指定某个接受者（设备），必须准确找到所有接受者，而且终端设备一般只会连到某个后端服务节点即单连接，和消息产生的节点不是同一个，必须有个较复杂的匹配查找目标的过程，还有如MQTT通配符这种更灵活的匹配特性。

（3）触达

触达即通过匹配查找后找到所有的接受者目标，需要将消息以某种可靠方式发给接受者。常见的触发方式有两种：Push、Pull。Push，即服务端主动推送消息给终端设备，主动权在服务端侧，终端设备通过ACK来反馈消息是否成功收到或处理，服务端需要根据终端是否返回ACK来决定是否重投。Pull，即终端设备主动来服务端获取其所有消息，主动权在终端设备侧，一般通过位点Offset来依次获取消息，RocketMQ就是这种消息获取方式。

对比两种方式，我们可以看到Pull方式需要终端设备主动管理消息获取逻辑，这个逻辑其实有一定的复杂性（可以参考RocketMQ的客户端管理逻辑），而终端设备运行环境和条件都很庞杂，不太适应较复杂的Pull逻辑实现，比较适合被动的Push方式。

另外，终端消息有一个很重要的区别是可靠性保证的ACK必须是具体到一个终端设备的，而服务端消息的可靠性在于只要有一台消费者机器成功处理即可，不太关心是哪台消费者机器，消息的可靠性ACK标识可以集中在消费组维度，而终端消息的可靠性ACK标识需要具体离散到终端设备维度。简单地说，一个是客户端设备维度的Retry队列，一个是消费组维度的Retry队列。

模型与组件

基于前面的消息基础一般性分析，我们来设计消息模型，主要是要解决好匹配查找和可靠触达两个核心问题。

（1）队列模型

消息能够可靠性触达的前提是要可靠存储，消息存储的目的是为了让接受者能获取到消息，接受者一般有两种消息检索维度：1）根据订阅的主题Topic去查找消息；2）根据订阅者ID去查找消息。这个就是业界常说的放大模型：读放大、写放大。

读放大：即消息按Topic进行存储，接受者根据订阅的Topic列表去相应的Topic队列读取消息。写放大：即消息分别写到所有订阅的接受者队列中，每个接受者读取自己的客户端队列。

可以看到读放大场景下消息只写一份，写到Topic维度的队列，但接受者读取时需要按照订阅的Topic列表多次读取，而写放大场景下消息要写多份，写到所有接受者的客户端队列里面，显然存储成本较大，但接受者读取简单，只需读取自己客户端一个队列即可。

我们采用的读放大为主，写放大为辅的策略，因为存储的成本和效率对用户的体感最明显。写多份不仅加大了存储成本，同时也对性能和数据准确一致性提出了挑战。但是有一个地方我们使用了写放大模式，就是通配符匹配，因为接受者订阅的是通配符和消息的Topic不是一样的内容，接受者读消息时没法反推出消息的Topic，因此需要在消息发送时根据通配符的订阅多写一个通配符队列，这样接受者直接可以根据其订阅的通配符队列读取消息。

上图描述的接受我们的队列存储模型，消息可以来自各个接入场景（如服务端的MQ/AMQP，客户端的MQTT），但只会写一份存到commitlog里面，然后分发出多个需求场景的队列索引（ConsumerQueue），如服务端场景（MQ/AMQP）可以按照一级Topic队列进行传统的服务端消费，客户端MQTT场景可以按照MQTT多级Topic以及通配符订阅进行消费消息。

这样的一个队列模型就可以同时支持服务端和终端场景的接入和消息收发，达到一体化的目标。

（2）推拉模型

介绍了底层的队列存储模型后，我们再详细描述一下匹配查找和可靠触达是怎么做的。

上图展示的是一个推拉模型，图中的P节点是一个协议网关或broker插件，终端设备通过MQTT协议连到这个网关节点。消息可以来自多种场景（MQ/AMQP/MQTT）发送过来，存到Topic队列后会有一个notify逻辑模块来实时感知这个新消息到达，然后会生成消息事件（就是消息的Topic名称），将该事件推送至网关节点，网关节点根据其连上的终端设备订阅情况进行内部匹配，找到哪些终端设备能匹配上，然后会触发pull请求去存储层读取消息再推送终端设备。

一个重要问题，就是notify模块怎么知道一条消息在哪些网关节点上面的终端设备感兴趣，这个其实就是关键的匹配查找问题。一般有两种方式：1）简单的广播事件；2）集中存储在线订阅关系（如图中的lookup模块），然后进行匹配查找再精准推送。事件广播机制看起来有扩展性问题，但是其实性能并不差，因为我们推送的数据很小就是Topic名称，而且相同Topic的消息事件可以合并成一个事件，我们线上就是默认采用的这个方式。集中存储在线订阅关系，这个也是常见的一种做法，如保存到Rds、Redis等，但要保证数据的实时一致性也有难度，而且要进行匹配查找对整个消息的实时链路RT开销也会有一定的影响。

可靠触达及实时性这块，上图的推拉过程中首先是通过事件通知机制来实时告知网关节点，然后网关节点通过Pull机制来换取消息，然后Push给终端设备。Pull+Offset机制可以保证消息的可靠性，这个是RocketMQ的传统模型，终端节点被动接受网关节点的Push，解决了终端设备轻量问题，实时性方面因为新消息事件通知机制而得到保障。

上图中还有一个Cache模块用于做消息队列cache，因为在大广播比场景下如果为每个终端设备都去发起队列Pull请求则对broker读压力较大，既然每个请求都去读取相同的Topic队列，则可以复用本地队列cache。

（3）lookup组件

上面的推拉模型通过新消息事件通知机制来解决实时触达问题，事件推送至网关的时候需要一个匹配查找过程，尽管简单的事件广播机制可以到达一定的性能要求，但毕竟是一个广播模型，在大规模网关节点接入场景下仍然有性能瓶颈。另外，终端设备场景有很多状态查询诉求，如查找在线状态，连接互踢等等，仍然需要一个KV查找组件，即lookup。

我们当然可以使用外部KV存储如Redis，但我们不能假定系统（产品）在用户的交付环境，尤其是专有云的特殊环境一定有可靠的外部存储服务依赖。

这个lookup查询组件，实际上就是一个KV查询，可以理解为是一个分布式内存KV，但要比分布式KV实现难度至少低一个等级。我们回想一下一个分布式KV的基本要素有哪些：

如上图所示，一般一个分布式KV读写流程是，Key通过hash得到一个逻辑slot，slot通过一个映射表得到具体的node。Hash算法一般是固定模数，映射表一般是集中式配置或使用一致性协议来配置。节点扩缩一般通过调整映射表来实现。

分布式KV实现通常有三个基本关键点：

1）映射表一致性

读写都需要根据上图的映射表进行查找节点的，如果规则不一致数据就乱了。映射规则配置本身可以通过集中存储，或者zk、raft这类协议保证强一致性，但是新旧配置的切换不能保证节点同时进行，仍然存在不一致性窗口。

2）多副本

通过一致性协议同步存储多个备份节点，用于容灾或多读。

3）负载分配

slot映射node就是一个分配，要保证node负载均衡，比如扩缩情况可能要进行slot数据迁移等。

我们主要查询和保存的是在线状态数据，如果存储的node节点宕机丢失数据，我们可以即时重建数据，因为都是在线的，所以不需要考虑多副本问题，也不需要考虑扩缩情况slot数据迁移问题，因为可以直接丢失重建，只需要保证关键的一点：映射表的一致性，而且我们有一个兜底机制——广播，当分片数据不可靠或不可用时退化到广播机制。

架构设计

基于前面的理论和模型分析介绍，我们在考虑用什么架构形态来支持一体化的目标，我们从分层、扩展、交付等方面进行一下描述。

（1）分层架构

我们的目标是期望基于RocketMQ实现一体化且自闭环，但不希望Broker被侵入更多场景逻辑，我们抽象了一个协议计算层，这个计算层可以是一个网关，也可以是一个broker插件。Broker专注解决Queue的事情以及为了满足上面的计算需求做一些Queue存储的适配或改造。协议计算层负责协议接入，并且要可插拔部署。

（2）扩展设计

我们都知道消息产品属于Paas产品，与上层Saas业务贴得最近，为了适应业务的不同需求，我们大致梳理一下关键的核心链路，在上下行链路上添加一些扩展点，如鉴权逻辑这个最偏业务化的逻辑，不同的业务需求都不一样，又比如Bridge扩展，其能够把终端设备状态和消息数据与一些外部生态系统（产品）打通。

（3）交付设计

好的架构设计还是要考虑最终的落地问题，即怎么交付。如今面临的现状是公共云、专有云，甚至是开源等各种环境条件的落地，挑战非常大。其中最大的挑战是外部依赖问题，如果产品要强依赖一个外部系统或产品，那对整个交付就会有非常大的不确定性。

为了应对各种复杂的交付场景，一方面会设计好扩展接口，根据交付环境条件进行适配实现；另一方面，我们也会尽可能对一些模块提供默认内部实现，如上文提到的lookup组件，重复造轮子也是不得已而为之，这个也许就是做产品与做平台的最大区别。

统一存储内核

前面对整个协议模型和架构进行了详细介绍，在Broker存储层这块还需要进一步的改造和适配。我们希望基于RocketMQ统一存储内核来支撑终端和服务端的消息收发，实现一体化的目标。

前面也提到了终端消息场景和服务端一个很大的区别是，终端必须要有个客户端维度的队列才能保证可靠触达，而服务端可以使用集中式队列，因为消息随便哪台机器消费都可以，但是终端消息必须明确可靠推送给具体客户端。客户端维度的队列意味着数量级上比传统的RocketMQ服务端Topic队列要大得多。

另外前面介绍的队列模型里面，消息也是按照Topic队列进行存储的，MQTT的Topic是一个灵活的多级Topic，客户端可以任意生成，而不像服务端场景Topic是一个很重的元数据强管理，这个也意味着Topic队列的数量级很大。

海量队列

我们都知道像Kafka这样的消息队列每个Topic是独立文件，但是随着Topic增多消息文件数量也增多，顺序写就退化成了随机写，性能下降明显。RocketMQ在Kafka的基础上进行了改进，使用了一个Commitlog文件来保存所有的消息内容，再使用CQ索引文件来表示每个Topic里面的消息队列，因为CQ索引数据较小，文件增多对IO影响要小很多，所以在队列数量上可以达到十万级。然而这终端设备队列场景下，十万级的队列数量还是太小了，我们希望进一步提升一个数量级，达到百万级队列数量，我们引入了Rocksdb引擎来进行CQ索引分发。

Rocksdb是一个广泛使用的单机KV存储引擎，具有高性能的顺序写能力。因为我们有了commitlog已具备了消息顺序流存储，所以可以去掉Rocksdb引擎里面的WAL，基于Rocksdb来保存CQ索引。在分发的时候我们使用了Rocksdb的WriteBatch原子特性，分发的时候把当前的MaxPhyOffset注入进去，因为Rocksdb能够保证原子存储，后续可以根据这个MaxPhyOffset来做Recover的checkpoint。我们提供了一个Compaction的自定义实现，来进行PhyOffset的确认，以清理已删除的脏数据。

轻量Topic

我们都知道RocketMQ中的Topic是一个重要的元数据，使用前要提前创建，并且会注册到namesrv上，然后通过Topicroute进行服务发现。前面说了，终端场景订阅的Topic比较灵活可以任意生成，如果基于现有的RocketMQ的Topic重管理逻辑显然有些困难。我们定义了一种轻量的Topic，专门支持终端这种场景，不需要注册namesrv进行管理，由上层协议逻辑层进行自管理，broker只负责存储。

总结

本文首先介绍了端云消息场景一体化的背景，然后重点分析了终端消息场景特点，以及终端消息场景支撑模型，最后对架构和存储内核进行了阐述。我们期望基于RocketMQ统一内核一体化支持终端和服务端不同场景的消息接入目标，以能够给使用者带来一体化的价值，如降低存储成本，避免数据在不同系统间同步带来的一致性挑战。

原文链接

本文为阿里云原创内容，未经允许不得转载。