团基于高可用分布式集群技术自主研发的云正式商用的专业消息中间件既可为分布式应用系统提供异步解耦和削峰填谷的能力同时也具备互联网应用所需的海量消息堆积、高吞吐、可靠重试等特性是阿里巴巴双

1.1

如果想要了解RocketMQ的历史则需了解阿里巴巴中间件团队中的历史。

2011年Linkin(领英全球知名的职场社交平台)推出Kafka消息引擎阿里巴巴中间件团队在研究了Kafka的整体机制和架构设计之后基于Kafka(Scala语言编写)的设计使用Java进行了完全重写并推出了MetaQ

1.0版本主要是用于解决顺序消息和海量堆积的问题由开源社区killme2008维护。

课程重点不在此版本具体见https://github.com/killme2008/Metamorphosis

2012年阿里巴巴发现MetaQ原本基于Kafka的架构在阿里巴巴如此庞大的体系下很难进行水平扩展于是对MetaQ进行了架构重组升级开发出了MetaQ

2.0同年阿里把Meta2.0从阿里内部开源出来取名RocketMQ为了命名上的规范以及版本上的延续对外称为RocketMQ3.0。

因为RocketMQ3只是RocketMQ的一个过渡版本课程重点也不在此。

2016年11月28日阿里巴巴宣布将开源分布式消息中间件RocketMQ捐赠给Apache成为Apache

孵化项目。

在孵化期间RocketMQ完成编码规约、分支模型、持续交付、发布规约等方面的产品规范化同时RocketMQ3也升级为RocketMQ4。

现在RocketMQ主要维护的是4.x的版本也是大家使用得最多的版本,所以本书重点将围绕此版本进行详细的讲解项目地址https://github.com/apache/rocketmq/

2015年阿里基于RocketMQ开发了阿里云上的Aliware

MQAliware

Queue)是RocketMQ的商业版本是阿里云商用的专业消息中间件是企业级互联网架构的核心产品基于高可用分布式集群技术搭建了包括发布订阅、消息轨迹、资源统计、定时延时、监控报警等一套完整的消息云服务。

因为Aliware

MQ是商业版本课程也不对此产品进行讲述产品地址https://www.aliyun.com/product/rocketmq

2021年伴随众多企业全面上云以及云原生的兴起RocketMQ也在github上发布5.0版本。

目前来说还只是一个预览版不过RocketMQ5的改动非常大同时也明确了版本定位RocketMQ

5.0定义为云原生的消息、事件、流的超融合平台。

本课程也将会根据目前所发布的版本进行针对性的讲述。

1.1.1

版本之后正式发布事务消息通过类似于两阶段的方式去解决上下游数据不一致问题。

在

RocketMQ

增加了消息轨迹的功能使用户可以更好定位每一条消息的投放接收路径帮助问题排查另外还增加

ACL

组内就拥有了自动故障转移的能力也解决了像高可用顺序消息这样的问题进一步提高了

RocketMQ

模式有了一个质的提升包括通过异步化、批量复制等手段将性能提升了数倍在稳定性上利用

OpenChaos

完成包括宕机、杀死进程OOM、各种各样的网络分区和延迟的测试修复了重要

Preferred

RocketMQ作为阿里双十一交易核心链路产品支撑千万级并发、万亿级数据洪峰。

读源码可以积累编写高效、优雅代码的经验。

Apache

RocketMQ作为Apache顶级项目它的架构设计是值得大家借鉴的。

在使用RocketMQ过程中遇到消费卡死、卡顿等问题可以通过阅读源码的方式找到问题并给予解决。

大厂面试中尤其是阿里系的公司你有RocketMQ源码体系化知识必定是一个很大的加分项。

1.3

读写锁原子操作类文件存储设计零拷贝MMAP线程池ConcurrentHashMap写时复制容器负载均衡策略故障延迟机制堆外内存

RocketMQ核心组件

broker服务。

NameServer的作用是为消息生产者、消息消费者提供关于主题

Topic

的路由信息NameServer除了要存储路由的基础信息还要能够管理

Producer/Consumer

它能接收producer和consumer的请求并调用store层服务对消息进行处理。

HA服务的基本单元支持同步双写异步双写等模式。

Store

基于netty的底层通信实现所有服务间的交互都基于此模块。

也区分服务端和客户端

