RocketMQ源码解析-PushConsumer(2)

继续之前文章的内容。

PushConsumer的启动已经到了mqClientInstance。

public void start() throws MQClientException {PackageConflictDetect.detectFastjson();synchronized (this) {switch (this.serviceState) {case CREATE_JUST:this.serviceState = ServiceState.START_FAILED;// If not specified,looking address from name serverif (null == this.clientConfig.getNamesrvAddr()) {this.clientConfig.setNamesrvAddr(this.mQClientAPIImpl.fetchNameServerAddr());}// Start request-response channelthis.mQClientAPIImpl.start();// Start various schedule tasksthis.startScheduledTask();// Start pull servicethis.pullMessageService.start();// Start rebalance servicethis.rebalanceService.start();// Start push servicethis.defaultMQProducer.getDefaultMQProducerImpl().start(false);log.info("the client factory [{}] start OK", this.clientId);this.serviceState = ServiceState.RUNNING;break;case RUNNING:break;case SHUTDOWN_ALREADY:break;case START_FAILED:throw new MQClientException("The Factory object[" + this.getClientId() + "] has been created before, and failed.", null);default:break;}}
}

继续先看rebalanceService的启动。

@Override
public void run() {log.info(this.getServiceName() + " service started");while (!this.isStoped()) {this.waitForRunning(WaitInterval);this.mqClientFactory.doRebalance();}log.info(this.getServiceName() + " service end");
}

一旦调用了rebalanceImpl的start()方法，那么会每隔一段时间去掉用mqClientInstance的doRebalance()方法。

public void doRebalance() {for (String group : this.consumerTable.keySet()) {MQConsumerInner impl = this.consumerTable.get(group);if (impl != null) {try {impl.doRebalance();}catch (Exception e) {log.error("doRebalance exception", e);}}}
}

在这里会不断的去掉用mqClientInstance下面的消费者的doRebalance(0方法。

@Override
public void doRebalance() {if (this.rebalanceImpl != null) {this.rebalanceImpl.doRebalance();}
}

在这里，pushConsumer会直接调用自己的rebalanceImpl的deRebalance()方法。激动人心的地方在于，pull消费者和push消费者在实现的区别在于，两者分别采用了rebalancePullImpl和rebalancePushImpl，也就是说两者在拉取消息上的区别终于要体现出来了！

前面的文章已经解释过rebalancePullImpl的具体实现。现在来看push消费者是如何实现rebalancePushImpl的。

不管是push还是pull也好，两个rebalanceImpl都是继承自rebalanceImpl的。doRebalance()方法都实现自rebalanceImpl中。

public void doRebalance() {Map<String, SubscriptionData> subTable = this.getSubscriptionInner();if (subTable != null) {for (final Map.Entry<String, SubscriptionData> entry : subTable.entrySet()) {final String topic = entry.getKey();try {this.rebalanceByTopic(topic);} catch (Exception e) {if (!topic.startsWith(MixAll.RETRY_GROUP_TOPIC_PREFIX)) {log.warn("rebalanceByTopic Exception", e);}}}}this.truncateMessageQueueNotMyTopic();
}

在这里都会调用rebalanceByTopic()方法根据topic进行负载均衡。

private void rebalanceByTopic(final String topic) {switch (messageModel) {case BROADCASTING: {Set<MessageQueue> mqSet = this.topicSubscribeInfoTable.get(topic);if (mqSet != null) {boolean changed = this.updateProcessQueueTableInRebalance(topic, mqSet);if (changed) {this.messageQueueChanged(topic, mqSet, mqSet);log.info("messageQueueChanged {} {} {} {}",//consumerGroup,//topic,//mqSet,//mqSet);}} else {log.warn("doRebalance, {}, but the topic[{}] not exist.", consumerGroup, topic);}break;}case CLUSTERING: {Set<MessageQueue> mqSet = this.topicSubscribeInfoTable.get(topic);List<String> cidAll = this.mQClientFactory.findConsumerIdList(topic, consumerGroup);if (null == mqSet) {if (!topic.startsWith(MixAll.RETRY_GROUP_TOPIC_PREFIX)) {log.warn("doRebalance, {}, but the topic[{}] not exist.", consumerGroup, topic);}}if (null == cidAll) {log.warn("doRebalance, {} {}, get consumer id list failed", consumerGroup, topic);}if (mqSet != null && cidAll != null) {List<MessageQueue> mqAll = new ArrayList<MessageQueue>();mqAll.addAll(mqSet);Collections.sort(mqAll);Collections.sort(cidAll);AllocateMessageQueueStrategy strategy = this.allocateMessageQueueStrategy;List<MessageQueue> allocateResult = null;try {allocateResult = strategy.allocate(//this.consumerGroup, //this.mQClientFactory.getClientId(), //mqAll,//cidAll);} catch (Throwable e) {log.error("AllocateMessageQueueStrategy.allocate Exception. allocateMessageQueueStrategyName={}",strategy.getName(), e);return;}Set<MessageQueue> allocateResultSet = new HashSet<MessageQueue>();if (allocateResult != null) {allocateResultSet.addAll(allocateResult);}boolean changed = this.updateProcessQueueTableInRebalance(topic, allocateResultSet);if (changed) {log.info("rebalanced allocate source. allocateMessageQueueStrategyName={}, group={}, topic={}, mqAllSize={}, cidAllSize={}, mqAll={}, cidAll={}",strategy.getName(), consumerGroup, topic, mqSet.size(), cidAll.size(), mqSet, cidAll);log.info("rebalanced result changed. allocateMessageQueueStrategyName={}, group={}, topic={}, ConsumerId={}, rebalanceSize={}, rebalanceMqSet={}",strategy.getName(), consumerGroup, topic, this.mQClientFactory.getClientId(),allocateResultSet.size(), mqAll.size(), cidAll.size(), allocateResultSet);this.messageQueueChanged(topic, mqSet, allocateResultSet);}}break;}default:break;}
}

在广播模式下，由于所有消费者都将收到所订阅的topic的消息，那么就没有调用负载均衡策略的必要，会直接去根据现有的消息队列去更新processQueue的数据。而在集群模式下，就会调用一开始配置好了的负载均衡策略去尝试对消息队列进行负载平衡，然后再去尝试用心分配完毕的消息队列去更新processQueue。在这里，pull和push的区别并没有体现出来，这里的具体操作也在前面的文章当中解释过。

重点在更新processQueue的updateProcessQueueTableInRebalance()方法。

private boolean updateProcessQueueTableInRebalance(final String topic, final Set<MessageQueue> mqSet) {boolean changed = false;Iterator<Entry<MessageQueue, ProcessQueue>> it = this.processQueueTable.entrySet().iterator();while (it.hasNext()) {Entry<MessageQueue, ProcessQueue> next = it.next();MessageQueue mq = next.getKey();ProcessQueue pq = next.getValue();if (mq.getTopic().equals(topic)) {if (!mqSet.contains(mq)) {pq.setDropped(true);if (this.removeUnnecessaryMessageQueue(mq, pq)) {it.remove();changed = true;log.info("doRebalance, {}, remove unnecessary mq, {}", consumerGroup, mq);}}else if (pq.isPullExpired()) {switch (this.consumeType()) {case CONSUME_ACTIVELY:break;case CONSUME_PASSIVELY:pq.setDropped(true);if (this.removeUnnecessaryMessageQueue(mq, pq)) {it.remove();changed = true;log.error("[BUG]doRebalance, {}, remove unnecessary mq, {}, because pull is pause, so try to fixed it",consumerGroup, mq);}break;default:break;}}}}List<PullRequest> pullRequestList = new ArrayList<PullRequest>();for (MessageQueue mq : mqSet) {if (!this.processQueueTable.containsKey(mq)) {PullRequest pullRequest = new PullRequest();pullRequest.setConsumerGroup(consumerGroup);pullRequest.setMessageQueue(mq);pullRequest.setProcessQueue(new ProcessQueue());long nextOffset = this.computePullFromWhere(mq);if (nextOffset >= 0) {pullRequest.setNextOffset(nextOffset);pullRequestList.add(pullRequest);changed = true;this.processQueueTable.put(mq, pullRequest.getProcessQueue());log.info("doRebalance, {}, add a new mq, {}", consumerGroup, mq);} else {log.warn("doRebalance, {}, add new mq failed, {}", consumerGroup, mq);}}}this.dispatchPullRequest(pullRequestList);return changed;
}

前面的操作仍旧是pull与push相同，根据新分配的消息队列去更新与之对应的processQueue，前文都有详细的讲过。重点在于最后调用的dispatchPullRequest()方法，在调用之前组装了pullRequest（消息队列，消费组，与消费队列对应的processQueue，以及该消息队列下一次消费的进度），在pull消费者中，dispatchPullRequest()方法并没有给出具体实现。但是，push消费者里给出了详细的实现，将刚才的pullRequest组成的list作为参数传入。

下面是rebalancePushImpl给出的dispatchPullRequest()方法的实现。

@Override
public void dispatchPullRequest(List<PullRequest> pullRequestList) {for (PullRequest pullRequest : pullRequestList) {this.defaultMQPushConsumerImpl.executePullRequestImmediately(pullRequest);log.info("doRebalance, {}, add a new pull request {}", consumerGroup, pullRequest);}
}

会将每一个pullRequest作为参数交由defaultMQPushConsumerImpl来调用。

public void executePullRequestImmediately(final PullRequest pullRequest) {this.mQClientFactory.getPullMessageService().executePullRequestImmediately(pullRequest);
}

在这里会直接交由mqClientInstance的pullMessageService调用executePullRequestImmediately()方法来实现。那么，我们就需要回到之前的pullMessageService了。

pullMessageService的启动也在mqClientInstance的start()方法当中。

@Override
public void run() {log.info(this.getServiceName() + " service started");while (!this.isStoped()) {try {PullRequest pullRequest = this.pullRequestQueue.take();if (pullRequest != null) {this.pullMessage(pullRequest);}}catch (InterruptedException e) {}catch (Exception e) {log.error("Pull Message Service Run Method exception", e);}}log.info(this.getServiceName() + " service end");
}

在pullMessageService里的run()方法会不断尝试往pullRequestQueue当中去取pullRequest。pullRequestQueue作为阻塞队列，那么在这个阻塞队列当中的pullFequest是什么时候加入队列的呢？很显然，在之前由rebalanceImpl调用的executePullRequestImmediately()方法，将pullRequest加入了阻塞队列。

public void executePullRequestImmediately(final PullRequest pullRequest) {try {this.pullRequestQueue.put(pullRequest);}catch (InterruptedException e) {log.error("executePullRequestImmediately pullRequestQueue.put", e);}
}

在run()方法中去得到了pullRequest之后，会直接调用pullRequest()方法，来拉消息。

private void pullMessage(final PullRequest pullRequest) {final MQConsumerInner consumer = this.mQClientFactory.selectConsumer(pullRequest.getConsumerGroup());if (consumer != null) {DefaultMQPushConsumerImpl impl = (DefaultMQPushConsumerImpl) consumer;impl.pullMessage(pullRequest);}else {log.warn("No matched consumer for the PullRequest {}, drop it", pullRequest);}
}

显而易见了，调用了defaultMQPushConsumerImpl的pullMessage()方法，这里push与pull对于pullMessage的实现有很大的不同。

public void pullMessage(final PullRequest pullRequest) {final ProcessQueue processQueue = pullRequest.getProcessQueue();if (processQueue.isDropped()) {log.info("the pull request[{}] is droped.", pullRequest.toString());return;}pullRequest.getProcessQueue().setLastPullTimestamp(System.currentTimeMillis());try {this.makeSureStateOK();}catch (MQClientException e) {log.warn("pullMessage exception, consumer state not ok", e);this.executePullRequestLater(pullRequest, PullTimeDelayMillsWhenException);return;}if (this.isPause()) {log.warn("consumer was paused, execute pull request later. instanceName={}",this.defaultMQPushConsumer.getInstanceName());this.executePullRequestLater(pullRequest, PullTimeDelayMillsWhenSuspend);return;}long size = processQueue.getMsgCount().get();if (size > this.defaultMQPushConsumer.getPullThresholdForQueue()) {this.executePullRequestLater(pullRequest, PullTimeDelayMillsWhenFlowControl);if ((flowControlTimes1++ % 1000) == 0) {log.warn("the consumer message buffer is full, so do flow control, {} {} {}", size,pullRequest, flowControlTimes1);}return;}if (!this.consumeOrderly) {if (processQueue.getMaxSpan() > this.defaultMQPushConsumer.getConsumeConcurrentlyMaxSpan()) {this.executePullRequestLater(pullRequest, PullTimeDelayMillsWhenFlowControl);if ((flowControlTimes2++ % 1000) == 0) {log.warn("the queue's messages, span too long, so do flow control, {} {} {}",processQueue.getMaxSpan(), pullRequest, flowControlTimes2);}return;}}final SubscriptionData subscriptionData =this.rebalanceImpl.getSubscriptionInner().get(pullRequest.getMessageQueue().getTopic());if (null == subscriptionData) {// 由于并发关系，即使找不到订阅关系，也要重试下，防止丢失PullRequestthis.executePullRequestLater(pullRequest, PullTimeDelayMillsWhenException);log.warn("find the consumer's subscription failed, {}", pullRequest);return;}final long beginTimestamp = System.currentTimeMillis();PullCallback pullCallback = new PullCallback() {@Overridepublic void onSuccess(PullResult pullResult) {if (pullResult != null) {pullResult =DefaultMQPushConsumerImpl.this.pullAPIWrapper.processPullResult(pullRequest.getMessageQueue(), pullResult, subscriptionData);switch (pullResult.getPullStatus()) {case FOUND:long prevRequestOffset = pullRequest.getNextOffset();pullRequest.setNextOffset(pullResult.getNextBeginOffset());long pullRT = System.currentTimeMillis() - beginTimestamp;DefaultMQPushConsumerImpl.this.getConsumerStatsManager().incPullRT(pullRequest.getConsumerGroup(), pullRequest.getMessageQueue().getTopic(), pullRT);long firstMsgOffset = Long.MAX_VALUE;if (pullResult.getMsgFoundList() == null || pullResult.getMsgFoundList().isEmpty()) {DefaultMQPushConsumerImpl.this.executePullRequestImmediately(pullRequest);}else {firstMsgOffset = pullResult.getMsgFoundList().get(0).getQueueOffset();DefaultMQPushConsumerImpl.this.getConsumerStatsManager().incPullTPS(pullRequest.getConsumerGroup(), pullRequest.getMessageQueue().getTopic(),pullResult.getMsgFoundList().size());boolean dispathToConsume = processQueue.putMessage(pullResult.getMsgFoundList());DefaultMQPushConsumerImpl.this.consumeMessageService.submitConsumeRequest(//pullResult.getMsgFoundList(), //processQueue, //pullRequest.getMessageQueue(), //dispathToConsume);if (DefaultMQPushConsumerImpl.this.defaultMQPushConsumer.getPullInterval() > 0) {DefaultMQPushConsumerImpl.this.executePullRequestLater(pullRequest,DefaultMQPushConsumerImpl.this.defaultMQPushConsumer.getPullInterval());}else {DefaultMQPushConsumerImpl.this.executePullRequestImmediately(pullRequest);}}if (pullResult.getNextBeginOffset() < prevRequestOffset//|| firstMsgOffset < prevRequestOffset) {log.warn("[BUG] pull message result maybe data wrong, nextBeginOffset: {} firstMsgOffset: {} prevRequestOffset: {}",//pullResult.getNextBeginOffset(),//firstMsgOffset,//prevRequestOffset);}break;case NO_NEW_MSG:pullRequest.setNextOffset(pullResult.getNextBeginOffset());DefaultMQPushConsumerImpl.this.correctTagsOffset(pullRequest);DefaultMQPushConsumerImpl.this.executePullRequestImmediately(pullRequest);break;case NO_MATCHED_MSG:pullRequest.setNextOffset(pullResult.getNextBeginOffset());DefaultMQPushConsumerImpl.this.correctTagsOffset(pullRequest);DefaultMQPushConsumerImpl.this.executePullRequestImmediately(pullRequest);break;case OFFSET_ILLEGAL:log.warn("the pull request offset illegal, {} {}",//pullRequest.toString(), pullResult.toString());pullRequest.setNextOffset(pullResult.getNextBeginOffset());pullRequest.getProcessQueue().setDropped(true);DefaultMQPushConsumerImpl.this.executeTaskLater(new Runnable() {@Overridepublic void run() {try {DefaultMQPushConsumerImpl.this.offsetStore.updateOffset(pullRequest.getMessageQueue(), pullRequest.getNextOffset(), false);DefaultMQPushConsumerImpl.this.offsetStore.persist(pullRequest.getMessageQueue());DefaultMQPushConsumerImpl.this.rebalanceImpl.removeProcessQueue(pullRequest.getMessageQueue());log.warn("fix the pull request offset, {}", pullRequest);}catch (Throwable e) {log.error("executeTaskLater Exception", e);}}}, 10000);break;default:break;}}}@Overridepublic void onException(Throwable e) {if (!pullRequest.getMessageQueue().getTopic().startsWith(MixAll.RETRY_GROUP_TOPIC_PREFIX)) {log.warn("execute the pull request exception", e);}DefaultMQPushConsumerImpl.this.executePullRequestLater(pullRequest,PullTimeDelayMillsWhenException);}};boolean commitOffsetEnable = false;long commitOffsetValue = 0L;if (MessageModel.CLUSTERING == this.defaultMQPushConsumer.getMessageModel()) {commitOffsetValue =this.offsetStore.readOffset(pullRequest.getMessageQueue(),ReadOffsetType.READ_FROM_MEMORY);if (commitOffsetValue > 0) {commitOffsetEnable = true;}}String subExpression = null;boolean classFilter = false;SubscriptionData sd =this.rebalanceImpl.getSubscriptionInner().get(pullRequest.getMessageQueue().getTopic());if (sd != null) {if (this.defaultMQPushConsumer.isPostSubscriptionWhenPull() && !sd.isClassFilterMode()) {subExpression = sd.getSubString();}classFilter = sd.isClassFilterMode();}int sysFlag = PullSysFlag.buildSysFlag(//commitOffsetEnable, // commitOffsettrue, // suspendsubExpression != null,// subscriptionclassFilter // class filter);try {this.pullAPIWrapper.pullKernelImpl(//pullRequest.getMessageQueue(), // 1subExpression, // 2subscriptionData.getSubVersion(), // 3pullRequest.getNextOffset(), // 4this.defaultMQPushConsumer.getPullBatchSize(), // 5sysFlag, // 6commitOffsetValue,// 7BrokerSuspendMaxTimeMillis, // 8ConsumerTimeoutMillisWhenSuspend, // 9CommunicationMode.ASYNC, // 10pullCallback// 11);}catch (Exception e) {log.error("pullKernelImpl exception", e);this.executePullRequestLater(pullRequest, PullTimeDelayMillsWhenException);}
}

首选更新pullRequest里的pullQueue的最新更新时间为当前时间。接下来是对于流量控制消费者状态消息长度等一系列的判断，如果当前发送条件不符合当前消费者配置的，将会将其丢入定时任务线程池中在一定的timeDelay之后重新尝试发送。

重点将是这里pullCallBack的生成。Push采用了异步的发送方式，在接收到来自broker的消息回复之后，将会回调pullCallBack里的onSuccess或者onException方法。具体是如何在异步条件下调用到的，前面的文章已经解释的非常清楚了。先继续看发送拉取消息请求的步骤，再回来看这里对于消息回复的处理。

如果是集群模式，将会重新去试图从Broker获取最新的对应的消费队列的进度，会从remoteBrokerOffsetStore当中调用readOffset()方法去获得新的消费进度。

之后消息的异步发送与pull一样，不加详细解释。

回到消息的回调部分，也就是onSuccess()方法，如果消息接收成功，将会调用processPullResut()方法来对消息进行反序列化以及tag过滤。

在这里，如果并没有接收到新的消息或者消息都被tag所过滤掉,那么pullRequest都将从新更新下一次的消费进度，重新放进pullMessageService的阻塞队列等待下一次消息的拉取。如果找到了新的消息，并且并没有被tag所过滤掉，那么将会调用processQueue的putMessage()方法将已经反序列化了的拉取过来的消息存放在processQueue里的treeset当中。

public boolean putMessage(final List<MessageExt> msgs) {boolean dispatchToConsume = false;try {this.lockTreeMap.writeLock().lockInterruptibly();try {int validMsgCnt = 0;for (MessageExt msg : msgs) {MessageExt old = msgTreeMap.put(msg.getQueueOffset(), msg);if (null == old) {validMsgCnt++;this.queueOffsetMax = msg.getQueueOffset();}}msgCount.addAndGet(validMsgCnt);if (!msgTreeMap.isEmpty() && !this.consuming) {dispatchToConsume = true;this.consuming = true;}if (!msgs.isEmpty()) {MessageExt messageExt = msgs.get(msgs.size() - 1);String property = messageExt.getProperty(MessageConst.PROPERTY_MAX_OFFSET);if (property != null) {long accTotal = Long.parseLong(property) - messageExt.getQueueOffset();if (accTotal > 0) {this.msgAccCnt = accTotal;}}}}finally {this.lockTreeMap.writeLock().unlock();}}catch (InterruptedException e) {log.error("putMessage exception", e);}return dispatchToConsume;
}

在将消息存放在treeMap当中之后，接下来将会在consumeMessageService进行消息的消费。

RocketMQ源码解析-PushConsumer(2)相关推荐

RocketMQ源码解析-PushConsumer(1)
PushConsumer的启动. DefaultMQPushConusmer执行start()方法,然后直接调用DefaultMQPushConusmer的start()方法. public void ...
6、RocketMQ 源码解析之 Broker 启动(上)
上面一篇我们介绍了 RocketMQ 的元数据管理,它是通过自定义一个 KV 服务器.并且其它服务在 NameServer 注册服务信息的时候都是全量注册.如果 RocketMQ 的拓扑图当中有多台 ...
RocketMQ源码解析之broker文件清理
原创不易,转载请注明出处文章目录 1. broker 清理文件介绍 1.1 哪些文件需要清理 1.2 RocketMQ文件清理的机制 2.源码解析 2.1 清理commitlog 2.2 Consu ...
RocketMQ源码解析之消息消费者(consume Message)
原创不易,转载请注明出处文章目录前言 1.消息流程介绍 2.源码解析 2.1 并发消费 2.2 顺序消费前言我们在<RocketMQ源码解析之消息消费者(pullMessage)> ...
rocketmq源码解析之name启动（一）
2019独角兽企业重金招聘Python工程师标准>>> 说在前面主要解析namrsrv启动部分,namesrv配置加载.netty server创建.注册出处理器. 正文源码解析 ...
RocketMQ源码解析：Filtersrv
???关注微信公众号:[芋艿的后端小屋]有福利: RocketMQ / MyCAT / Sharding-JDBC 所有源码分析文章列表 RocketMQ / MyCAT / Sharding-JDB ...
消息中间件RocketMQ源码解析-- --调试环境搭建
1. 依赖工具 JDK :1.8+ Maven IntelliJ IDEA 2. 源码拉取从官方仓库 [https://github.com/apache/rocketmq) Fork 出属于自己的 ...
RocketMQ源码解析-Consumer启动(1)
DefaultMQPullConsumer继承了ClientConfig类,作为主动拉获取消息的消费者实现接口的管理与相关属性的配置(与PushConsumer对应).相比生产者,消费者配置的属性要复 ...
RocketMQ源码解析-Broker的HA实现
以master异步复制为例子. 在rocketmq的slave broker机子当中,会在DefaultMessageStore的启动当中启动自己的HaService来进行自己的ha服务. publi ...

RocketMQ源码解析-PushConsumer(2)

RocketMQ源码解析-PushConsumer(2)相关推荐

最新文章

热门文章