RocketMQ源码分析_草稿图

2024-03-04 21:46:53   3  举报





这是一份草稿图，展示了RocketMQ的源码分析。图中详细标注了RocketMQ的主要组件和模块，包括NameServer、Broker、Producer、Consumer等。每个组件都用不同的颜色进行了区分，以便于理解和记忆。此外，图中还标注了一些关键的代码逻辑和数据流，如消息的发送和接收过程，以及各种消息过滤和处理策略。这张草稿图是学习RocketMQ源码的重要参考，可以帮助开发者快速理解其工作原理和设计思想。

架构图

作者其他创作

大纲/内容

保存消费位点

Java进程

启动通信模块服务

对比已同步位点

CommitLog

Yes:返回Pull结果

带Tag的消息

异步转存服务

执行本地事务

初始化位点管理器，并且加载消费位点

start()

initialize()

本地事务是否成功

索引数据

按照Offset Pull消息

Push消费启动流程

返回生产者存储成功

启动rebalance服务

Tag过滤流程

1.保存前执行Hook2.解析请求参数3.死信重试消息处理4.保存后执行Hook5.返回结果

启动消费者实例

同步刷盘

handleHA()

消费消息

消费者

#5.4

业务消费代码

启动消费者实现类

启动Pull服务

通知刷盘请求处理结果

发送消息

等待offset同步完成锁

映射物文件

文件内容按Page保存

consumeMessage()

Half消息存储(此时不会被消费)

BrokerStartup启动

置为服务启动失败状态

拷贝订阅关系

查找CommitLog中的消息体内容

CommitLog异步刷盘成功后

启动

初始化Checkpoint文件

加载全部ConsumeQueue文件

Span=(N+K)-N

ClassRebalancePushImpl

DefualtMQProducer

....

将生产者状态置为启动失败

1.同步Topic配置2.同步消费位点信息3.同步延迟位点信息4.同步消费订阅信息

Master Broker

00000000236222320128

MappedByteBuffer

config

SendMessageProcessor.java

0000000000000

Yes

初始化消息拉取服务

Broker异常退出检查

执行全部刷盘

拉取数目是否满足

文件内容

./bin/mqshutdownRocketMQ环境设置

NamesrvStartup.java

返回消息

发送消息请求

拉取消息

.....

Bloom过滤器是否通过

6789

启动各种存储服务

./bin/mqnamesrv

doCommit()

4字节

HAService

createTempFIle()

第1个子文件

MappedFileQueue

下一轮询

send()

commitlog

RMQ_SYS_TRANS_OP_HALF_TOPIC

查询消息

进程正常关闭时

Push消费流程

Pull线程

异步刷盘

20200420204115925

Consume Queue添加索引成功

用户态内存

服务是否停止

零拷贝读写文件流程

保存刷盘请求

Tag的Hash值

取消之前发送的Half消息（此时不会被消费）

abort文件创建流程

CommitLog复制流程

操作系统Page Cache

20200421055715499

是否有新消息

轮询全部队列Pull消息

shutdown()

doReput()

物理CommitLog文件2

Direct Memory-Page Cache读写分离

Yes：返回Pull结果

DefualtMQPullConsumer

SQL过滤是否通过

结束

执行刷盘

预读

DefaultMQProducer

文件映射-写

interfaceMQConsumer

ClassRemoteBrokerOffsetStore

一个消费者实例下线后Rebalance过程

发送Half消息

是否有推荐的Broker id

extends

返回全部消息

ClassRebalamcePullImpl

MappedFile实例2

RocketMQ Broker

生产者启动流程

MQClientAPIImpl

磁盘

Broker关闭流程

配置数据同步流程

Broker

线程等待刷盘请求处理结果通知

拉取前进行check

Consume Queue刷盘成功

N+1

N+2

N+K

20200421140640517

unregister_client

构造函数初始化实例

发送刷盘请求

同步完成，等待被唤醒

消息返回给消费者消费

编译SQL

调用

./bin/runbroker.sh

Pull

主要的发送逻辑:1.发送前check2.选择Queue发送3.可靠发送4.发送结果处理

拉取消息并进行消费

DefualtMQPushConsumerImpl

返回客户端消息存储结果

临时存储

文件映射-读

processRequest()

执行转存

putMessage()

BrokerController.java

1.恢复TopicQueue关系数据2.纠正ConsumeQueue最小消息位点信息

消息对象

GroupCommitService

DefaultMQProducerImpl

pullMessage()

BrokerStartup.java

appendMessage()

interfaceMQAdmin

更新logincsMsgTimestamp(最后一条写Consume Queue消息的存储时间)

CommitLog查询消息

保存消息体

#5.1

Commit/Rollback

启动生产者实例

注册生产者

notifyTransferSome()

DefaultMessageStore

MQClientInstance

拉取配置信息

Master BrokerOuterAPI

启动消息拉取服务

初始化Pull消费者

invokeOneWay

生产者

持久化消费位点

ClassLocalFileOffsetStore

默认生产者组名转换为进程id

Hash槽位

存储成功

lock

临时存储消息

yes

pull请求

8字节

删除aobrt文件

置为服务启动状态

Master Slave读写分离

启动RebalanceService服务

设置JVM参数

关闭各种存储服务、刷盘服务

call

Header

启动BrokerController

业务消费消息

关闭各个服务1.启动Broker监控服务2.启动Client探活服务3.关闭长轮询服务4.关闭远程通信服务Server5.关闭VIP通信服务Server6.关闭messageStore服务7.关闭定时任务8.取消注册Broker-to-Namesrv9.关闭消息发送处理线程池10.关闭Pull请求处理线程池11.关闭Admin管理请求处理线程池12.关闭brokerOuterAPI服务13.持久化消费位点14.关闭Filter Server服务15.关闭快速失败服务16.持久化消费过滤进度信息

