探析RocketMQ内部成分

Posted 2021-11-06 毛奇志

文章目录

• 一、前言
• 二、 RocketMQ简介
• • 2.1 RocketMQ是阿里的开源消息中间件
  • 2.2 RocketMQ处理高并发做了两件事情
  • 2.3 RocketMQ优缺点以及业务用途
  • 2.4 RocketMQ项目结构
• 三、RocketMQ分布式架构
• • 3.1 Rocket的分布式架构
  • 3.2 NameServer
  • 3.3 Producer
  • 3.4 Broker
  • 3.5 Consumer
• 四、RocketMQ消息领域模型
• 五、Rocket内部成分
• • 5.1 RocketMQ如何保证消息的可用性
  • 5.2 RocketMQ如何存储消息
  • 5.3 RocketMQ的生产者如何保证消息顺序发送
  • 5.4 RocketMQ实现分布式事务
  • 5.5 RocketMQ消息过滤
  • 5.6 RocketMQ的缓存问题
  • 5.7 RocketMQ的回溯消费/重复消费
  • 5.8 RocketMQ的消息堆积问题
  • 5.9 RocketMQ的定时消息
• 六、尾声

一、前言

二、 RocketMQ简介

2.1 RocketMQ是阿里的开源消息中间件

RocketMQ是一个纯Java、分布式、队列模型的开源消息中间件，前身是MetaQ，是阿里参考Kafka特点研发的一个队列模型的消息中间件（RocketMQ是阿里开源其自研的第三代分布式消息中间件），后开源给apache基金会成为了apache的顶级开源项目，具有高性能、高可靠、高实时、分布式特点。

RocketMQ英文直译：Rocket火箭、MQ message queue 消息队列。Apache基金会中的342个项目中，暂时还只有Kylin、CarbonData、Eagle 、Dubbo和 RocketMQ 共计五个中国技术人主导的项目，我们比较熟悉的是Dubbo和RocketMQ，都是阿里的，RocketMQ可以实用于电商领域，金融领域，大数据领域，这些领域正好是阿里的专长。

阿里巴巴内部围绕着RocketMQ内核打造了三款产品，分别是MetaQ、Notify和Aliware MQ，这三者分别采用了不同的模型，MetaQ主要使用了拉模型，解决了顺序消息和海量堆积问题；Notify主要使用了推模型，解决了事务消息；而云产品Aliware MQ则是提供了商业化的版本。

2.2 RocketMQ处理高并发做了两件事情

说到高并发，就是双11洗礼，为了在高并发下保护数据库，RocketMQ团队重点做了两件事情，优化慢请求与统一存储引擎。

（1）优化慢请求：这里主要是解决在海量高并发场景下降低慢请求对整个集群带来的抖动，毛刺问题。这是一个极具挑战的技术活，团队同学经过长达1个多月的跟进调优，从双十一的复盘情况来看，99.996%的延迟落在了10ms以内，而99.6%的延迟在1ms以内。优化主要集中在RocketMQ存储层算法优化、JVM与操作系统调优。更多的细节大家可以参考《万亿级数据洪峰下的分布式消息引擎》。

（2）统一存储引擎：主要解决的消息引擎的高可用，成本问题。在多代消息引擎共存的前提下，我们对Notify的存储模块进行了全面移植与替换。

RocketMQ天生为金融互联网领域而生，追求高可用、高并发、低延迟，是一个阿里巴巴由内而外成功孕育的典范，RocketMQ在阿里集团也被广泛应用在订单，交易，充值，流计算，消息推送，日志流式处理，binglog分发等场景。

2.3 RocketMQ优缺点以及业务用途

RocketMQ优点：
(1) 单机吞吐量：十万级
(2) 可用性：非常高，分布式架构
(3) 消息可靠性：经过参数优化配置，消息可以做到0丢失
(4) 功能支持：MQ功能较为完善，还是分布式的，扩展性好
(5) 支持10亿级别的消息堆积，不会因为堆积导致性能下降
(6) 源码是java，我们可以自己阅读源码，定制自己公司的MQ，可以掌控
(7) 天生为金融互联网领域而生，对于可靠性要求很高的场景，尤其是电商里面的订单扣款，以及业务削峰，在大量交易涌入时，后端可能无法及时处理的情况
(8) RoketMQ在稳定性上可能更值得信赖，这些业务场景在阿里双11已经经历了多次考验，如果你的业务有上述并发场景，建议可以选择RocketMQ

