RocketMQ技术讲解V2.0_rocketmq消息模式

RocketMQ

原理解析

需积分: 15 104 浏览量更新于2023-03-16 评论 1 收藏 3.02MB PDF 举报

身份认证购VIP最低享 7 折!

领优惠券(最高得80元）

资源详情

资源评论

资源推荐

 
 
RocketMQ 技术讲解 
   
目录 
ROCKETMQ 技术讲解 ......................................................................................................................... 1 
存储篇.............................................................................................................................................. 6 
1  整体结构 ................................................. 6 
2  大文件的磁盘操作——MapedFile ............................ 6 
2.1  向文件顺序写操作（appendMessage） ................. 7 
2.2  消息刷盘操作（commit） ............................ 7 
2.3  随机读操作（selectMapedBuffer） ................... 7 
2.4  清理内存操作 ...................................... 8 
2.5  判断文件是否写满 .................................. 8 
3  MapedFileQueue ........................................... 8 
3.1  获取在某时间点之后更新的文件（getMapedFileByTime） 8 
3.2  清理指定偏移量所在文件之后的文件（truncateDirtyFiles）

3.3  获取或创建最后一个文件（getLastMapedFile） ........ 9 
3.4  获取列表中的最后一个文件（getLastMapedFile2） ..... 9 
3.5  统计内存的数据还有多少未持久化（howMuchFallBehind）

3.6  获取 MapedFile 队列中最小 Offset 值（getMinOffset） 10 
3.7  获取 MapedFile 队列中最大 Offset 值（getMaxOffset） 10 
3.8  删除某类文件中的最后一个文件（deleteLastMaped） .. 10 
3.9  根据指定的 offset 找到所在文件（findMapedFileByOffset）

3.10  MapedFile 队列中的消息刷盘（commit）.............. 10 
4  Commitlog ............................................... 10 
4.1  文件的消息单元存储结构 ........................... 11 
4.2  CommitLog 类结构 ..................................... 12 
4.3  获取最小 Offset（getMinOffset）................... 12 
4.4  获取最大物理偏移量（getMaxOffset） ............... 13 
4.5  读取指定起始位置 offset 所在文件的全部剩余消息
（getData）................................................ 13 
4.6  正常恢复 CommitLog 内存数据（recoverNormally） .... 13 
4.7  异常恢复 CommitLog 内存数据（recoverAbnormally） .. 13 
4.8  写入消息（putMessage） ........................... 15 
4.9  读取消息（getMessage） ........................... 16 
4.10  指定位置开始写入二进制消息（appendData） ......... 17 

 
 
4.11  获取指定位置所在文件的下一个文件的起始偏移量
（rollNextFile）........................................... 17 
4.12  DefaultAppendMessageCallback 类的实现............. 17 
5  Consumequeue ............................................ 18 
5.1  ConsumeQueue 类结构 .................................. 19 
5.2  删除指定偏移量之后的逻辑文件
（truncateDirtyLogicFiles）................................ 19 
5.3  恢复 ConsumeQueue 内存数据（recover） ............. 20 
5.4  查找消息发送时间最接近 timestamp 逻辑队列的 offset
（getOffsetInQueueByTime）................................. 20 
5.5  获取最后一条消息对应物理队列的下一个偏移量
（getLastOffset）.......................................... 21 
5.6  消息刷盘（commit） ............................... 21 
5.7  将 commitlog 物理偏移量/消息大小等信息直接写入
consumequeue 中（putMessagePostionInfoWrapper） ............ 21 
5.8  根据消息序号索引获取 consumequeue 数据（getIndexBuffer）

5.9  根据物理队列最小 offset 计算修正逻辑队列最小 offset
（correctMinOffset）....................................... 21 
5.10  获取指定位置所在文件的下一个文件的起始偏移量
（rollNextFile）........................................... 22 
6  IndexFile ............................................... 22 
6.1  Index 文件的数据结构 ................................. 23 
6.2  向 index 文件中写入索引消息（putKey） ............. 24 
6.3  以 topic-key 值从 Index 中获取在一个时间区间内的物理偏
移量列表（selectPhyOffset）................................ 24 
7  IndexService ............................................ 25 
7.1  创建消息的索引（buildIndex） ..................... 25 
7.2  查找 topic 和 key 的物理偏移量 offset（queryOffest） 25 
8  Config .................................................. 26 
8.1  ScheduleMessageService 执行延迟消息 .................. 26 
8.2  收到发送消息时创建 topic 的配置信息
（createTopicInSendMessageMethod）......................... 28 
8.3  根据 GroupName 查找订阅组信息
（findSubscriptionGroupConfig）............................ 28 
8.4  收到消费失败时的回传消息时创建 topic 的配置信息
（createTopicInSendMessageBackMethod）..................... 28 
9  DefaultMessageStore—所有文件的访问入口 ................. 29 
9.1  根据 topic 和 queueId 查找 ConsumeQueue
（findConsumeQueue）....................................... 29 
9.2  根据物理偏移量和数据大小获取消息内容
（lookMessageByOffset）.................................... 29 
9.3  将消息写入 commitlog 中（putMessage） ............. 29 
9.4  读取 commitlog 消息（getMessage） ................. 30 

 
 
9.5  获取最大物理偏移量（getMaxPhyOffset） ............ 32 
9.6  获取指定偏移量之后的所有 commitlog 数据
（getCommitLogData）....................................... 32 
9.7  从指定位置开始追加 commitlog 数据（appendToCommitLog）

9.8  DefaultMessageStore.ReputMessageService 服务线程 ..... 32 
9.9  DefaultMessageStore.DispatchMessageService 服务线程 .. 33 
9.10  加载 ConsumeQueue 队列数据（loadConsumeQueue） .... 34 
9.11  获取指定队列的最大逻辑 Offset（getMaxOffsetInQuque）

9.12  获取指定队列的最小逻辑 Offset（getMinOffsetInQuque）

10  Abort ................................................... 35 
11  Checkpoint .............................................. 35 
12  HA 高可用................................................ 35 
12.1  Topic 配置同步.................................... 36 
12.2  消费进度信息同步 ................................. 36 
12.3  延迟消费进度信息同步 ............................. 36 
12.4  订阅关系同步 ..................................... 37 
12.5  消息数据同步 ..................................... 37 
13  事务消息相关的文件 ...................................... 41 
13.1  事务消息状态文件 ................................. 42 
13.2  事务消息 REDO 日志文件 ............................ 42 
13.3  定期向 Producer 回查事务消息的最新状态 ............ 42 
NAME SERVER 篇 ..................................................................................................................... 43 
1  NameServer 的功能........................................ 43 
2  NameServer 的初始化及启动过程............................ 43 
3  处理 Broker 注册请求 ..................................... 45 
4  根据 Topic 获取 Broker 信息和 topic 配置信息
（getRouteInfoByTopic）........................................ 46 
BROKER 篇 ................................................................................................................................. 47 
1  Broker 的初始化过程...................................... 47 
2  Broker 的启动过程........................................ 50 
3  向 NameServer 注册 Broker................................. 52 
4  清理未使用的 topic 数据 .................................. 52 
5  根据 topic 和 group 查找 Consumer 订阅信息（findSubscriptionData）

6  处理 Producer 发来的消息 ................................. 53 
7  处理与客户端的心跳消息 .................................. 54 
8  注册 Consumer 信息 ....................................... 55 
9  注册 Producer 信息 ....................................... 56 
10  查询消费进度（QUERY_CONSUMER_OFFSET） ................... 56 
11  更新消费进度（UPDATE_CONSUMER_OFFSET） .................. 57 

 
 
12  处理 Consumer 拉取消息 ................................... 57 
13  对未拉取到的消息进行重试（PullRequestHoldService） ...... 60 
14  客户端顺序消费时锁住 MessageQueue 队列的请求（LOCK_BATCH_MQ）

15  客户端顺序消费时解锁 MessageQueue 队列的请求（UNLOCK_BATCH_MQ）

16  获取 consumerGroup 名下的所有 Consumer 的 ClientId
（GET_CONSUMER_LIST_BY_GROUP）................................. 62 
17  将 Consumer 消费失败的消息写入延迟消息队列中
（CONSUMER_SEND_MSG_BACK）..................................... 62 
18  处理结束事务消息的请求（END_TRANSACTION） ............... 64 
19  客户端发起更新或创建 Topic（UPDATE_AND_CREATE_TOPIC）.... 64 
PRODUCER 篇 ............................................................................................................................ 65 
1  启动 Producer............................................ 65 
2  启动 MQClientInstance 类（MQClientInstance.start） ....... 66 
3  从 NameServer 更新 Topic 的本地路由信息 ................... 68 
4  向 Broker 发送心跳消息 ................................... 71 
5  发送普通消息 ............................................ 71 
6  发送定时消息 ............................................ 74 
7  发送顺序消息 ............................................ 74 
8  发送事务消息 ............................................ 75 
9  处理 Broker 检查事务状态的消息（CHECK_TRANSACTION_STATE） 77 
10  创建 Topic............................................... 78 
CONSUMER 篇 .............................................................................................................................. 78 
1  启动 Consumer............................................ 78 
2  向 Broker 同步消费进度的定时任务 ......................... 83 
3  负载均衡服务线程（RebalanceService） .................... 84 
3.1  为 topic 下的所有 MessageQueue 创建拉取消息请求（广播
模式下）  85 
3.2  为 Topic 所分配每个 MessageQueue 创建拉取消息请求（集
群模式下）................................................. 85 
3.3  为每个消息队列维护一个消息处理队列
（updateProcessQueueTableInRebalance）..................... 87 
3.4  删除未使用的消息队列的消费进度
（removeUnnecessaryMessageQueue）.......................... 88 
3.5  获取 MessageQueue 队列的下一个消费偏移量
（computePullFromWhere）................................... 89 
4  拉取消息服务线程（PullMessageService） .................. 91 
4.1  拉取消息的处理逻辑 ............................... 91 
4.2  延迟拉取消息 ..................................... 91 
4.3  立即拉取消息 ..................................... 92 
5  PUSH 模式下的消息消费（DefaultMQPushConsumer）........... 92 
5.1  拉取消息（pullMessage） .......................... 92 