RocketMQ源码下载及安装

官方下载地址http://rocketmq.apache.org/dowloading/releases/

3.2

2下载且已经解压后的代码导入后执行Maven命令install

mvn

-Dmaven.test.skiptrue使用Maven验证下没问题

3.4

RocketMQ启动必须先启动NameServer启动类是namesrv/。

RocketMQ的运行主中创建conf、logs、store三个文件夹然后从源码下的中将broker.conf、logback_broker.xml、logback_namesrv.xml复制到conf下的NamesrvStartup类的main方法NameServer启动成功

3.4.2

在broker模块找到broker模块同时找到启动类BrokerStartup.java

3.5

运行前确保已经有jdk1.8Maven(打包需要安装Maven3.2.x)

3.5.2

老版本地址下载https://codeload.github.com/apache/rocketmq-externals/zip/master

新版本地址https://github.com/apache/rocketmq-dashboard

3.5.3

后端管理界面是一个Java工程独立部署同时也需要根据不同的环境进行相关的配置。

控制台端口及服务地址配置

下载完成之后进入‘\rocketmq-console\src\main\resources’文件夹打开‘application.properties’进行配置。

进入‘\rocketmq-externals\rocketmq-console’文件夹执行‘mvn

clean

-Dmaven.test.skiptrue’编译生成运行jar包。

编译成功之后cmd命令进入‘target’文件夹执行‘java

-jar

rocketmq-console-ng-2.0.0.jar’启动‘rocketmq-console-ng-2.0.0.jar’。

浏览器中输入‘127.0.0.1:8089’成功后即可进行管理端查看。

整体模块如下

的路由信息NameServer除了要存储路由的基础信息还要能够管理

2.rocketmq-broker

它能接收producer和consumer的请求并调用store层服务对消息进行处理。

HA服务的基本单元支持同步双写异步双写等模式。

3.rocketmq-store

消息过滤服务相当于在broker和consumer中间加入了一个filter代理。

8.rocketmq-srvutil

RocketMQ的源码是非常的多我们没有必要把RocketMQ所有的源码都读完所以我们把核心、重点的源码进行解读RocketMQ核心流程如下

启动流程

RocketMQ服务端由两部分组成NameServer和BrokerNameServer是服务的注册中心Broker会把自己的地址注册到NameServer生产者和消费者启动的时候会先从NameServer获取Broker的地址再去从Broker发送和接受消息。

消息生产流程

Producer将消息写入到RocketMQ集群中Broker中具体的Queue。

消息消费流程

Comsumer从RocketMQ集群中拉取对应的消息并进行消费确认。

NameServer源码分析

NameServer是整个RocketMQ的“大脑”它是RocketMQ的服务注册中心,所以RocketMQ需要先启动NameServer再启动Rocket中的Broker。

NameServer启动

启动监听等待Broker、Producer、Comsumer连接。

Broker在启动时向所有NameServer注册生产者在发送消息之前先从NameServer获取Broker服务器地址列表然后根据负载均衡算法从列表中选择一台服务器进行消息发送。

消费者在订阅某个主题的消息之前从NamerServer获取Broker服务器地址列表有可能是集群但是消费者选择从Broker中订阅消息订阅规则由

Broker

Broker启动后向所有NameServer发送路由及心跳信息。

路由剔除

移除心跳超时的Broker相关路由信息。

NameServer与每台Broker服务保持长连接并间隔10S检查Broker是否存活如果检测到Broker宕机则从路由注册表中将其移除。

这样就可以实现RocketMQ的高可用。

5.2

createNamesrvController()-start()

initialize()

核心解读NamesrvController类中createNamesrvController()

就可以打印这个NameServer的所有的参数信息不过NameServer会自动终止说明这个-p是一个测试参数。

5.2.2

Broker每隔30s向NameServer发送一个心跳包心跳包包含BrokerIdBroker地址Broker名称Broker所属集群名称、Broker关联的FilterServer列表。

但是如果Broker宕机NameServer无法收到心跳包此时NameServer如何来剔除这些失效的Broker呢NameServer会每隔10s扫描brokerLiveTable状态表如果BrokerLive的lastUpdateTimestamp的时间戳距当前时间超过120s则认为Broker失效移除该Broker关闭与Broker连接同时更新topicQueueTable、brokerAddrTable、brokerLiveTable、filterServerTable。