RocketMQ缺点：
(1) 支持的客户端语言不多，目前是java及c++，其中c++不成熟
(2) 社区活跃度不是特别活跃那种
(3) 没有在 mq 核心中去实现JMS等接口，有些系统要迁移需要修改大量代码

RocketMQ的业务用途：
(1) 发布/订阅消息传递模型
(2) 财务级交易消息
(3) 各种跨语言客户端，例如Java，C / C ++，Python，Go
(4) 可插拔的传输协议，例如TCP，SSL，AIO
(5) 内置的消息跟踪功能，还支持开放式跟踪
(6) 多功能的大数据和流生态系统集成
(7) 按时间或偏移量追溯消息
(8) 可靠的FIFO和严格的有序消息传递在同一队列中
(9) 高效的推拉消费模型
(10) 单个队列中的百万级消息累积容量
(11) 多种消息传递协议，例如JMS和OpenMessaging
(12) 灵活的分布式横向扩展部署架构
(13) 快如闪电的批量消息交换系统
(14) 各种消息过滤器机制，例如SQL和Tag
(15) 用于隔离测试和云隔离群集的Docker映像
(16) 功能丰富的管理仪表板，用于配置，指标和监视
(17) 认证与授权

2.4 RocketMQ项目结构

GitHub地址：https://github.com/apache/rocketmq

RocketMQ核心模块(下载源码，idea打开)：
rocketmq-broker：接受生产者发来的消息并存储（通过调用rocketmq-store），消费者从这里取得消息。
rocketmq-client：提供发送、接受消息的客户端API。
rocketmq-namesrv：NameServer，类似于Zookeeper，这里保存着消息的TopicName，队列等运行时的元信息。
rocketmq-common：通用的一些类，方法，数据结构等。
rocketmq-remoting：基于Netty4的client/server + fastjson序列化 + 自定义二进制协议。
rocketmq-store：消息、索引存储等。
rocketmq-filtersrv：消息过滤器Server，需要注意的是，要实现这种过滤，需要上传代码到MQ！（一般而言，我们利用Tag足以满足大部分的过滤需求，如果更灵活更复杂的过滤需求，可以考虑filtersrv组件）。
rocketmq-tools：命令行工具。

三、RocketMQ分布式架构

3.1 Rocket的分布式架构

Rocket的分布式架构是高并发、高效率、高可用的保证，它主要有四大核心组成部分：NameServer、Broker、Producer以及Consumer四部分。

RocketMQ的四个部分（NameServer、Broker、Producer以及Consumer）都是采用分布式集群，这是他高并发（吞吐量大），高可用的原因之一，集群的模式有多种，包括多master 模式、多master多slave异步复制模式、多 master多slave同步双写模式。

这个模式很像Kafka，因为就是阿里参考Kafka特点研发的一个队列模型的消息中间件。

3.2 NameServer

NameServer定义：主要负责对于源数据的管理，包括了对于Topic和路由信息的管理。

NameServer的地位：NameServer在RocketMQ中是注册中心，Zookeeper在Dubbo也是注册中心。NameServer是一个功能齐全的服务器，从功能上看，类似Dubbo中的Zookeeper，但NameServer与Zookeeper相比更轻量。主要是因为每个NameServer节点互相之间是独立的，没有任何信息交互。

NameServer心跳：NameServer压力不会太大，平时主要开销是在维持心跳和提供Topic-Broker的关系数据。但是，每个Broker向NameServer发心跳时， 会带上当前自己负责的所有Topic信息，如果一个Broker中的Topic个数太多（万级别），会导致一次心跳中，就Topic的数据就几十M，网络情况差的话， 网络传输失败，心跳失败，导致NameServer误认为Broker心跳失败。

NameServer 是无状态的，可以横向扩展，成为一个注册中心集群：NameServer 被设计成几乎无状态的，可以横向扩展，节点之间相互之间无通信，通过部署多台机器来标记自己是一个伪集群。

