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

偏执的太偏执、 2022-12-25 15:58 331阅读 0赞

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

### 文章目录 ###

*   *   *  前言
         *  1.消息流程介绍
         *  2.源码解析
         *   *  2.1 并发消费
             *  2.2 顺序消费

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

### 前言 ###

我们在《[RocketMQ源码解析之消息消费者(pullMessage)][RocketMQ_pullMessage]》一文中介绍了消息消费者push模式拉取消息的流程，消息从broker拉取回来之后，将消息放入对应的ProcessQueue的treeMap中，接着就是提交消费请求了，本文主要是介绍下RocketMQ提供的2种消费模式（并发消费，顺序消费）的执行流程与源码剖析。

### 1.消息流程介绍 ###

RocketMQ支持并发消费与顺序消费，这个采用顺序消费还是并发消费，是与你采用实现的MessageListener（这个监听器就是具体处理业务要实现的那个接口）类型有关，如果你实现的MessageListenerConcurrently这个接口的话，就是采用并发消费，如果是实现MessageListenerOrderly 这个接口的话，就是采用顺序消费。  
这里先介绍下并发消费，并发消费是由ConsumeMessageConcurrentlyService 这个服务来处理的，然后当消费请求提交到这个服务的时候，它会根据你每次消费多少（默认是1），按照这个数量将消息列表分成一个个的小集合，封装消费请求，然后提交到线程池中。每个消费请求在执行的时候，会去调用你实现的MessageListener 来处理业务逻辑，然后处理消费结果，如果有失败的，会将失败消息发送给broker 的重试topic中，接着就是更新本地存储消费offset中的值（这个消费offset 会在一定条件下 同步给broker，ProcessQueue移除的时候或者是定时任务，默认是5s触发一次）。  
我们在来介绍下这个顺序消费，顺序消费由ConsumeMessageOrderlyService 这个服务组件来处理，这个组件收到消费请求的时候，会封装成一个ConsumeRequest，然后扔到线程池中，就一个ConsumeRequest，然后就是当线程循环获取到的消息集合，也是根据你每次消费多少，然后一批一批的去调用你实现的MessageListener 来处理业务逻辑，执行完这一批后处理执行结果，如果你处理业务逻辑的时候发生了异常或者执行结果返回null，这个时候就不会继续消费了，会等待一会再消费，默认情况下是暂停1s，1s后再提交消费请求，如果正常消费，没有出现异常啥的，就会更新消费offset，可以说这个顺序消费能够保证你这个消费是有序的，单线程，异常暂停，其实还有就是再rebalance的时候要向broker 申请锁，也就是说只有一个消费者实例能够消费对应的一个queue，消费之前要判断这个锁有没有，然后有没有失效。接下来我们分别看下并发消费与顺序消费的实现源码是什么样子的

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

这个要从DefaultMQPushConsumerImpl启动的时候判断监听器MessageListener类型说起。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70]  
可以看到，不同类型实现的监听器，对应的ConsumeMessageService也就不一样，接下来在看下提交消费请求的代码  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 1]  
第一个参数就是 从broker拉取的消息集合，然后第二个参数是ProcessQueue，第三个是对应的MessageQueue，第四个是要不要去消费

#### 2.1 并发消费 ####

先来看下并发消费的实现  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 2]  
上边这一块就是拉取回来的消息个数正好在你批量消费范围之内的话，就直接封装ConsumeRequst，提交给线程池处理，如果是拉取回来的消息个数大于你批量消费范围的话，就分批封装成ConsumeRequst，扔给线程池处理，默认批量消费是1个，如果你拉回来32个消息，他就会给你封装成ConsumeRequst对象，扔到线程池中并发消费， 注意这个线程池core是20 ，maxCore是60 ，然后队列没有限制大小。  
接下来看下这个ConsumeRequst 任务的run方法实现  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 3]  
先是获取你的messageListener ，然后执行前置hook，接着就是设置重试topic，然后循环给这些消息添加消费时间，最后就是执行你messageListener中实现的consumeMessage 方法进行消费，这里还没有完，注意它的返回状态status。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 4]  
上面这一大块其实就是处理这个status，也就是消费结果，如果你异常了，它这个status就是null，后面如果你是null的话，就设置RECONSUME\_LATER，也就是稍后重新消费，接着就是执行消费后的钩子，接着下面就是一堆记录指标的东西了，如果processQueue没有销毁的话，就处理这个消费结果。我们来看下是怎样处理消费结果的  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 5]  
这里这一段就是计算成功几个，然后失败几个的，这个ackIndex，你可以在消费的时候根据实际成功情况来确认ack，当然你也可以不用设置，只要你消费的时候异常不往上抛，它就认为你消费成功了，一旦让它捕获到异常，你这一批消费都要被放到重试队列中。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 6]  
这一块就是根据这个ackIndex，往后的消息它认为是失败的，然后发给broker 的重试队列中，进行重试消费，如果发送重试消息失败就加到集合中，等一会在本地再消费一下。  
最后就是在processQueue的treeMap中删除消费完的消息，获取一个最小的offset，更新本地对应的消费offset。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 7]  
这个就是将失败的消息发送到broker的重试队列中，可以看到有个延迟等级，这个其实就是根据你的重试次数进行延迟消费的，这里默认的延迟等级是0，但是到broker端，它发现你传过来的是0，它会给你设置成3+重试次数  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 8]