路由剔除机制中Borker每隔30S向NameServer发送一次心跳而NameServer是每隔10S扫描确定有没有不可用的主机120S没心跳那么问题就来了这种设计是存在问题的就是NameServer中认为可用的Broker实际上已经宕机了那么某一时间段从NameServer中读到的路由中包含了不可用的主机会导致消息的生产/消费异常不过不用担心在生产和消费端有故障规避策略及重试机制可以解决以上问题原理后续源码解读。

这个设计符合RocketMQ的设计理念整体设计追求简单与性能同时这样设计NameServer是可以做到无状态化的可以随意的部署多台其代码也非常简单非常轻量。

NameServer定期扫描brokerLiveTable检测上次心跳包与当前系统的时间差如果时间超过120s则需要移除broker。

Broker在正常关闭的情况下会执行unregisterBroker指令这两种方式路由删除的方法都是一样的都是从相关路由表中删除与该broker相关的信息。

在消费者启动之后第一步都要从NameServer中获取Topic相关信息

5.3

消息发送时客户端会从NameServer获取路由信息同时Broker会定时更新NameServer的路由信息所以路由表会有非常频繁的以下操作

RouteInfoManager类

因为Broker每隔30s向NameServer发送一个心跳包这个操作每次都会更新Broker的状态但同时生产者发送消息时也需要Broker的状态要进行频繁的读取操作。

所以这个地方就有一个矛盾Broker的状态会被经常性的更新同时也会被更加频繁的读取。

这里如何提高并发尤其是生产者进行消息发送时的并发所以这里使用了读写锁机制针对读多写少的场景。

synchronized和ReentrantLock基本都是排他锁排他锁在同一时刻只允许一个线程进行访问而读写锁在同一时刻可以允许多个读线程访问但是在写线程访问时所有的读线程和其他写线程均被阻塞。

读写锁维护了一对锁一个读锁和一个写锁通过分离读锁和写锁使得并发性相比一般的排他锁有了很大提升。

5.3.2

topicQueueTableTopic消息队列路由信息消息发送时根据路由表进行负载均衡

brokerAddrTableBroker基础信息包括brokerName、所属集群名称、主备Broker地址

clusterAddrTableBroker集群信息存储集群中所有Broker名称

brokerLiveTableBroker状态信息NameServer每次收到心跳包是会替换该信息

filterServerTableBroker上的FilterServer列表用于类模式消息过滤。

NameServer的实现基于内存NameServer并不会持久化路由信息持久化的重任是交给Broker来完成。

这样设计可以提高NameServer的处理能力。

5.3.3

NameServer集群中它们相互之间是不通讯主从架构中Broker都会向所有NameServer注册路由、心跳信息生产者/消费者同一时间与NameServer集群中其中一台建立长连接

项目实战部署分析

假设一个RocketMQ集群部署在两个机房每个机房都有一些NameServer、Broker和客户端节点当两个机房的链路中断时所有的NameServer都可以提供服务客户端只能在本机房的NameServer中找到本机房的Broker。

RocetMQ集群中NameSever之间是不需要互相通信的所以网络分区对NameSever本身的可用性是没有影响的如果NameSever检测到与Broker的连接中断了NameServer会认为这个Broker不再能提供服务NameServer会立即把这个Broker从路由信息中移除掉避免客户端连接到一个不可用的Broker上去。

网络分区后NameSever

收不到对端机房那些Broker的心跳这时候每个Namesever上都只有本机房的Broker信息。

Broker源码分析

5、亮点分析同步双写数倍性能提升的CompletableFuture

8、亮点分析堆外内存机制

在RocketMQ中Broker的处理是最多的所以我们先分析Broker的启动流程。

核心流程图如下

6.2

每一个分区一个物理文件夹在分区文件级别实现文件顺序写如果一个Kafka集群中拥有成百上千个主题每一个主题拥有上百个分区消息在高并发写入时其IO操作就会显得零散消息分散的落盘策略会导致磁盘IO竞争激烈成为瓶颈其操作相当于随机IO即

Kafka

RocketMQ在消息写入时追求极致的顺序写所有的消息不分主题一律顺序写入

commitlog

分区数量的增加而影响其顺序性。

在消息发送端消费端共存的场景下随着Topic数的增加Kafka吞吐量会急剧下降而RocketMQ则表现稳定。