每个 Broker 在启动的时候会到 NameServer 注册：每个 Broker 在启动的时候会到 NameServer 注册，然后，各个Broker将信息注册到NameServer，然后Producer和Consumer就可以来NameServer取Broker的信息了，即每个 Producer 在发送消息前会根据 Topic 到 NameServer 获取到 Broker 的路由信息，每个Consumer 也会定时获取 Topic 的路由信息。

NameServer作为RocketMQ的注册中心，在交互逻辑上，和Dubbo中注册中心的角色，几乎一模一样。

3.3 Producer

Producer定义：消息生产者，负责产生消息，一般由业务系统负责产生消息。

Producer分布式部署：Producer由用户进行分布式部署，消息由Producer通过多种负载均衡模式发送到Broker集群，发送低延时，支持快速失败。

Producer拥有了三种方式发送消息：同步发送、异步发送和单向发送

同步双向发送定义：同步发送指消息发送方Producer发出数据后，一定要收到接收方发回响应之后才发下一个数据包。
同步双向发送特点：可靠性最好，耗时最长，注重可靠性高，适合发重要信息。
同步双向发送用途：用于重要通知消息，例如重要通知邮件、营销短信。

异步双向发送定义：异步发送指消息发送方Producer发出数据后，不等接收方发回响应，接着发送下个数据包。
异步双向发送特点：可靠性稍差，耗时较长。
异步双向发送用途：用于可能链路耗时较长而对响应时间敏感的业务场景，例如用户视频上传后，消息队列发送消息，通知启动转码服务。

单向发送定义：单向发送是指只负责发送消息而不等待服务器回应且没有回调函数触发，单向，发送方自己发自己的，不管接收方。
单向发送特点：可靠性最差，但是耗时最短，注重耗时短。
单向发送用途：适用于某些耗时非常短但对可靠性要求并不高的场景，例如日志收集。

3.4 Broker

Broker定义：消息中转角色，负责存储消息，转发消息。

Broker是具体提供业务的服务器，单个Broker节点与所有的NameServer节点保持长连接及心跳，并会定时将Topic信息注册到NameServer，顺带一提底层的通信和连接都是基于Netty实现的

(1)netty基于NIO,网络通信高效率
(2)阿里的两种开源中间件，Dubbo和RocketMQ的结构相同，都需要网络通信，网络通信都是netty。

Broker负责消息存储，以Topic为纬度支持轻量级的队列，单机可以支撑上万队列规模，支持消息推拉模型。官网上有数据显示：具有上亿级消息堆积能力，同时可严格保证消息的有序性。

3.5 Consumer

Consumer定义：消息消费者，负责消费消息，一般是后台系统负责异步消费。

Consumer也由用户部署，支持PUSH和PULL两种类型的消费模式，支持集群消费和广播消息，提供实时的消息订阅机制。
Pull：拉取型消费者（Pull Consumer）主动从消息服务器拉取信息，只要批量拉取到消息，用户应用就会启动消费过程，所以 Pull 称为主动消费型。
Push：推送型消费者（Push Consumer）封装了消息的拉取、消费进度和其他的内部维护工作，将消息到达时执行的回调接口留给用户应用程序来实现。所以 Push 称为被动消费类型，但从实现上看还是从消息服务器中拉取消息，不同于 Pull 的是 Push 首先要注册消费监听器，当监听器处触发后才开始消费消息。

四、RocketMQ消息领域模型

第一，Message 消息
Message本质：就是消息，就是要传输的信息。
Message与Topic关系：一条消息必须有一个主题（Topic），主题可以看做是你的信件要邮寄的地址。
Message与Tag的关系：一条消息也可以拥有一个可选的标签（Tag）和额处的键值对，它们可以用于设置一个业务 Key 并在 Broker 上查找此消息以便在开发期间查找问题。

第二，Topic 主题
Topic定义：Topic 可以看做消息的分类，它是消息的第一级类型。比如一个电商系统可以分为：交易消息、物流消息等，
Topic与Message的关系：一条消息必须有一个 Topic 。
Topic与Producer的关系：一个 Topic 可以有0个、1个、多个生产者向其发送消息，一个生产者也可以同时向不同的 Topic 发送消息。
Topic与Consumer的关系：一个 Topic 也可以被 0个、1个、多个消费者订阅。