#### 2.2 顺序消费 ####

顺序消费是在ConsumeMessageOrderlyService 这个服务组件中完成的，我们来看下  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 9]  
这里就封装一个消费请求，然后扔到线程池中，也就是从broker 拉回来的数据是顺序执行的。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 10]  
我们这个run方法比较长，但是大体思想就是获取锁，然后再做一些判断，比如说顺序消费的话有没有从broker申请到锁，申请的锁有没有过期，接着就是循环消费了，每次循环之前都要做一堆的判断，再接着往下看  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 11]  
这里上来先判断了一下ConsumeRequest消耗的时间有没有超过1分钟，如果超过了就等会再消费，接着获取每次消费的个数，从ProcessQueue中取出来对应个数的消息，再往后就是执行消费前的hook  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 12]  
你会发现它每次消费前都要判断一下有没有drop，drop了就不要再消费了，drop出现在rebalance的时候，这次你这个消费者实例没有分配到这个MessageQueue，或者这个MessageQueue好久没有消费进行拉取消息消费了，然后它就会给你drop，同时将本地存储的offset 同步给broker ，接下来就是调用你实现的MessageListener 来处理你的业务逻辑进行消费，返回status，下面就需要关注status了  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 13]  
如果你处理业务异常了，或者返回了null，不管怎么说只要status是null，就是暂停一小会，从字面意思上也能看出来，接着就是执行消费后的hook了，最后是处理消费结果，我们看下处理消费结果  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 14]  
先是判断是否是自动提交，默认是的当然你可以改成非自动提交，如果成功的话，就获取一下要提交的offset，如果是等一会的话，就等会提交消费请求，然后continueConsume=false，这个时候，前面的那个for循环就停了，就会等1s后再将消费请求提交，继续消费，反正就是暂停1s。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 15]  
前面那一大段代码就是手动提交的，其实都差不多，主要是看最后面这一行，如果没有drop的话，就更新一下本地存储消费offset，这个一般是存在内存里，然后5s向broker 同步一下。  
到这我们消息消费者的并发消费与顺序消费就介绍完了，并解析了对应的源码实现。

[RocketMQ_pullMessage]: https://inetyoung.blog.csdn.net/article/details/110458916
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70]: /images/20221120/6b34f5478ced4d8d9be500d31d23c04e.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 1]: /images/20221120/2fb6b7470ce342a98ca883521c0c749e.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 2]: /images/20221120/96c01874fb5c42e683eccec2c5ef725a.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 3]: /images/20221120/85d2830688db45029bffe69f1b6391f0.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 4]: /images/20221120/b5551556912c43b0a61d0d139923d496.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 5]: /images/20221120/deb60d6b607f43ef8d545ce443cb0e73.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 6]: /images/20221120/47c9ae8e5c574fcca96bdb6ea715b07a.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 7]: /images/20221120/64852a016d96439ca7800b9de1151466.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 8]: /images/20221120/626ec32395eb430394a7911e33afb495.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 9]: /images/20221120/6c3ffeb5e33b40f6b1cba38718e42072.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 10]: /images/20221120/40b36a398d414decb09ee04489fcad9d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 11]: /images/20221120/88616336268341789f28393aeb7ab1c2.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 12]: /images/20221120/4a954c2ea2064add8af5aa521157a19c.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 13]: /images/20221120/ec8ca1cdea8c4a28ba4482fedd79688b.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 14]: /images/20221120/2f18cbfe7a1546a3be26f70e19136a25.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 15]: /images/20221120/37a909acf81d42e38a36cb5d51916a27.png