因此Kafka适合Topic和消费端都比较少的业务场景而RocketMQ更适合多Topic多消费端的业务场景。

6.3

将所有主题的消息存储在同一文件确保消息发送时顺序写文件尽最大的能力确保消息发送的高性能与高吞吐量。

但由于一般的消息中间件是基于消息主题的订阅机制这样便给按照消息主题检索消息带来了极大的不便。

为了提高消息消费的效率

RocketMQ

消息队列文件每个消息主题包含多个消息消费队列每个消息队列有一个消息文件。

RocketMQ

索引文件其主要设计理念就是为了加速消息的检索性能可以根据消息的属性快速从

Commitlog

Commitlog文件存储的逻辑视图如下每条消息的前面4个字节存储该条消息的总长度。

但是一个消息的存储长度是不固定的。

ConsumeQueue

是消息的逻辑队列类似数据库的索引文件存储的是指向物理存储的地址。

每个Topic下的每个

Message

Consumequeue条目不会存储消息的全量信息消息条目如下

ConsumeQueue

由专门的线程产生消息转发任务从而构建消息消费队列文件ConsumeQueue

CommitLog

可以大大提高写入效率。

实际上磁盘有时候会比你想象的快很多有时候也比你想象的慢很多关键在如何使用使用得当磁盘的速度完全可以匹配上网络的数据传输速度。

目前的高性能磁盘顺序写速度可以达到600MB/s

但是磁盘随机写的速度只有大概100KB/s,和顺序写的性能相差

6000

存到内存中加速后续的读取速度。

同时因为ConsumeQueue中每一条消息的索引是固定长度所以也能够确保消息消费时的时间复杂度保持在O(1)。

3

ConsumeQueue

能够被全部读入内存所以这个中间结构的操作速度很快可以认为是内存读取的速度。

此外为了保证

CommitLog

存储着Topic和Consumer等相关信息。

主题和消费者群组相关的信息就存在在此。

topics.json

topic

集群消费模式消息消进度。

consumerFilter.json

6.3.2

RocketMQ使用Netty处理网络broker收到消息写入的请求就会进入SendMessageProcessor类中processRequest方法。

最终进入DefaultMessageStore类中asyncPutMessage方法进行消息的存储

然后消息进入commitlog类中的asyncPutMessage方法进行消息的存储

业务层也可以称之为网络层就是收到消息之后一般交给SendMessageProcessor来分配交给哪个业务来处理。

DefaultMessageStore这个是存储层最核心的入口。

6.5

RocketMQ的通讯使用的是Netty,作为客户端核心类有两种RemotingCommand与NettyRemotingClient。

RemotingCommand主要处理消息的组装包括消息头、消息序列化与反序列化。

NettyRemotingClient主要处理消息的发送包括同步、异步、单向、注册等操作。

因为RocketMQ消息种类比较众多所以对于消息的发送使用了一个类似于功能号的设计。

客户端发送消息时定义一个code对应一个功能服务端注册一个业务处理对应一个code的业务处理。

例如从生产者客户端代码跳入到NRC的代码NettyRemotingClient

MQClientAPIImpl类中的sendMessage()中

NettyRemotingClient类

而在NRS中只需要将服务端需要处理的ExecutorService注册到NRS组件中即可。

在启动流程中BrokerController类中的initialize()中

注意功能号的设计并对客户端和服务端不是一对一的在服务端往往处理是可以将不同的功能号对应到一个处理的任务中。

6.5.2

在RocketMQ4.7.0之后RocketMQ大量使用Java中的异步编程接口CompletableFuture。

尤其是在Broker端进行消息接收处理时。

比如DefaultMessageStore类中asyncPutMessage方法

Future接口正是设计模式中Future模式的一种实现如果一个请求或任务比较耗时可以将方法调用改为异步方法立即返回任务则使用主线程外的其他线程异步执行主线程继续执行。

当需要获取计算结果时再去获取数据。

节点与

节点之间数据同步/复制的方式有同步双写和异步复制两种模式。

同步双写是指Master将消息成功落盘后需要等待Slave节点复制成功(如果有多个Slave,成功复制一个就可以)后再告诉客户端消息发送成功。

RocketMQ

以后合理使用CompletableFuture对同步双写进行性能优化使得对消息的处理流式化大大提高了Broker的接收消息的处理能力。