第三，Group 分组
Group定义：分为ProducerGroup，ConsumerGroup，代表某一类的生产者和消费者，一般来说同一个服务可以作为Group，同一个Group一般来说发送和消费的消息都是一样的
Group与Topic的关系：一个组可以订阅多个Topic。

第四，Queue 队列 和 Message Queue 消息队列
Queue队列：在Kafka中叫Partition，每个Queue内部是有序的，在RocketMQ中分为读和写两种队列，一般来说读写队列数量一致，如果不一致就会出现很多问题。
Message Queue消息队列：主题被划分为一个或多个子主题，即子主题就是消息队列。
消息队列和Topic的关系：一个 Topic 下可以设置多个消息队列，发送消息时执行该消息的 Topic ，RocketMQ 会轮询该 Topic 下的所有队列将消息发出去。
消息队列是消息的物理管理单位：一个Topic下可以有多个Queue，Queue的引入使得消息的存储可以分布式集群化，具有了水平扩展能力。

第五，Offset 偏移量
在RocketMQ 中，所有消息队列都是持久化，长度无限的数据结构，所谓长度无限是指队列中的每个存储单元都是定长，访问其中的存储单元使用Offset 来访问，Offset 为 java long 类型，64 位，理论上在 100年内不会溢出，所以认为是长度无限。也可以认为 Message Queue 是一个长度无限的数组，Offset 就是下标。

第六，Tag 标签
Tag定义：Tag 可以看作子分类/子主题，它是消息的第二级类型，用于为用户提供额外的灵活性。使用标签，同一业务模块不同目的的消息就可以用相同 Topic 而不同的 Tag 来标识。比如交易消息又可以分为：交易创建消息、交易完成消息等，
Tag意义：标签有助于保持您的代码干净和连贯，并且还可以为 RocketMQ 提供的查询系统提供帮助。
Tag与Message的关系：一条消息可以没有 Tag 。

第七，消息消费模式
消息消费模式有两种：Clustering（集群消费）和Broadcasting（广播消费）。
第一，集群消费模式：默认情况下就是集群消费，该模式下一个消费者集群共同消费一个主题的多个队列，一个队列只会被一个消费者消费，如果某个消费者挂掉，分组内其它消费者会接替挂掉的消费者继续消费。
第二，广播消费模式：广播消费消息会发给消费者组中的每一个消费者进行消费。

第八，Message Order
Message Order（消息顺序）有两种：Orderly（顺序消费）和Concurrently（并行消费）。
第一，顺序消费：顺序消费表示消息消费的顺序同生产者为每个消息队列发送的顺序一致，所以如果正在处理全局顺序是强制性的场景，需要确保使用的主题只有一个消息队列。
第二，并行消费：并行消费不再保证消息顺序，消费的最大并行数量受每个消费者客户端指定的线程池限制。

五、Rocket内部成分

5.1 RocketMQ如何保证消息的可用性

问题：RocketMQ如何保证消息的可用性 ？
回到：刷盘 + 主从同步。
1、刷盘-内存中的消息持久化到磁盘（同步刷盘+异步刷盘）：当我们选择好了集群模式之后，那么我们需要关心的就是怎么去存储和复制这个数据，只有讲broker内存中的消息持久化到磁盘，才能保证broker宕机，消息不丢失，RocketMQ对消息的持久化到磁盘，提供了同步和异步的策略来满足我们的，
情况1：选择同步刷盘，如果刷盘超时会给返回 刷盘超时 FLUSH_DISK_TIMEOUT，
情况2：选择异步刷盘，如果是异步刷盘不会返回刷盘成功与否的任何信息，
所以，选择同步刷盘可以尽最大程度保证刷盘的时候消息不会丢失。
2、主从partition复制（同步复制+异步复制）：RocketMQ的主从同步提供了同步复制和异步复制两种模式，当然选择同步复制可以提升可用性，但是消息的发送RT时间会下降10%左右。

