RocketMQ源码解析之broker（pullMessage）

淡淡的烟草味﹌ 2022-12-27 01:21 200阅读 0赞

> 原创不易，转载请注明出处

### 文章目录 ###

*   *   *  前言
         *  1.broker处理消费请求的流程
         *  2.源码解析
         *   *  2.1PullMessageProcessor\#processRequest前半部分
             *  2.2 DefaultMessageStore\#getMessage方法
             *  2.3 PullMessageProcessor\#processRequest后半部分
         *  总结

--------------------

### 前言 ###

我们在之前的消费者文章中介绍了消息消费者消费整体流程，包括rebalance，pullMessage，consumeMessage这个三个重要的步骤，其中pullMessage这个步骤处理拉取消息请求，然后从broker上面拉取请求，接下来我们就看下broker是怎样处理消费请求的，本文还是老套路，先介绍下broker 处理消费请求的整体流程，然后就是从源码角度触发，看看源码是怎样实现的。

### 1.broker处理消费请求的流程 ###

broker 处理消费请求其实很简单，就是拿到消费请求后，消费请求里面肯定带着topic ，queueId， offset，取多少个这四个重要信息，然后获取到对应topic对应queueId的consumeQueue，然后定位到offset位置处，往下取出你要个数的信息。  
这里举个例子，比如说有一个topic是xxx，然后有2个queue，这个时候我们消息消费者 发起消费请求，要消费topic是xxx，queueId是0 ，然后offset =3开始拉取，拉取大小maxMsgNums=2，就是下图这个样子  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70]  
接着获取到3，4这两个之后，遍历，根据它们里面的commitlog offset 再去commitlog获取到对应真实的消息。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 1]  
大概就是这么个意思，就是找到对应在commitlog中的真实消息，封装一下，就返回了。当然这里面我省略了很多细节，我们下面看源码的时候能够介绍到。

### 2.源码解析 ###

#### 2.1PullMessageProcessor\#processRequest前半部分 ####

broker 拉取消息请求是在PullMessageProcessor 这个processor中处理的，有个processRequest 方法我们看下，这个方法前面其实是一堆验证的代码  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 2]  
我们稍微看下，先是反序列化请求头，然后判断当前broker 状态是否是可读的，根据消费者组去订阅组管理器组件中获取这个订阅信息，判断存不存在，这个订阅信息在 消费者给broker发送心跳的时候就会带过去，其实下面还有一片验证的代码，我们直接略过去  
![在这里插入图片描述][20201207134640347.png]  
这里有一行非常重要的，就是去消息存储器中获取消息。下面我们就着重介绍下这个方法

#### 2.2 DefaultMessageStore\#getMessage方法 ####

这个方法很长，我在这里给它分成了三部分，第一部分是获取前准备工作，第二部分就是从ConsumeQueue获取到之后，然后循环处理，第三部分就是获取完成，然后设置带回去的参数。  
先来看下第一部分获取前准备工作  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 3]  
看下这个方法里面的一些参数，第一个是消费者组，第二个是topic，第三个是queueId，第四个是从哪个offset开始获取，第五个是获取多少个消息，第六个是消息过滤器，这里我们主要关心前5个。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 4]  
先是判断broker的存储器是否关闭，然后这个broker运行状态是否可读，接着就是定义了一堆变量，后面那两个红框框起来的非常重要，一个是获取commitlog中最大的offset，一个是根据topic与queueId获取对应的ConsumeQueue。接着往下看  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 5]  
这里是先获取ConsumeQueue最大与最小offset，然后就是判断了，来纠正从哪个offset 开始获取，如果最大offset是0 ，就说明这个ConsumeQueue中没有消息，这个时候状态也就是在这个queue中没有消息，然后纠正了一下下次从哪个offset开始获取，如果你这个offset小于当前ConsumeQueue中最小的那个offset，说明你要的这个offset太小了，它没有存在，把下次开始offset 纠正成最小offset，如果你这个offset 与最大的那个offset一样，它也会给你返回说你offset 超了一个，如果你这个offset大于它最大offset 很多的话，就把返回状态给设置成offset 超了很多。  
下面就是第二部分了，offset在最小offset 与最大offset之间，从ConsumeQueue获取消息，遍历获取真实消息，封装等等。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 6]  
先是根据offset 从ConsumeQueue中获取消息，这一块我们稍微看下  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 7]  
其实就是计算出在ConsumeQueue中真实的偏移量，根据这个偏移量计算出在那个MappedFile中，然后从截取这个offset到MappedFile最后可读的那块buffer  
接着就是遍历这块buffer了，需要注意的是ConsumeQueue每个unit大小是固定的就是20个字节，里面存储着真实消息在commitlog的offset。所以这里遍历的时候增长是按照20 增长的。  
取出每个unit对应的东西，大概是这个样子的。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 8]  
接着就是判断一下这个offsetPy 是否在磁盘上，这个是这个样子判断的，就是看看你这个offset 距离最大的offset多少字节，然后超过内存的40% 就说明在磁盘上了  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 9]  
接着就是判断是否还要继续取数据，这个是根据你这次已经取了多少消息（消息大小，消息数量）与实际要取多少，还有是否在磁盘上决定的。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 10]  
就是如果这次已经取了消息大小跟消息总数是0 的话，就继续取，如果 已经取了消息数量大于等于要取的消息数量的话，就不继续了，如果是在磁盘上的话，取出来消息大小 不能大于64\*1024，也就是64k，消息数量不能大于7，如果超了就不能再取了，如果不是在磁盘上不能超了256k与31个消息。  
接着往下看  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 11]  
这里其实就是根据commitlog 偏移量从commitlog 中获取真实的消息，最后将消息添加到结果集里面，到这我们这个for循环就分析完事了。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 12]  
这一块就是计算你下次从哪个offset 开始获取，然后再就是看看你这个commitlog offset离着最大的那个commitlog offset差距有多大，如果是大于内存的40% 的话，它就会建议你下次别从我这broker 上面读取了，建议你去slave上面读去  
接下来就是看第三部分收尾动作了  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 13]  
前面就是一堆记录统计信息的，重要的是设置下次从哪个offset开始获取，设置一下本次获取到的最大offset与最小offset。

