大橙子网站建设,新征程启航
为企业提供网站建设、域名注册、服务器等服务
这篇文章主要介绍RocketMQ中如何实现push consumer消息拉取,文中介绍的非常详细,具有一定的参考价值,感兴趣的小伙伴们一定要看完!
成都创新互联主要从事成都网站建设、做网站、网页设计、企业做网站、公司建网站等业务。立足成都服务东莞,十载网站建设经验,价格优惠、服务专业,欢迎来电咨询建站服务:028-86922220
RebalanceImpl.updateProcessQueueTableInRebalance方法的末尾,对于每一个新生成的ProcessQueue都会创建一个PullRequest执行首次消息拉取操作。PullRequest会通过RebalanceImpl.dispatchPullRequest方法达到DefaultMQPushConsumerImpl.executePullRequestImmediately,然后被投递到PullMessageService的本地队列中。
PulMessageService会启动一个服务线程,不断消费投递到本地队列中的PullRequest,最终调用到DefaultMQPushConsumerImpl.pullMessage方法。PullMessageService被MQClientInstance持有,同一个客户端实例中所有的push consumer产生的PullRequest都会被投递到同一个PullMessageService本地队列中排队等待执行。
DefaultMQPushConsumerImpl.pullMessage是消息拉取的核心方法。该方法首先会执行一系列的限流判断,若命中限流条件则本次执行结束,等待一个固定时间之后会再次将同一个PullRequest投递到PullMessageService中重新触发消息拉取。
DefaultMQPushConsumerImpl.pullMessage核心逻辑:
public void pullMessage(final PullRequest pullRequest) { // 限流判断 .... final long beginTimestamp = System.currentTimeMillis(); // 消息拉取callback PullCallback pullCallback = new PullCallback() {...}; boolean commitOffsetEnable = false; long commitOffsetValue = 0L; if (MessageModel.CLUSTERING == this.defaultMQPushConsumer.getMessageModel()) { // 获取当前队列已经被消费到最新的offset,通过本次pull请求附带在broker上commit该offset 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, // commitOffset true, // suspend subExpression != null, // subscription classFilter // class filter ); try { // 发起pull请求,成功后异步回调pullCallback this.pullAPIWrapper.pullKernelImpl( pullRequest.getMessageQueue(), subExpression, subscriptionData.getExpressionType(), subscriptionData.getSubVersion(), pullRequest.getNextOffset(), this.defaultMQPushConsumer.getPullBatchSize(), sysFlag, commitOffsetValue, BROKER_SUSPEND_MAX_TIME_MILLIS, CONSUMER_TIMEOUT_MILLIS_WHEN_SUSPEND, CommunicationMode.ASYNC, pullCallback ); } catch (Exception e) { log.error("pullKernelImpl exception", e); this.executePullRequestLater(pullRequest, pullTimeDelayMillsWhenException); } }
pullMessage方法中创建的匿名PullCallback用来处理拉取到的消息列表:
PullCallback pullCallback = new PullCallback() { @Override public void onSuccess(PullResult pullResult) { if (pullResult != null) { // 1. 反序列化,并执行过滤 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()); // 2. 保存到本地ProcessQueue中缓存 boolean dispatchToConsume = processQueue.putMessage(pullResult.getMsgFoundList()); // 3. 提交到ConsumeMessageService中,被push到message listener执行业务处理 DefaultMQPushConsumerImpl.this.consumeMessageService.submitConsumeRequest( pullResult.getMsgFoundList(), processQueue, pullRequest.getMessageQueue(), dispatchToConsume); // 4. 提交下一次PullRequest 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; // .... } } } // .... };
上面的第3步consumeMessageService.submitConsumeRequest中将根据并行或串行不同的方式将message提交给listener执行业务处理动作。
消息拉取的整体流程如下:
以上是“RocketMQ中如何实现push consumer消息拉取”这篇文章的所有内容,感谢各位的阅读!希望分享的内容对大家有帮助,更多相关知识,欢迎关注创新互联行业资讯频道!