RocketMQ两种刷盘方式的最终实现都是使用NIO中的 MappedByteBuffer.force() 将映射区的数据写入到磁盘，如果是同步刷盘，在Broker把消息写到CommitLog映射区后，就会等待写入完成。如果是异步刷盘，只是唤醒对应的线程，不保证执行的时机。

5.2 RocketMQ如何存储消息

问题：RocketMQ如何存储消息？
回答：RocketMQ混合型存储结构。RocketMQ采用的是混合型的存储结构，：为Broker单个实例下所有的队列共用一个日志数据文件（即为CommitLog）来存储。会存在较多的随机读操作，因此读的效率偏低，同时消费消息需要依赖ConsumeQueue，构建该逻辑消费队列需要一定开销。

Kafka采用的是独立型的存储结构，即每个队列/分区一个文件，这样就不会存在随机读操作了，都是顺序读。

5.3 RocketMQ的生产者如何保证消息顺序发送

问题：RocketMQ的生产者如何保证消息顺序发送？
生产者消费者一般需要保证顺序消息的话，可能就是一个业务场景下的，比如订单的创建、支付、发货、收货。那这些东西是不是一个订单号呢？一个订单的肯定是一个订单号的说，那简单了呀。
一个topic下有多个队列，为了保证发送有序，RocketMQ提供了MessageQueueSelector队列选择机制，他有三种实现:SelectMessageQueueByHash
SelectMessageQueueByMachineRoom
SelectMessageQueueByRandom
以SelectMessageQueueByHash实现类为例，使用Hash取模法，让同一个订单发送到同一个队列中，再使用同步发送，只有同个订单的创建消息发送成功，再发送支付消息。这样，我们保证了发送有序。RocketMQ的topic内的队列机制,可以保证存储满足FIFO（First Input First Output 简单说就是指先进先出）,剩下的只需要消费者顺序消费即可。

RocketMQ仅保证顺序发送，顺序消费由消费者业务保证。
解释：一个订单你发送的时候放到一个队列里面去，你同一个的订单号Hash一下是不是还是一样的结果，那肯定是一个消费者消费，那顺序就可以保证了。

5.4 RocketMQ实现分布式事务

问题：RocketMQ如何实现支持分布式事务？
回答：一图顶千言，如下：

对于上图解释 1-8
1、A服务先发送个Half Message给Brock端，消息中携带 B服务 即将要+100元的信息。
2、服务端发送响应，半消息接收成功，
3、执行本地事务(会有三种情况1、执行成功。2、执行失败。3、网络等原因导致没有响应)
4、服务A根据本地事务，发送 commit/rollback 给broker，转到8.
5、如果因为网络等原因迟迟没有返回失败还是成功，那么会执行RocketMQ的回调接口,来进行事务的回查。
6、服务A检查本地事务；
7、服务A根据本地事务，发送 commit/rollback 给broker，转到8.
8、如果本地事务成功，那么Product像Brock服务器发送Commit,这样B服务就可以消费该message；如果本地事务失败，那么Product像Brock服务器发送Rollback,那么就会直接删除上面这条半消息。

步骤1：Producer发送半消息给Broker(对应图中的 1 2 3 4)
是指暂不能被Consumer消费的消息。Producer 已经把消息成功发送到了 Broker 端，但此消息被标记为暂不能投递状态，处于该种状态下的消息称为半消息。需要 Producer对消息的二次确认后，Consumer 才能去消费它。
步骤2：消息回查，Broker询问Producer(对应图中的 5 6 7 8)
由于网络闪段，生产者应用重启等原因。导致 Producer 端一直没有对 Half Message(半消息) 进行 二次确认。这是Brock服务器会定时扫描长期处于半消息的消息，会主动询问 Producer端 该消息的最终状态(Commit或者Rollback),该消息即为 消息回查。
未使用消息回查的流程 1 2 3 4 8
已使用消息回查的流程 1 2 3 4 5 6 7 8

5.5 RocketMQ消息过滤

问题：RocketMQ的消息过滤机制？

回答：第一，Broker端消息过滤　　
定义：在Broker中，按照Consumer的要求做过滤；
优点：减轻网络传输负担，减少了对于Consumer无用消息的网络传输；
缺点：增加Broker的负担，实现相对复杂。

