RocketMQ源码分析讲解_rocketmq源码资源-CSDN文库

5星 · 超过95%的资源需积分: 44 58 浏览量 2017-08-04 11:41:11 上传评论 2 收藏 1.27MB PDF 举报

资源推荐

资源详情

资源评论

RocketMQ 技术讲解 
   Csdn 博客: http://blog.csdn.net/meilong_whpu 
 
目录 
ROCKETMQ 技术讲解 ......................................................................................................................... 1 
存储篇.............................................................................................................................................. 6 
1  整体结构 ................................................... 6 
2  大文件的磁盘操作——MapedFile .............................. 6 
2.1 向文件顺序写操作（appendMessage）..................... 7 
2.2 消息刷盘操作（commit）................................ 7 
2.3 随机读操作（selectMapedBuffer）....................... 8 
2.4 清理内存操作.......................................... 8 
2.5 判断文件是否写满...................................... 8 
3  MapedFileQueue——为操作 commitlog/consumequeue 文件提供访问
服务 8 
3.1 获取在某时间点之后更新的文件（getMapedFileByTime）.... 9 
3.2 清理指定偏移量所在文件之后的文件（truncateDirtyFiles） 9 
3.3 获取或创建最后一个文件（getLastMapedFile）............ 9 
3.4 获取列表中的最后一个文件（getLastMapedFile2）........ 10 
3.5 统计内存的数据还有多少未刷到磁盘文件中
（howMuchFallBehind）...................................... 10 
3.6 获取 MapedFile 队列中最小 Offset 值（getMinOffset）.... 10 
3.7 获取 MapedFile 队列中最大 Offset 值（getMaxOffset）.... 10 
3.8 删除某类文件中的最后一个文件（deleteLastMaped）...... 10 
3.9 根据指定的 offset 找到所在文件（findMapedFileByOffset）

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

4.11 获取指定位置所在文件的下一个文件的起始偏移量
（rollNextFile）........................................... 18 
4.12DefaultAppendMessageCallback 类的实现 ................ 18 
5  Consumequeue .............................................. 19 
5.1 ConsumeQueue 类结构 .................................. 20 
5.2 删除指定偏移量之后的逻辑文件（truncateDirtyLogicFiles）

5.3 恢复 ConsumeQueue 内存数据（recover）................. 21 
5.4 查找消息发送时间最接近 timestamp 逻辑队列的 offset
（getOffsetInQueueByTime）................................. 21 
5.5 获取最后一条消息对应物理队列的下一个偏移量
（getLastOffset）.......................................... 22 
5.6 消息刷盘（commit）................................... 22 
5.7 将 commitlog 物理偏移量、消息大小等信息直接写入
consumequeue 中（putMessagePostionInfoWrapper） ............ 22 
5.8 根据消息序号索引获取 consumequeue 数据（getIndexBuffer）

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

9.6 获取指定偏移量之后的所有 commitlog 数据............... 33 
9.7 从指定位置开始追加 commitlog 数据（appendToCommitLog） 33 
9.8 DefaultMessageStore.ReputMessageService 服务线程 ..... 33 
9.9 DefaultMessageStore.DispatchMessageService 服务线程 .. 34 
9.10 加载 ConsumeQueue 队列数据（loadConsumeQueue） ....... 35 
9.11 获取指定队列的最大逻辑 Offset（getMaxOffsetInQuque）. 35 
9.12 获取指定队列的最小逻辑 Offset（getMinOffsetInQuque）. 35 
10 Abort ..................................................... 36 
11 Checkpoint ................................................ 36 
12 HA 高可用 ................................................. 36 
12.1Topic 配置同步 ....................................... 36 
12.2 消费进度信息同步 .................................... 37 
12.3 延迟消费进度信息同步 ................................ 37 
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 配置信息 ................. 46 
BROKER 篇 ................................................................................................................................. 47 
1  Broker 的初始化过程 ....................................... 47 
2  Broker 的启动过程 ......................................... 50 
3  向 NameServer 注册 Broker .................................. 52 
4  清理未使用的 topic 数据 .................................... 53 
5  根据 topic 和 group 查找当前的订阅信息 ...................... 53 
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） ........ 61 
15  处理解锁 MessageQueue 队列的请求（UNLOCK_BATCH_MQ） ...... 62 
16  获取 consumerGroup 名下的所有 Consumer 的 ClientId
（GET_CONSUMER_LIST_BY_GROUP）................................. 62 
17  写入 Consumer 消费失败的消息（CONSUMER_SEND_MSG_BACK） ... 63 

18  处理结束事务消息的请求（END_TRANSACTION） ............... 64 
19  客户端发起更新或创建 Topic（UPDATE_AND_CREATE_TOPIC）.... 64 
PRODUCER 篇 ............................................................................................................................ 65 
1  启动 Producer ............................................. 65 
2  启动 MQClientInstance 类（MQClientInstance.start） ......... 66 
3  更新 Topic 的路由信息（updateTopicRouteInfoFromNameServer） 68 
4  向 Broker 发送心跳消息 ..................................... 71 
5  发送普通消息（send） ...................................... 71 
6  发送定时消息 .............................................. 73 
7  发送顺序消息 .............................................. 74 
8  发送事务消息 .............................................. 75 
9  处理 Broker 检查事务状态的消息（CHECK_TRANSACTION_STATE） .. 76 
10  创建 Topic............................................... 77 
CONSUMER 篇 .............................................................................................................................. 78 
1  启动 Consumer ............................................. 78 
2  向 Broker 同步消费进度 ..................................... 83 
3  RebalanceService 服务线程 ................................. 83 
3.1 广播模式下的 RebalanceImpl.rebalanceByTopic 方法...... 84 
3.2 集群模式下的 RebalanceImpl.rebalanceByTopic 方法...... 85 
3.3 updateProcessQueueTableInRebalance 方法逻辑 .......... 86 
3.4 removeUnnecessaryMessageQueue 方法逻辑 ............... 87 
3.5 获取 MessageQueue 队列的下一个消费偏移量.............. 88 
4  PullMessageService 服务线程 ............................... 90 
4.1 不间断地处理 pullRequestQueue 队列中的 PullRequest 对象 90 
4.2 让 PullRequest 对象延迟加入 pullRequestQueue 队列...... 90 
5  PUSH 模式下的消息消费（DefaultMQPushConsumer） ............ 90 
5.1 拉取消息（pullMessage）.............................. 90 
5.2 内部匿名类 PullCallback .............................. 92 
6  PULL 模式下的消息消费（DefaultMQPullConsumer） ............ 95 
6.1 应用层的使用方式..................................... 96 
6.2 获取队列的消费进度（fetchConsumeOffset）............. 97 
7  顺序消费的逻辑（ConsumeMessageOrderlyService） ............ 97 
7.1 回调业务层定义的消费方法............................. 98 
7.2 处理回调方法 consumeMessage 的执行结果............... 100 
7.3 重新获取锁再消费（tryLockLaterAndReconsume）........ 102 
8  并发消费的逻辑（ConsumeMessageConcurrentlyService） ...... 102 
8.1 回调业务层定义的消费方法............................ 102 
8.2 消息重试——消费失败的信息回传给 Broker ............. 104 
9  底层的拉取消息的 API 接口（PullAPIWrapper.pullKernelImpl） 105 
10  发送远程请求 PULL_MESSAGE 的逻辑 ........................ 106 
10.1 同步发送方式 ....................................... 107 
10.2 异步发送方式 ....................................... 108 
10.3 收到 Broker 的响应消息的处理逻辑 .................... 109 