#### 2.3 PullMessageProcessor\#processRequest后半部分 ####

这里就是看看获取完之后做了什么处理  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 14]  
就是将下次从哪个offset 获取，本次最大最小offset 设置到response中，然后建议下次从哪个broker 拉取消息，默认是broker\_id是1的，下面那个switch就是判断slave角色让不让读的问题，不让读的话，就是让消息消费立马去master上拉消息去。  
下面的代码就是一堆拉取状态判断了，我们这里就不看，这里挑出来些比较重要的  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 15]  
这里有两种转换方式，其实也是传输方式吧，一个就是普通的io进行传输，将bytebuffer转成字节数组，设置到response中去，使用堆内存来处理的。一个就是使用ManyMessageTransfer来处理的，重写了transferTo方法。 默认是使用堆来完成转换的。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 16]  
这一段代码就是判断它这次拉取消息请求里面带没带着消费offset，如果带着的话，就找到ConsumerOffset组件，然后更新一下消费offset。  
这个发现也是挺重要的，因为消息消费者除了定时任务5更新一下消费进度，还可以通过拉取消息的时候带着消费offset，进行消费进度的更新。

### 总结 ###

本文主要是介绍了broker 处理消息消费者拉取消息的请求，就是从ConsumeQueue中获取指定消息数量的unit，遍历获取的unit，在根据unit里面存储的commitlog offset 从commitlog 中获取真实的消息这么一个流程，同时我们还有几个新发现，就是broker 根据你消费进度来判断让你下次从master获取消息还是从slave上面获取消息，这就是RocketMQ的读写分离实现，还有就是消息消费者的拉取消息请求中还有提交消费offset的能力。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70]: /images/20221120/a188b7cf664b42db883caa47135c0d8d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 1]: /images/20221120/0900ad03f27348f9b2b42b0cad562400.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 2]: /images/20221120/8c6ef8c43c2c4fa8a7e65cb9ccb139df.png
[20201207134640347.png]: /images/20221120/173c07de12ee418f97b7bbabc8a99c76.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 3]: /images/20221120/418b650410814ed384747cde0dca859d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 4]: /images/20221120/bfae166c64f646188f8b57282da13c38.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 5]: https://img-blog.csdnimg.cn/20201207141811145.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 6]: https://img-blog.csdnimg.cn/2020120714274380.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 7]: https://img-blog.csdnimg.cn/20201207142908537.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 8]: https://img-blog.csdnimg.cn/2020120714355938.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 9]: https://img-blog.csdnimg.cn/20201207143839870.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 10]: https://img-blog.csdnimg.cn/2020120714403879.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 11]: https://img-blog.csdnimg.cn/20201207144450154.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 12]: https://img-blog.csdnimg.cn/20201207144642772.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 13]: https://img-blog.csdnimg.cn/2020120714495159.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 14]: https://img-blog.csdnimg.cn/20201207145359295.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 15]: https://img-blog.csdnimg.cn/20201207150242158.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 16]: https://img-blog.csdnimg.cn/20201207151503308.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv,size_16,color_FFFFFF,t_70