第二，Consumer端消息过滤
定义：Consumer端，程序员完全自定义实现过滤规则来过滤；
优点：减轻Broker的负担。
缺点：增加网络传输负担，很多无用的消息要传输到Consumer端。

5.6 RocketMQ的缓存问题

问题：RocketMQ中，Broker的Buffer相关知识？
回答：RocketMQ,读写磁盘代替内存buffer。Broker的Buffer通常指的是Broker中一个队列的内存缓冲Buffer大小，这个Buffer通常大小有限。值得注意的是，RocketMQ同其他MQ有非常显著的区别，RocketMQ没有内存Buffer概念，RocketMQ的队列都是持久化磁盘。从这方面说，RocketMQ的内存Buffer抽象成一个无限长度的队列(理由：磁盘可以不断存入)，不管有多少数据进来都能装得下，这个无限是有前提的，Broker会定期删除过期的数据。例如Broker只保存3天的消息，那么这个Buffer虽然长度无限，但是3天前的数据会被从队尾删除。

问题：为什么RocketMQ没有内存Buffer?
回答：不需要，RocketMQ的队列都是持久化磁盘，不需要的时候读盘，为了防止磁盘爆满，数据定期清除。

5.7 RocketMQ的回溯消费/重复消费

问题：消息队列的 “回溯消费/重复消费” 吗？
回答：回溯消费是指Consumer已经消费成功的消息，由于业务上的需求需要重新消费，要支持此功能，Broker在向Consumer投递成功消息后，消息仍然需要保留。并且重新消费一般是按照时间维度，隔一段时间后回溯消费/重复消费。RocketMQ支持按照时间回溯消费，时间维度精确到毫秒，可以向前回溯，也可以向后回溯。例如由于Consumer系统故障，恢复后需要重新消费1小时前的数据，那么Broker要提供一种机制，可以按照时间维度来回退消费进度。

5.8 RocketMQ的消息堆积问题

问题：消息队列中的消息堆积问题？
回答：消息中间件三个功能：异步、解耦、流量控制，消息堆积主要是涉及流量控制功能，流量控制就是挡住前端的数据洪峰，保证后端系统的稳定性，这就要求消息中间件具有一定的消息堆积能力，消息堆积分以下两种情况：
1、消息堆积在broker的内存缓冲Buffer，一旦超过内存缓冲Buffer，可以根据一定的丢弃策略来丢弃消息，如CORBA Notification规范中描述。适合能容忍丢弃消息的业务，这种情况消息的堆积能力主要在于内存Buffer大小，而且消息堆积后，性能下降不会太大，因为内存中数据多少对于对外提供的访问能力影响有限。
2、消息堆积到持久化存储系统中，例如DB，KV存储，文件记录形式。当消息不能在内存Cache命中时，要不可避免的访问磁盘，会产生大量读IO，读IO的吞吐量直接决定了消息堆积后的访问能力。

所以，评估消息堆积能力主要有以下四点：
（1）内存中，消息能堆积多少条，多少字节？即消息的堆积容量。
（2）消息堆积后，发消息的吞吐量大小，是否会受堆积影响？消息堆积在内存中吞吐量不受影响，消息堆积在磁盘持久化(db kv存储 文件记录存储)，吞吐量受影响。
（3）消息堆积后，正常消费的Consumer是否会受影响？
（4）消息堆积后，访问堆积在磁盘的消息时，吞吐量有多大？

5.9 RocketMQ的定时消息

问题：RocketMQ的定时消息
回答：定时消息是指消息发到Broker后，不能立刻被Consumer消费，要到特定的时间点或者等待特定的时间后才能被消费。RocketMQ支持定时消息，但是不支持任意时间精度，支持特定的level，例如定时5s，10s，1m等。

问题：为什么不支持任意时间精度？
回答：如果要支持任意的时间精度，在Broker层面，必须要做消息排序，如果再涉及到持久化，那么消息排序要不可避免的产生巨大性能开销。

六、尾声

探析RocketMQ内部原理，完成了。

天天打码，天天进步！！！