6.5.3

RocketMQ在写入消息到CommitLog中时使用了锁机制即同一时刻只有一个线程可以写CommitLog文件。

CommitLog

这里注意lock锁的标准用法是try-finally处理防止死锁问题

RocketMQ

官方文档优化建议异步刷盘建议使用自旋锁同步刷盘建议使用重入锁调整Broker配置项useReentrantLockWhenPutMessage默认为false

同步刷盘时锁竞争激烈会有较多的线程处于等待阻塞等待锁的状态如果采用自旋锁会浪费很多的CPU时间所以“同步刷盘建议使用重入锁”。

异步刷盘是间隔一定的时间刷一次盘锁竞争不激烈不会存在大量阻塞等待锁的线程偶尔锁等待就自旋等待一下很短的时间不要进行上下文切换了所以采用自旋锁更合适。

6.5.4

RocketMQ底层对commitLog、consumeQueue之类的磁盘文件的读写操作都采用了mmap技术。

具体到代码里面就是利用JDK里面NIO的MapperByteBuffer的map()函数来先将磁盘文件CommitLog文件、consumeQueue文件映射到内存里来。

假如没有使用mmap技术的时候使用最传统和基本普通文件进行io操作会产生数据多拷贝问题。

比如从磁盘上把数据读取到内核IO缓冲区里面然后再从内核IO缓冲区中读取到用户进程私有空间里去然后我们才能拿到这个数据。

MMAP内存映射是在硬盘上文件的位置和应用程序缓冲区(application

buffers)进行映射建立一种一一对应关系由于mmap()将文件直接映射到用户空间所以实际文件读取时根据这个映射关系直接将文件从硬盘拷贝到用户空间只进行了一次数据拷贝不再有文件内容从硬盘拷贝到内核空间的一个缓冲区。

