简介：本次分享主要介绍了针对端端消息收发应用场景的架构模型设计，以及如何实现基于RocketMQ的一体化消息平台。作者：五环融合的背景不仅限于分布式我们都知道，以RocketMQ为代表的消息（队列）起源于不同应用服务之间的异步解耦通信，与以Dubbo为代表的RPC服务通信一起承载分布式系统间的通信场景（服务），所以服务之间的消息分发是消息的基本诉求。但是我们看到，在消息（队列）领域，我们行业这几年有一个非常重要的趋势，就是基于消息的数据可以扩展到流批计算、事件-驱动，如RocketMQ-streams、Kafka-Streams、Rabbit-Streams等不局限于服务端传统消息队列MQ主要用于服务（端）之间的消息通信，如交易消息、支付消息、物流消息等电子商务领域。但是在消息的范畴下，还有一个非常重要且常见的消息领域，那就是终端消息。消息的本质就是发送和接收，终端和服务器没有本质区别。集成价值如果有一个统一的消息系统（产品）提供多场景计算（如流、事件）和多场景（IoT、APP）接入，其实是非常有价值的，因为消息也是重要的数据，如果数据只存在于一个系统中，可以最大程度降低存储成本，同时可以有效避免不同系统之间数据同步带来的一致性问题。终端消息分析本文将主要介绍终端消息和服务端消息的集成设计与实践，所以首先对这一大类面向终端的消息做一个基础分析。场景介绍近年来，我们看到智能家居、工业互联等物联网设备端消息的爆发式增长，而发展了十余年的移动互联网移动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中，其实也有这个类似的匹配过程。通过rebalance将一个Queue分配给消费者组中的一台机器，消息通过Queue直接对应到消费者机器上，再通过订阅过滤（Tag或SQL）精准匹配消费者。之所以可以通过Queue来匹配消费机，是因为服务器场景消息不需要指定具体的消费机。一条消息可以放在任意一个Queue中，任意一个消费机都可以对应这个Queue。该消息不需要显式匹配消费者机器。在终端场景下，一条消息必须明确指定某个接收者（设备），所有接收者都必须准确找到，而终端设备一般只连接某个后端服务节点，即单连接，即与生成消息的节点不同。同样，必须有更复杂的匹配和寻找目标的过程，以及更灵活的匹配特性，例如MQTT通配符。(3)reach和reach是指匹配搜索后找到所有的接收者目标，消息需要以可靠的方式发送给接收者。常见的触发方式有两种：Push和Pull。推送是指服务器主动向终端设备推送消息。主动权在服务器端。终端设备使用ACK来反馈消息是否被成功接收或处理。服务器需要根据终端是否返回ACK来决定是否重新提交。拉取，即终端设备主动来到服务器获取其所有消息。主动权在终端设备侧。通常，消息是通过偏移点顺序获取的。RocketMQ就是这种获取消息的方法。对比两种方式可以看出，Pull方式需要终端设备主动管理消息获取逻辑。这个逻辑其实有些复杂（可以参考RocketMQ客户端管理逻辑），终端设备的运行环境和条件非常复杂。不适合复杂的Pull逻辑实现，更适合被动的Push方式。另外，终端消息的一个很重要的区别是可靠性保证的ACK必须是特定于一个终端设备的，而服务器端消息的可靠性是只要一台消费者机器成功处理就可以了，不太关心关于哪一台消费机，消息的可靠性ACK标识可以集中在消费者组维度，而终端消息的可靠性ACK标识需要具体离散到终端设备维度。简单的说，一个是clientdevice维度的Retryqueue，一个是consumergroup维度的Retryqueue。模型和组件基于前面消息基础的总体分析，我们设计消息模型，主要解决匹配搜索和可靠联系两个核心问题。(1)消息能够可靠到达的队列模型的前提是可靠存储。消息存储的目的是让接收者获得消息。接收者一般有两个消息检索维度：1）根据订阅的主题Topic，去寻找消息；2)根据订阅者ID查找消息。这就是业界常说的放大模型：读放大和写放大。读放大：即消息按主题存储，接收者根据订阅的主题列表从对应的主题队列中读取消息。写放大：即消息写入所有订阅的接收者队列，每个接收者读取自己的客户端队列。可以看到在读放大场景下，主题维度的队列中只写入了一份消息，但是接收方在读的时候需要根据订阅的主题列表多次读取，而在写的时候-放大场景，消息需要写成多份，写入到所有接收者的客户端队列中，存储成本明显高，但是接收者易读，只需要读自己的一个队列客户。我们采用读放大为主，写放大为辅的策略，因为存储的成本和效率对用户来说是最明显的。写入多个副本不仅会增加存储成本，还会对性能和数据的准确性和一致性提出挑战。但是有一个地方我们使用了写放大模式，就是通配符匹配，因为接收方订阅了一个通配符和消息的主题不一样，接收方在读取消息的时候不能反转消息的主题，所以在消息发送的时候需要根据通配符订阅写一个额外的通配符队列，这样接收方就可以直接根据自己订阅的通配符队列读取消息。接受我们上面描述的队列存储模型，消息可以来自各种访问场景（比如服务端的MQ/AMQP，客户端的MQTT），但是只会写入一份存储在commitlog中，然后分发到多种需求场景队列索引（ConsumerQueue），比如服务端场景（MQ/AMQP）可以根据一级Topic队列进行传统服务端消费，客户端MQTT场景可以根据MQTTmulti进行消息消费级主题和通配符订阅。这样的队列模型可以同时支持服务端和终端场景的访问和消息收发，达到一体化的目的。(2)推拉模型介绍完底层的队列存储模型，我们来详细描述一下匹配搜索和可靠访问是如何进行的。上图显示了一个推拉模型。图中的P节点是一个协议网关或者broker插件，终端设备通过MQTT协议连接到网关节点。可以从多种场景（MQ/AMQP/MQTT）发送消息。存入Topic队列后，会有notify逻辑模块实时感知新消息的到来，然后产生消息事件（即消息的主题名）。事件推送到网关节点，网关节点根据连接的终端设备的订阅状态进行内部匹配，找到可以匹配到哪些终端设备，然后触发拉取请求到存储层读取消息和推送终端设备。一个重要的问题是通知模块如何知道网关节点上的哪些终端设备对消息感兴趣。这其实就是关键匹配搜索问题。一般有两种方式：1）简单的广播事件；2）集中存储在线订阅关系（如图lookup模块），然后进行匹配搜索和精准推送。事件广播机制貌似有扩展性问题，但是性能还不错，因为我们推送的数据很小，只是主题名，同主题的消息事件可以合并为一个事件，这是默认的方式我们在线使用。在线订阅关系的集中存储也是一种常见的做法，比如保存到Rds、Redis等，但是也很难保证数据的实时一致性，而且会有实时链路RT开销要搜索的整个消息以匹配特定影响。为了可靠到达和实时性，在上图中的推拉过程中，首先通过事件通知机制实时通知网关节点，然后网关节点使用Pull机制交换消息，然后推送到终端设备。Pull+Offset机制可以保证消息的可靠性。这是RocketMQ的传统模型。终端节点被动接受网关节点的Push，解决了终端设备轻量化的问题。实时性由新的消息事件通知机制保证。上图中还有一个Cache模块作为消息队列的缓存，因为在广播比大的场景下，如果对每个终端设备发起队列Pull请求，broker的读取压力会更大。由于每次请求都读取同一个主题队列，可以复用本地队列缓存。(3)查找组件上的推拉模型，通过新的消息事件通知机制解决了实时访问的问题。当事件被推送到网关时，需要一个匹配的搜索过程。简单的事件广播机制虽然可以满足一定的性能需求，但毕竟是一种广播模型，在大规模网关节点接入的场景下还是存在性能瓶颈的。另外，终端设备场景有很多状态查询需求，比如查找在线状态、连接踢对方等，还需要一个KV查找组件，即lookup。当然，我们可以使用Redis等外部KV存储，但我们不能假设系统（产品）在用户交付环境，尤其是专有云的特殊环境下，必须依赖可靠的外部存储服务。这个查找查询组件其实就是一个KV查询，可以理解为分布式内存KV，但是比分布式KV的实现难度至少低了一个层次。我们回顾一下分布式KV的基本要素有哪些：如上图所示，一般一个分布式KV的读写过程是key通过hash得到一个逻辑槽，槽通过映射表得到一个具体的节点。Hash算法一般是固定模数，映射表一般采用中心化方式或采用共识协议配置。节点缩放一般通过调整映射表来实现。分布式KV实现通常有3个基本关键点：1）映射表的一致性读写需要根据上图中的映射表找到节点。如果规则不一致，数据就会乱七八糟。映射规则配置本身可以集中存储，或者zk、raft等协议可以保证强一致性，但是新旧配置的切换不能保证节点同时执行，仍然存在一个不一致的窗口。2）多副本通过一致性协议同步存储多个备份节点，用于容灾或多读。3）负载分配槽映射节点是一个分配，保证节点负载均衡，如扩容和收缩可能需要槽数据迁移等。我们主要查询和保存在线状态数据。如果存储的node节点宕机丢失数据，我们可以马上重建数据。因为都是在线的，所以我们不需要考虑多副本的问题，也不需要考虑slot数据的扩容和扩容。迁移问题，因为可以直接丢失重建，只需要保证一个关键点：映射表的一致性，我们有一个来回机制——广播，分片时退化为广播机制数据不可靠或不可用。架构设计是基于前面的理论和模型分析。我们正在考虑使用什么架构来支持集成的目标。我们将从分层、扩展、交付等方面进行描述。(1)分层架构我们的目标是基于RocketMQ实现集成和自闭环，但不希望Broker被侵入到更多的场景逻辑中。我们抽象出一个协议计算层，可以是网关，也可以是Abrokerplugin。Broker专注于解决Queue问题，做一些Queue存储适配或改造，以满足上述计算需求。协议计算层负责协议接入，必须可插拔部署。（2）扩展设计我们都知道消息产品属于PaaS产品，最接近上层Saas业务。为了满足业务的不同需求，我们大致梳理了关键的核心环节，在上行和下行增加了一些扩展点。比如鉴权逻辑，它是最面向业务的逻辑，不同的业务需求。再比如Bridgeextension，可以将终端设备状态和消息数据与一些外部生态系统（产品）连接起来。(3)交付设计良好的架构设计，必须考虑最终的实现问题，即如何交付。我们今天面临的现状是公有云、私有云、甚至开源等各种环境条件的实施，挑战非常大。最大的挑战是外部依赖的问题。如果一个产品强烈依赖于外部系统或产品，那么整个交付就会存在很大的不确定性。为应对各种复杂的交付场景，一方面会根据交付环境条件设计和实现扩展接口；另一方面，我们会尽量为一些模块提供默认的内部实现，比如上面提到的查找组件，重复造轮子也是不得已而为之。这可能是做产品和做平台最大的区别。统一存储内核详细介绍了整个协议模型和架构，Broker存储层需要进一步改造适配。我们希望基于RocketMQ统一存储内核，支持终端与服务端之间的消息收发，达到一体化的目的。前面提到，终端消息场景和服务端一个很大的区别是，终端必须有一个客户端维度的队列来保证可靠到达，而服务端可以使用集中式队列，因为消息可以被任何机器消费。是的，但必须将终端消息清晰可靠地推送到特定客户端。客户端队列意味着它在量级上比传统的RocketMQ服务端Topic队列要大很多。另外，在前面介绍的队列模型中，消息也是按照主题队列进行存储的。MQTT的主题是灵活的多级主题，可以由客户端随意生成，不像服务端场景中的主题是重元数据，管理性强，这也意味着Topic队列的数量级很大。海量队列我们都知道像Kafka这样的消息队列的每个topic都是一个独立的文件，但是随着topic数量的增加，消息文件的数量也随之增加，顺序写退化为随机写，性能明显下降。RocketMQ在Kafka的基础上进行了改进。它使用一个Commitlog文件保存所有的消息内容，然后使用CQ索引文件来表示每个主题中的消息队列。因为CQ索引数据小，所以文件越多对IO的影响越大。它要小得多，因此队列数可以达到100,000。但是在终端设备队列的场景下，10万级别的队列数量还是太少了。我们希望进一步增加队列数量一个数量级，达到百万级队列数量。我们引入了Rocksdb引擎来分发CQ索引。RocksDB是一款应用广泛的单机KV存储引擎，具有高性能的顺序写入能力。因为我们有一个已经有消息序列流存储的commitlog，所以我们可以去掉Rocksdb引擎中的WAL，保存基于Rocksdb的CQ索引。在分发的时候，我们利用了Rocksdb的WriteBatch原子特性，在分发的时候把当前的MaxPhyOffset注入进去，因为Rocksdb可以保证原子存储，后续恢复的checkpoints可以根据这个MaxPhyOffset来做。我们为PhyOffset确认提供了Compaction的自定义实现，以清理已删除的脏数据。轻量级Topic我们都知道RocketMQ中的Topic是一个重要的元数据，在使用前必须提前创建，并注册到namesrv中，然后通过Topicroute进行服务发现。前面提到，终端场景下订阅的主题是灵活的，可以任意生成。基于已有的RocketMQ主题重新管理主题显然是有难度的。我们定义了一个专门支持终端场景的轻量级topic。它不需要注册namesrv来进行管理。上层协议逻辑层进行自我管理，broker只负责存储。总结本文首先介绍了端云消息场景融合的背景，然后着重分析了终端消息场景的特点和终端消息场景的支撑模型，最后阐述了架构和存储核心。我们希望基于RocketMQ统一内核，能够支持终端和服务端不同场景下的消息访问目标，从而为用户带来综合价值，比如降低存储成本，避免不同平台之间数据同步带来的一致性问题。系统。挑战。原文链接本文为阿里云原创内容，未经许可不得转载。
