首页 > 其他 > 详细

RocketMQ源码之 consumer是怎样消费消息的

时间:2020-01-29 20:31:00      阅读:79      评论:0      收藏:0      [点我收藏+]

DefaultMQPushConsumer(以push为例).start()调用mQClientFactory.start(),其中pullMessageService.start在一个循环中从pullRequestQueue中获取pullRequest,

执行pullMessage方法,调用的是remotingClient的invokeAsync方法,传入一个封装了PullCallback的回调函数,等响应到来的时候,PullCallback

的onSuccess方法processQueue.putMessage先把消息放进队列,再由consumeMessageService(实际是一个线程池,分为orderly和concurrently

两种消费模式,具体由自己的内部类ConsumeRequest分别实现)消费,最后调用executePullRequestImmediately把新的pullRequest放入

上面说的pullRequestQueue种,保证循环的去broker获取新的消息,可以看出push其实还是pull。

有个问题就是:pullRequestQueue中第一个pullRequest是什么时候放进去的?

上面说的pullMessageService.start的下面有this.rebalanceService.start(),run方法的循环中,countDownlatch等待,被唤醒后会doRebalance一直

到dispatchPullRequest再到熟悉的上面说的executePullRequestImmediately放入pullRequest。而mQClientFactory.start()方法的下面有

this.mQClientFactory.rebalanceImmediately(),里面即时放行countDownlatch。

上面说的doRebalance值得细看。

RocketMQ源码之 consumer是怎样消费消息的

原文:https://www.cnblogs.com/chuliang/p/12241116.html

(0)
(0)
   
举报
评论 一句话评论(0
关于我们 - 联系我们 - 留言反馈 - 联系我们:wmxa8@hotmail.com
© 2014 bubuko.com 版权所有
打开技术之扣,分享程序人生!