RocketMQ源码解析之消息消费者(pullMessage)

ゞ浴缸里的玫瑰 2022-12-25 12:57 197阅读 0赞

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

### 文章目录 ###

*   *   *  前言
         *  1.pullMessage原理介绍
         *  2.pullMessage源码解析
         *  总结

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

### 前言 ###

在《[RocketMQ源码解析之消息消费者(rebalance)][RocketMQ_rebalance]》一文中我们介绍了消息消费者push模式的一个rebalance的实现，关于这个rebalance 其实就是解决多个消息消费者实例怎么分配多个MessageQueue的问题，这边默认实现是平均分配的方式，当然RocketMQ还提供了很多种其他分配算法，分配完之后，要对新分配的MessageQueue创建对应ProcessQueue，然后封装PullRequest（其实就是拉取消息请求）然后将这些PullRequest分发出去，关于PullRequest 的处理就是咱们这一篇要介绍的内容了，本文会先介绍PullRequest拉取消息的一个执行流程，接着就是剖析源码。

### 1.pullMessage原理介绍 ###

我们在前言中也介绍了，当消费者实例分配到新的MessageQueue的时候，会生成对应的ProcessQueue同时封装一个PullRequest，然后分发出去，其实这个PullRequest会被交给PullMessageService 这么一个服务来处理，就是塞到这个服务维护的一个队列中国呢，这个PullMessageService 不用想多了，其实也是一个线程，这个线程就不停的从队列中取出PullRequest对象，然后进行拉取消息的动作，经过一堆的检查之后，就取找broker拉取一批消息回来（默认是拉取32个消息），采用异步的方式进行拉取，拉取回来消息之后，就将消息放入ProcessQueue中的一个TreeMap中，key就是消息的offset ，然后value就是消息，大致上拉取消息的核心原理就这么多，如下图：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70]

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

这里我们就从rebalance分发PullRequest开始  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 1]  
可以看到这里循环调用了defaultMQPushConsumerImpl 的立即执行PullRequest方法，我们跟进去看看  
![在这里插入图片描述][20201202005838638.png]  
这里又将PullRequest交给了PullMessageService 这个服务，我们接着往下，进入PullMessageService 的executePullRequestImmediately 方法中看看  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 2]  
这里是直接将PullRequest对象放到一个LinkedBlockingQueue中了，这个对列是在PullMessageService 维护着，这里介绍一下PullMessageService ，它其实是一个线程，在启动MQClient的时候，会将它也启动，然后就一直运行自己那个run方法。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 3]  
我们可以看到，他就是一直不停的从这个队列中取出PullRequest对象，然后交给自己的pullMessage 方法处理  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 4]  
这里其实就是根据消费者组获取对应的消费者DefaultMQPushConsumerImpl对象，然后交给这个对象的pullMessage 方法来去拉取消息，这个方法比较长，前面是一堆校验的，我们这里就不贴出来了，我只看下这个重要的部分  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 5]  
在这个方法的最后面有这么一段代码，这个代码其实就是向broker 发起拉取消息的请求，这里有个pullCallback，我们要注意下，这个是broker响应回来消息的时候，会回调这个对象。发送网络请求的部分我们就不看了，直接看下broker响应回来的处理，看下这个pullCallback的实现  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 6]  
这里如果请求成功的话会执行onSuccess方法，返回一个PullResult回来，然后处理一下Result  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 7]  
这里有个比较重要的点，就是更新下次从哪个broker 上面去获取，这个其实就是broker 返回给你拉取结果的时候，还会带给你建议下次去哪个broker上面获取，接着就是如果发现消息的话，进行解码，然后消息的过滤，消息的一些处理，比如说下面的那个for循环里面的操作，然后返回处理完的PullResult。  
接着会到回调方法中  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 8]  
如果是有消息返回的话，设置一下pullRequest中下次开始拉取消息的offset，这个offset 是broker 给带回来的，然后判断消息列表是不是空，如果是空的话，就将这个pullRequest接着扔到PullMessageService服务的那个队列中，继续下一次拉取，如果有消息的话，就会将消息放入ProcessQueue的一个treeMap中，我们来看看是怎么放的。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 9]  
首先是获取写锁，然后遍历消息列表，将消息放到这个treeMap中，key就是消息的offset ，value就是那个消息，这个样子，就会按照消息offset从小到大排列起来了，如果这个treeMap不是空的话，并且没有在消费状态，就要将它的消费状态设置成running，然后允许分发去消费。  
接着会到 回调的onSuccess方法中，将消息列表放入ProcessQueue中之后，会返回一个是否分发消费的判断值，接着就是向ConsumeMessageService这个服务提交消费请求，进行具体的消费，关于消费这块我们后面的文章会有介绍，这里我们先不看，接着往下走，判断拉取间隔，如果没有的话就将这个PullRequest放到PullMessageService 这个服务的队列中，如果有间隔的话，就等会再放进去，这里默认是没有间隔的。  
接着判断 broker带个你下次从哪个offset拉取的值，如果小于当前这次从哪offset开始的值，或者是当前拉取回来的第一个消息小于这次从哪个offset开始拉取的话，就打印warn日志，就说拉出消息结果可能数据错误。这里我们found部分就介绍完了  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 10]  
如果broker返回 NO\_NEW\_MSG 或者是NO\_MATCHED\_MSG ，这时候会设置一下pullRequest的下次拉取起始offset，然后将PullRequest放到PullMessageService 这个服务的队列中，如果是OFFSET\_ILLEGAL ，非法offset，这个其实broker里面都没有这个消息了，然后还拉取，然后它会设置这ProcessQueue的Dropped 状态是true，也就是不能再去broker拉取消息，不能再消费，并且10s后更新消费offset，将消费offset提交给broker，移除这个ProcessQueue。  
好了，到这我们拉取消息的源码就解析完成了。

### 总结 ###

本文主要是介绍了消息消费者push模式拉取消息的流程，先是rebalance服务将PullRequest提交到PullMessageService 这个服务的一个队列中，然后PullMessageService这个线程会不停的从队列中获取PullRequest，然后发起异步请求去broker中拉取消息，拉取消息回来会回调一个pullcallback对象方法，将拉取回来的消息存储到对应ProcessQueue的一个TreeMap中，接着提交消费请求，并将拉取请求PullRequest 再次放到PullMessageService这个服务的那个队列中，就是这么一个拉取过程，最后我们剖析了关于拉取消息流程的源码。

[RocketMQ_rebalance]: https://inetyoung.blog.csdn.net/article/details/110411433
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70]: /images/20221120/fa70459737aa4c37bd16779b86e701c4.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 1]: https://img-blog.csdnimg.cn/20201202005755213.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv,size_16,color_FFFFFF,t_70
[20201202005838638.png]: https://img-blog.csdnimg.cn/20201202005838638.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 2]: https://img-blog.csdnimg.cn/20201202010033407.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 3]: https://img-blog.csdnimg.cn/20201202010248246.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 4]: https://img-blog.csdnimg.cn/2020120201034333.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 5]: https://img-blog.csdnimg.cn/20201202010637258.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/20201202011201906.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/20201202011551173.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/20201202012100785.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/20201202012647983.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/20201202013754578.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv,size_16,color_FFFFFF,t_70