零拷贝(英语:

技术是指计算机执行操作时CPU不需要先将数据从某处内存复制到另一个特定区域。

这种技术通常用于通过网络传输文件时节省CPU周期和内存带宽。

mmap技术在地址映射的过程中对文件的大小是有限制的在1.5G2G之间所以RocketMQ就会把单个的commitLog文件大小控制在1GBconsumeQueue文件大小控制在5.72MB这样就在读写的时候方便的进行内存映射了。

Broker启动时MMAP相关源码如下

CommitLog类中的doAppend方法:具体进行消息格式的排列

6.5.5

一般情况下RocketMQ是通过MMAP内存映射生产时消息写入内存映射文件然后消费的时候再读。

但是RocketMQ还提供了一种机制。

堆外内存机制TransientStorePool,短暂的存储池(堆外内存)。

6.5.5.1

开启堆外内存需要修改配置文件brokertransientStorePoolEnabletrue

同时如果开启了堆外内存缓冲区的话集群模式必须是异步刷盘的模式同时该Broker必须为主节点通过查看源码我们可以可以看到这一限制

DefaultMessageStore.

从堆外内存的流程图也可以看出堆外内存的消息写入明显要多一个步骤所以堆外内存缓冲区的设置一定要求是异步才行。

6.5.5.2

RocketMQ单独创建一个ByteBuffer内存缓存池用来临时存储数据数据先写入该内存映射中然后由commit线程定时将数据从该内存复制到与目标物理文件对应的内存映射中。

RocketMQ引入该机制主要的原因是提供一种内存锁定将当前堆外内存一直锁定在内存中避免被进程将内存交换到磁盘。

同时因为是堆外内存这么设计可以避免频繁的GC。

6.5.5.3

在DefaultMessageStore类中开启堆外缓冲配置则进行堆外内存池初始化TransientStorePool

在创建MappedFile时如果检测到有堆外内存配置。

这里就会把堆外内存通过borrowBuffer()赋给writeBuffer

从6.4章节中的消息写入流程继续消息最终会进入mappedFile中处理。

经过几个转跳后进入appendMessagesInner方法中

以上就可知如果有堆外内存缓冲区的话ByteBuffer的来源是不同的。

不过这里要注意如果是堆外内存缓冲区消息写入到ByteBuffer的话还只是写入一个临时区域不像默认模式本身就是mmap映射的内存直接写入就进入了磁盘和内存的映射所以这里还需要一个步骤就是从临时区域到正式区域。

RocketMQ这里使用的是定时任务处理并且是借用异步刷盘的定时任务来处理。

这里需要回到CommitLog的构造方法在CommitLog构造的时候会选择启动一个定时任务来处理堆外内存

上述跳转比较麻烦但是记住一个点就是堆外内存的数据写入本质上是分成两个阶段

6.5.5.4

从图中可以发现默认方式MmapPageCache的方式读写消息都走的是pageCache(MappedByteBuffer类)这样子读写都在pagecache里面不可避免会有锁的问题在并发的读写操作情况下会出现缺页中断降低内存加锁污染页的回写脏页面。

而如果采用堆外缓冲区DirectByteBuffer(堆外内存)PageCache的两层架构方式这样子可以实现读写消息分离写入消息时候写到的是DirectByteBuffer——堆外内存中,读消息走的是PageCache(MappedByteBuffer类)带来的好处就是避免了内存操作的很多容易堵的地方降低了时延比如说缺页中断降低内存加锁污染页的回写。

所以使用堆外缓冲区的方式相对来说会比较好但是肯定的是需要消耗一定的内存如果服务器内存吃紧就不推荐这种模式同时的话堆外缓冲区的话也需要配合异步刷盘才能使用(因为写数据分成了两步同步刷盘延迟就会比较大)。

Producer源码分析

整个JVM中只存在一个MQClientManager实例维护一个MQClientInstance缓存表

一个clientId只会创建一个MQClientInstance

clientId生成规则IPinstanceNameunitName

ClientConfig类

每一个客户端就是一个MQClientInstance每一个ClientConfig对应一个实例。

故不同的生产者、消费端如果引用同一个客户端配置(ClientConfig)则它们共享一个MQClientInstance实例。

所以我们在定义的的时候要注意这种问题生产者和消费者如果分组名相同容易导致这个问题

7.2.3

MQClientInstance类startScheduledTask()

7.3

我们从一个生产者案例的代码进入代码可知DefaultMQProducerImpl中的sendDefaultImpl()是生产者消息发送的核心方法

从核心方法可知消息发送就是4个步骤验证消息、查找路由、选择队列、消息发送。

7.4

采用了最简单的轮询算法这种算法有个很好的特性就是保证每一个Queue队列的消息投递数量尽可能均匀。

这种算法只要消息投递过程中没有发生重试的话基本上可以保证每一个Queue队列的消息投递数量尽可能均匀。

当然如果投递中发生问题比如第一次投递就失败那么很大的可能性是集群状态下的一台Broker挂了所以在重试发送中进行规避。

这样设置也是比较合理的。

7.4.2

采用此策略后每次向Broker成功或者异常的发送RocketMQ都会计算出该Borker的可用时间发送结束时间-发送开始时间失败的按照30S计算并且保存方便下次发送时做筛选。

除了记录Broker的发送消息时长之外还要计算一个Broker的不可用时长。

这里采用一个经验值

在开启故障延迟机制策略步骤如下

从这种策略上可以很明显看到默认队列选择是轮训策略而故障延迟选择队列则是优先考虑消息的发送时长短的队列。

那么如何选择呢

首先RocketMQ默认的发送失败有重试策略默认是2也就是如果向不同的Broker发送三次都失败了那么这条消息的发送就失败了作为RocketMQ肯定是尽力要确保消息发送成功。

所以给出以下建议。

如果是网络比较好的环境推荐默认策略毕竟网络问题导致的发送失败几率比较小。

如果是网络不太好的环境推荐故障延迟机制消息队列选择时会在一段时间内过滤掉RocketMQ认为不可用的broker以此来避免不断向宕机的broker发送消息从而实现消息发送高可用。

当然以上成立的条件是一个Topic创建在2个Broker以上的的基础上。

7.4.4

DefaultMQProducerImpl类中sendKernelImpl方法

根据源码分析客户端(MQClientInstance)中连接的建立时机为按需创建也就是在需要与对端进行数据交互时才建立的。

建立的是长连接。

Consumer源码分析

Commitlog中每一条消息放入Commitlog的时候都需要上锁确保顺序的写入。

当Commitlog写成功了之后。

数据通过ReputMessageService类定时同步到ConsunmeQueue中写入Consume

8个、size

查找消息的时候可以直按根据队列的消息序号计算出索引的全局位置比如序号2就知道偏移量是20然后直接读取这条索引再根据索引中记录的消息的全局位置找到消息。

这两次查找是差不多的第一次在通过序号在consumer

Queue中获取数据的时间复杂度是O(1)第二次查找commitlog文件的时间复杂度也是O(1)所以消费时查找数据的时间复杂度也是O(1)。

8.1.1

ReputMessageService.doReput源码分析

DefaultMessageStore.

ReputMessageService线程每执行一次任务推送休息1毫秒就继续尝试推送消息

1返回reputFromOffset偏移量开始的全部有效数据(commitlog文件)。

然后循环读取每一条消息。

2从result返回的ByteBuffer中循环读取消息一次读取一条创建DispatchRequest对象。

如果消息长度大于0则调用doDispatch方法。

最终将分别调用CommitLogDispatcherBuildConsumeQueue(构建消息消费队列)CommitLogDispatcherBuildIndex(构建索引文件)。

3构建消息消费队列

Group中的各个Consumer实例分摊去消费消息即一条消息只会投递到一个Consumer

Queue的去做拉取消费。

例如某个Topic有3条Q其中一个Consumer

Group

而由Producer发送消息的时候是轮询所有的Q,所以消息会平均散落在不同的Q上可以认为Q上的消息是平均的。

那么实例也就平均地消费消息了。

这种模式下消费进度

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-ARadJII4-1677492460948)(file:///C:\Users\ADMINI~1\AppData\Local\Temp\msohtmlclip1\01\clip_image004.jpg)]

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-fEoQyoof-1677492460949)(file:///C:\Users\ADMINI~1\AppData\Local\Temp\msohtmlclip1\01\clip_image006.jpg)]

8.3.2

实际上是一个消费组下的每个消费者实例都获取到了topic下面的每个Message

这种模式下消费进度

在集群消费模式下每条消息只需要投递到订阅这个topic的Consumer

Group下的一个实例即可。

RocketMQ采用主动拉取的方式拉取并消费消息在拉取的时候需要明确指定拉取哪一条message

而每当实例的数量有变更都会触发一次所有实例的负载均衡这时候会按照queue的数量和实例的数量平均分配queue给每个实例。

默认的分配算法是AllocateMessageQueueAveragely

还有另外一种平均的算法是AllocateMessageQueueAveragelyByCircle也是平均分摊每一条queue只是以环状轮流分queue的形式

如下图

需要注意的是集群模式下queue都是只允许分配只一个实例这是由于如果多个实例同时消费一个queue的消息由于拉取哪些消息是consumer主动控制的那样会导致同一个消息在不同的实例下被消费多次所以算法上都是一个queue只分给一个consumer实例一个consumer实例可以允许同时分到不同的queue。

通过增加consumer实例去分摊queue的消费可以起到水平扩展的消费能力的作用。

而有实例下线的时候会重新触发负载均衡这时候原来分配到的queue将分配到其他实例上继续消费。

queue的总数量还多的话多出来的consumer实例将无法分到queue也就无法消费到消息也就无法起到分摊负载的作用了。

所以需要控制让queue的总数量大于等于consumer的数量。

8.4.2

由于广播模式下要求一条消息需要投递到一个消费组下面所有的消费者实例所以也就没有消息被分摊消费的说法。

在实现上其中一个不同就是在consumer分配queue的时候所有consumer都分到所有的queue。

8.5

一般我们在消费时使用回调函数的方式使用得最多的是并发消费消费者客户端代码如下

参考RocketMQ核心流程

在消费者启动之后第一步都要从NameServer中获取Topic相关信息。

这一步设计到组件之间的交互RocketMQ使用功能号来设计的。

GET_ROUTEINFO_BY_TOPIC在idea上使用ctrlH

MQClientInstance类中

在消费消息前需要获取当前分组已经消费的相关信息ConsumerList

进入RebalanceImpl类

再进入具体的类,如果是广播消费模式则不需要从服务器获取消费进度广播消费模式把进度在本地消费端进行存储

8.5.3

在分配完消费者对应的Queue之后如果是集群模式的话需要获取这个消费者对应Queue的消费Offset,便于后续拉取未消费完的消息。

RebalanceImpl类中rebalanceByTopic方法

进入RebalancePushImpl类

最终进入DefaultMQPushConsumerImpl类的pullMessage方法

8.5.5

这里要注意因为RocketMQ的推模式是基于拉模式实现的因为拉消息是一批批拉所以不能做到拉一批提交一次偏移量所以这里使用定时任务进行偏移量的更新。

8.5.6

顺序消费代码顺序消费的流程和并发消费流程整体差不多唯一的多的就是使用锁机制来确保一个队列同时只能被一个消费者消费从而确保消费的顺序性

这里有一个定时任务是每个20秒运行一次周期性的去续锁锁的有效期是60S

8.7

之前在消费的流程中尤其是针对顺序消息感觉上会有卡死的现象由于顺序消息中需要到Broker中加锁如果消费者某一个挂了那么在Broker层是维护了60s的时间才能释放锁所以在这段时间只能消费者是消费不了的在等待锁。

另外如果还有Broker层面也挂了如果是主从机构获取锁都是走的Master节点如果Master节点挂了走Slave消费但是slave节点上没有锁所以顺序消息如果发生了这样的情况也是会有卡死的现象。

8.8

在并发消费的时候当我们启动了非常多的消费者维护了非常多的topic的时候、或者queue比较多的时候你可以看到消费的流程的交互是比较多的5~6步要启动多线程也要做相当多的事情所以你会感觉要启动较长的时间才能消费。

还有顺序消费的时候如果是之前的消费者挂了这个锁要60秒才会释放也会导致下一个消费者启动的时候需要等60s才能消费。

8.9

如果在同一个数据库中进行事务可以保证这两步操作要么同时成功要么同时不成功。

这样就保证了转账的数据一致性。

但是在微服务架构中因为各个服务都是独立的模块都是远程调用都没法在同一个事务中都会遇到分布式事务问题。

9.2

RocketMQ采用两阶段提交把扣款业务和加钱业务异步化在A系统扣款成功后发送“扣款成功消息”到消息中间件B系统中加钱业务订阅“扣款成功消息”再对用户进行加钱。

9.2.1

先扣款再发送消息万一发送消息超时了(MQ中有可能成功有可能失败)那这个状态就很难判断了

2、先向RocketMQ发消息后再扣款

扣款成功消息发送成功但是如果本地扣款业务失败了那消息已经发给MQ了第二阶段的加钱就会执行成功。

其实仔细分析下问题的关键点就是RocketMQ改变不了消息发送者的事务状态。

所以RocketMQ的分布式事务方案进行了优化。

9.3

所以RocketMQ在分布式事务中引入了半事务及事务回查机制。

半事务

发一个消息到rocketmq但该消息只储存在commitlog中但consumeQueue中不可见也就是消费端订阅端无法看到此消息。

事务回查

RocketMq会定时遍历commitlog中的半事务消息这个事务回查机制就可以站在

9.4

通过事务回查如果在TransactionListenerImpl类executeLocalTransaction方法中如果本地事务执行成功则提交commit_message,消费端即可消费消息

如果有一些比较耗时的操作导致不能在这个步骤确认的话可以提交UNKNOW交给定时的任务回查来处理

另外一点如果担心生产者发生故障导致分布式事务的问题话定时事务回查是可以在生产者群组中做的。

我们可以做一个这样的案例一个生产者1一个生产者2消费发送时的分组是一样都使用分布式事务消息。

生产者1发生故障了消息状态还是一个UNKNOW状态只要生产者2还存活生产者2就可以帮助生产者1完成事务回查的确认从而不会有单点故障问题。

9.5

RocketMQ使用Netty处理网络broker收到消息写入的请求就会进入SendMessageProcessor类中processRequest方法。

最终进入DefaultMessageStore类中asyncPutMessage方法进行消息的存储

结合图同时结合代码我们可以看到在事务消息发送时消息实际存储的主题是一个系统主题RMQ_SYS_TRANS_HALF_TOPIC

9.5.2

DefaultMQProducerImpl类sendMessageInTransaction方法

Broker

DefaultMQProducerImpl类checkTransactionState方法

DefaultMQProducerImpl类processTransactionState方法

Broker

在Broker启动的时候是要作为客户端定期的访问客户端做事务回查。

回顾到之前讲到《6.1

事务回查是Broker发起的一次定时的网络调用每隔60s所以事务回查在客户端启动的时候第一次不一定是60s的间隔一般会小于60s因为事务回查是broker发起的并不是client端定时发起