加载延迟消息的位点信息

subscriptionGroup.json

Push消费者拉取消息过程

记录异常关闭日志

恢复全部数据

线程池

checkpoint

RemotingClient

启动失败

循环

synchornizedswapRequests()

classDefaultPullConsumer

DefaultMessageStore.java

Pull 消息

无消息

abort

处理同步刷盘请求

返回拉取结果

异常恢复CommitLogrecoverAbnormally()

启动脚本

DefualtMQProducer构造函数初始化实例

调用发送方法

启动Push服务

发送失败

查找Consume Queue中位点对应的PageCache

第2个子文件

加载全部ConsumeQueue数据

更新indexMsgTimestamp(被刷盘的indexFile的最后更新时间)

带SQL的订阅信息

...

Filter Server

回查服务

NamesrvController.java

设置同步发送

全部Queue

topics.json

Pull Consumer

YES

同步刷盘过程

20200420134806913

Slave BrokerController

#5.3

开始

更新当前Slave已同步位点

消息成功存储到映射文件

等待转存时间间隔

创建abort文件

消息物理位点

设置超时时间

物理CommitLog文件N

更新physicMsgTimestamp(最后一条刷盘的CommitLog存储时间)

CommitLog同步刷盘成功后

sendKernelImpl()

初始化Pull接口包装类

查询Topic全部Queue

执行全部转存

启动各种定时任务

1.将消费者状态置为启动失败2.消费者参数校验3.默认消费者组名转换为进程id

查找Consume Queue

处理关闭消费者实例

main0()

记录Checkpoint信息

同步刷盘服务线程

提交消费位点

向Pull接口包装类注册消息过滤器

消息不返回消费者

HAConnection

MessageListener

恢复ConsumeQueue

执行一次rebalance

KV配置加载、初始化通信层，初始化各个定时任务

ClassGroupCommitService

interfaceRunnable

Slave Broker

doAppend()

本地注册消费者

业务层

DefualtMQProducerImpl

B、C、D消费者实例

生产者参数校验

notify_consumer_ids_changed

delayOffset.json

线程

MappedFile实例M

加载IndexFile索引

RocketMQ Client

BrokerController

映射物理文件

直接内存(DM)

存储消息线程

通知复制

Broker保存消息流程

JVM Shutdown Hook

启动命令

consumeOffset.json

获取Topic全部队列信息

Java堆内存

链表

syncall()

Abstract ClassRebalanceImpl

链表下一个节点指针

处理同步复制

通过createBrokerController()方法解析命令行参数，初始化全部配置:1.初始化Remoting配置2.初始化Broker配置3.初始化Broker存储配置4.通过JVM Hook关闭进程

启动Client实例

执行过滤方法

全部CommitLog是否已创建索引

命令行参数处理启动配置文件加载

设置消息对消费者可见

appendMessagesInner()

拉取并消费消息

registerProducer

CommitLog.java

是否成功

异步实时刷盘

异步定时刷盘

数据一致性

中

低

高

数据可靠性

数据可用性

系统吞吐量

唤醒异步刷盘线程

时间差

忽略该消息

准备消息对象

事务消息发送过程

A消费实例

拉取消息、Master高负载判断

1.保存Topic配置2.保存消费位点信息3.保存延迟位点信息4.保存消费订阅关系

load()

Broker启动流程

Direct Memory

GroupTransferService

MappedFile实例1

BrokerStartup JVM关闭hook

waitForRunning(10)

是否发送成功

ConsumeMessageService

./bin/runserver.sh

关闭命令

FlushRealTimeService

初始化Push消费者

MappedFile实例N

Topic

No:Rollback消息

Index File刷盘成功

1234

Yes:Commit消息

implements

shutdown

内核态内存

consumequeue

存储消息

./bin/mqbroker

index

waitForFlush()等待同步Slave结果

发送网络请求，发送消息

interfaceMQPullConsumer

是否异常退出

正常恢复CommitLogrecoverNormally()

文件头

物理CommitLog文件M

无Hash碰撞

ClassServiceThread

Namesrv启动流程

copy

消息发送流程

消费方式/对比项

Push

备注

是否需要主动拉取

理解分区后，需要主动拉取各个分区消息

自动

Pull消息灵活；Push使用更简单

位点管理

用户自行管理或者主动提交给Broker管理

Broker管理

Pull自主管理位点，消费灵活；Push位点交由Broker管理

Topic路由变更是否影响消费

否

Pull模式需要编码实现路由感知，Push模式自动执行Rebalance，以适应路由变更

Pull消费启动流程

是否初始化启动

Index Data

保存消费进度

main()

注册本地路由信息

PageCache

putRequest()

等待刷盘时间间隔

执行方法start()

物理CommitLog文件1

sendDefaultImpl()

CommitRealTimeService

SlaveACK的数据位点是否大于Master上标记的位点

执行方法start()1.启动messageStore服务2.启动远程通信服务Server3.启动VIP通信服务Server4.启动BrokerQuterAPI服务5.启动长轮询服务6.启动Client探活服务7.启动FilterServer服务8.注册Broker-to-Namesrv9.启动自动注册Broker-to-Namesrv定时任务10.启动Broker监控服务11.启动快速失败服务

抛出程序异常

发送request封装

消费者拉取消息

interfaceOffsetStore

Pull请求

执行方法main()

获取刷盘参数

#5.2

ClassCOmmitRealTimeService

消息Key的Hash值

在Consume Queue中找到下一条消息物理位点、消息大小、Tag的Hash值

CREATE_JUST