RocketMQ的push模式
我们都知道RocketMQ在消费端有push和pull两种模式,pull模式需要我们手动调用consumer拉消息,而push模式则只需要我们提供一个listener即可实现对消息的监听,而实际上,RocketMQ的push模式是基于pull模式实现的,它没有实现真正的push。
Push模式的消息消费
从之前介绍rocketmq的文章中(见:RocketMQ的负载均衡与Rebalance机制),我们可知道,在应用初始化时,rocketmq的consumer会做一次rebalance获取需要消费的MessageQueue,我们从新回到RebalanceImpl的rebalanceByTopic方法,其中在获取到需要消费的MessageQueue后,会有一个updateProcessQueueTableInRebalance方法的调用,其中包含了消费MessageQueue的入口:
这个dispatchPullRequest就已经说明了一切,是pull,不是push,至于为什么没有实现真正的push,我觉得应该是和消费者的负载均衡有关,如果需要实现真正的push,那么就需要broker感知消费者的变化便做rebalance,而rocketmq的broker是分布式的,实现push的难度相对比较大。
我们继续看dispatchPullRequest的具体实现,在RebalanceImpl类中,dispatchPullRequest方法是一个抽象方法,它有两个实现,分别是RebalancePullImpl和RebalancePushImpl,虽然push也是依靠pull实现,但这里却体现出了二者微妙的差异。
我们先看看push实现的源码:
可以看到,它调用了defaultMQPushConsumerImpl的executePullRequestImmediately方法,很显然,这行代码调用是要立即发起消息的拉取操作,那我们再看看pull的实现类RebalancePullImpl中实现:
可以看到,RebalancePullImpl的dispatchPullRequest方法中是空的,因为pull模式需要我们应用层面调用消息的拉取。
Push模式消息拉取处理
我们最后来看看push模式下,defaultMQPushConsumerImpl.executePullRequestImmediately方法中如何处理消息的接取请求,从调用链中可以走到PullMessageService中,其中代码执行如下:
该方法中的代码处理非常简单,即将PullRequest对象加入到pullRequestQueue中,pullRequestQueue是一个BlockingQueue,那么如果想要知道push模式是如何处理消费消息就非常简单了,我们看看pullRequestQueue的take方法的调用代码就行了,同样是在PullMessageService中:
这个run方法中,不断的从pullRequestQueue中获取PullRequest,然后调用的pullMessage方法,在pullMessage方法调用中又会调用DefaultMQPushConsumerImpl的pullMessage方法,其中会处理listener的回调:
pullMessage中的代码写的比较捉急,这里截取其中的一部分,可以看到,其中调用了pullAPIWrapper.pullKernelImpl方法真正执行消息的拉取,并且有一个pullCallback用于执行拉取消息的加调,我们可以看看callback中的相关代码:
其中submitConsumeRequest是处理消息,而再下面则是重新发起消费,我们最后看看消息处理:
其中是向线程池中提交了一个ConsumeRequest,ConsumeRequest是一个Runnable,我们看看它的run方法的实现:
在这个里面调用了listener的consumeMessage方法。
总结
以上就是整个push模式的处理过程,其中几个关键点:
- 通过pull模式实现
- 主动发起消息的接取,RebalanceImpl类的子类RebalancePushImpl是拉取消息的入口
- 在rebalance后会触发消息的拉取
- 拉取成功后,调用listener的consumeMessage方法