该文所涉及的 RocketMQ 源码版本为 4.9.3。

RocketMQ 消息拉取流程

之前在消费者启动流程中描述过 MQClientInstance 的启动流程，在启动过程中会启动 PullMessageService，它继承了ServiceThread，并且 ServiceThread 实现了 Runnable 接口，所以是单独启动了一个线程

public class PullMessageService extends ServiceThread

public abstract class ServiceThread implements Runnable

PullMessageService 的 run 方法如下：

protected volatile boolean stopped = false;

  1. public void run() {
  2.     log.info(this.getServiceName() + " service started");
  4.     while (!this.isStopped()) {
  5.         try {
  6.             PullRequest pullRequest = this.pullRequestQueue.take();
  7.             this.pullMessage(pullRequest);
  8.         } catch (InterruptedException ignored) {
  9.         } catch (Exception e) {
  10.             log.error("Pull Message Service Run Method exception", e);
  11.         }
  12.     }
  14.     log.info(this.getServiceName() + " service end");
  15. }

只要没有停止，线程一直会从 PullRequestQueue 中获取 PullRequest 消息拉取任务，如果队列为空，会一直阻塞，直到有 PullRequest 被放入队列中，如果拿到了 PullRequest 就会调用 pullMessage 方法拉取消息

添加 PullRequest 有两个方法，一个是延迟添加，另一个是立即添加

  1. public void executePullRequestLater(final PullRequest pullRequest, final long timeDelay) {
  2.     if (!isStopped()) {
  3.         this.scheduledExecutorService.schedule(new Runnable() {
  4.             @Override
  5.             public void run() {
  6.                 PullMessageService.this.executePullRequestImmediately(pullRequest);
  7.             }
  8.         }, timeDelay, TimeUnit.MILLISECONDS);
  9.     } else {
  10.         log.warn("PullMessageServiceScheduledThread has shutdown");
  11.     }
  12. }
  14. public void executePullRequestImmediately(final PullRequest pullRequest) {
  15.     try {
  16.         this.pullRequestQueue.put(pullRequest);
  17.     } catch (InterruptedException e) {
  18.         log.error("executePullRequestImmediately pullRequestQueue.put", e);
  19.     }
  20. }

org.apache.rocketmq.client.impl.consumer.PullMessageService#pullMessage

拉取消息流程：

根据消费组获取MQConsumerInner，根据推模式还是拉模式，强转为DefaultMQPushConsumerImpl还是DefaultLitePullConsumerImpl

org.apache.rocketmq.client.impl.consumer.DefaultMQPushConsumerImpl#pullMessage

第1步：获取处理队列,如果队列被丢弃结束

  1. final ProcessQueue processQueue = pullRequest.getProcessQueue();
  3. if (processQueue.isDropped()) {
  4.     log.info("the pull request[{}] is dropped.", pullRequest.toString());
  5.     return;
  6. }

第 2 步：设置最后一次拉取时间戳

pullRequest.getProcessQueue().setLastPullTimestamp(System.currentTimeMillis());

第 3 步：确认消费者是启动的状态，如果不是启动的状态，将PullRequest延迟3s放入队列

  1. try {
  2.     this.makeSureStateOK();
  3. } catch (MQClientException e) {
  4.     log.warn("pullMessage exception, consumer state not ok", e);
  5.     this.executePullRequestLater(pullRequest, pullTimeDelayMillsWhenException);
  6.     return;
  7. }

第 4 步：如果消费者停止了，将PullRequest延迟1s放入队列

  1. if (this.isPause()) {
  2.     log.warn("consumer was paused, execute pull request later. instanceName={}, group={}", this.defaultMQPushConsumer.getInstanceName(), this.defaultMQPushConsumer.getConsumerGroup());
  3.     this.executePullRequestLater(pullRequest,PULL_TIME_DELAY_MILLS_WHEN_SUSPEND);
  4.     return;
  5. }

第 5 步：缓存的消息数量大于1000，将PullRequest延迟50ms放入队列,每触发1000次流控输出警告信息

  1. if (cachedMessageCount > this.defaultMQPushConsumer.getPullThresholdForQueue()) {
  2.     this.executePullRequestLater(pullRequest,PULL_TIME_DELAY_MILLS_WHEN_FLOW_CONTROL);
  3.     if ((queueFlowControlTimes++ % 1000) == 0) {
  4.         log.warn("the cached message count exceeds the threshold {}, so do flow control, minOffset={}, maxOffset={}, count={}, size={} MiB, pullRequest={}, flowControlTimes={}",
  5.             this.defaultMQPushConsumer.getPullThresholdForQueue(), processQueue.getMsgTreeMap().firstKey(), processQueue.getMsgTreeMap().lastKey(), cachedMessageCount, cachedMessageSizeInMiB, pullRequest, queueFlowControlTimes);
  6.     }
  7.     return;
  8. }

第 6 步：缓存的消息大小大于100M 将PullRequest延迟50ms放入队列，每触发1000次输出警告信息

  1. if (cachedMessageSizeInMiB > this.defaultMQPushConsumer.getPullThresholdSizeForQueue()) {
  2.     this.executePullRequestLater(pullRequest,PULL_TIME_DELAY_MILLS_WHEN_FLOW_CONTROL);
  3.     if ((queueFlowControlTimes++ % 1000) == 0) {
  4.         log.warn("the cached message size exceeds the threshold {} MiB, so do flow control, minOffset={}, maxOffset={}, count={}, size={} MiB, pullRequest={}, flowControlTimes={}",
  5.             this.defaultMQPushConsumer.getPullThresholdSizeForQueue(), processQueue.getMsgTreeMap().firstKey(), processQueue.getMsgTreeMap().lastKey(), cachedMessageCount, cachedMessageSizeInMiB, pullRequest, queueFlowControlTimes);
  6.     }
  7.     return;
  8. }

第 7 步：ProcessQueue中消息的最大偏移量与最小偏移量的差值不能大于2000，如果大于2000，触发流控，输出警告信息

  1. if (!this.consumeOrderly) {
  2.     if (processQueue.getMaxSpan() > this.defaultMQPushConsumer.getConsumeConcurrentlyMaxSpan()) {
  3.         this.executePullRequestLater(pullRequest,PULL_TIME_DELAY_MILLS_WHEN_FLOW_CONTROL);
  4.         if ((queueMaxSpanFlowControlTimes++ % 1000) == 0) {
  5.             log.warn("the queue's messages, span too long, so do flow control, minOffset={}, maxOffset={}, maxSpan={}, pullRequest={}, flowControlTimes={}",
  6.                 processQueue.getMsgTreeMap().firstKey(), processQueue.getMsgTreeMap().lastKey(), processQueue.getMaxSpan(),
  7.                 pullRequest, queueMaxSpanFlowControlTimes);
  8.         }
  9.         return;
  10.     }
  11. }

第 8 步：如果ProcessQueue被锁了，判断上一个PullRequest是否被锁，如果没有被锁通过RebalanceImpl计算拉取消息偏移量，如果计算异常，将请求延迟3s加入队列，如果下一次拉取消息 的偏移量大于计算出来的偏移量，说明要拉取的偏移量 大于消费偏移量，对 偏移量 进行修正，设置下一次拉取的偏移量为计算出来的偏移量

  1. if (processQueue.isLocked()) {
  2.     if (!pullRequest.isPreviouslyLocked()) {
  3.         long offset = -1L;
  4.         try {
  5.             offset = this.rebalanceImpl.computePullFromWhereWithException(pullRequest.getMessageQueue());
  6.         } catch (Exception e) {
  7.             this.executePullRequestLater(pullRequest, pullTimeDelayMillsWhenException);
  8.             log.error("Failed to compute pull offset, pullResult: {}", pullRequest, e);
  9.             return;
  10.         }
  11.         boolean brokerBusy = offset < pullRequest.getNextOffset();
  12.         log.info("the first time to pull message, so fix offset from broker. pullRequest: {} NewOffset: {} brokerBusy: {}",
  13.             pullRequest, offset, brokerBusy);
  14.         if (brokerBusy) {
  15.             log.info("[NOTIFYME]the first time to pull message, but pull request offset larger than broker consume offset. pullRequest: {} NewOffset: {}",
  16.                 pullRequest, offset);
  17.         }
  19.         pullRequest.setPreviouslyLocked(true);
  20.         pullRequest.setNextOffset(offset);
  21.     }
  22. } else {
  23.     this.executePullRequestLater(pullRequest, pullTimeDelayMillsWhenException);
  24.     log.info("pull message later because not locked in broker, {}", pullRequest);
  25.     return;
  26. }

第 9 步：根据主题名称获取订阅信息，如果为空，将请求延迟3s放入队列

  1. final SubscriptionData subscriptionData = this.rebalanceImpl.getSubscriptionInner().get(pullRequest.getMessageQueue().getTopic());
  2. if (null == subscriptionData) {
  3.     this.executePullRequestLater(pullRequest, pullTimeDelayMillsWhenException);
  4.     log.warn("find the consumer's subscription failed, {}", pullRequest);
  5.     return;
  6. }

第 10 步：创建PullCallback，为后面调用 拉取消息api做准备

  1. PullCallback pullCallback = new PullCallback() {
  2.     @Override
  3.     public void onSuccess(PullResult pullResult) {
  4.         if (pullResult != null) {
  5.             pullResult = DefaultMQPushConsumerImpl.this.pullAPIWrapper.processPullResult(pullRequest.getMessageQueue(), pullResult,
  6.                 subscriptionData);
  8.             switch (pullResult.getPullStatus()) {
  9.                 caseFOUND:
  10.                     long prevRequestOffset = pullRequest.getNextOffset();
  11.                     pullRequest.setNextOffset(pullResult.getNextBeginOffset());
  12.                     long pullRT = System.currentTimeMillis() - beginTimestamp;
  13.                     DefaultMQPushConsumerImpl.this.getConsumerStatsManager().incPullRT(pullRequest.getConsumerGroup(),
  14.                         pullRequest.getMessageQueue().getTopic(), pullRT);
  16.                     long firstMsgOffset = Long.MAX_VALUE;
  17.                     if (pullResult.getMsgFoundList() == null || pullResult.getMsgFoundList().isEmpty()) {
  18.                         DefaultMQPushConsumerImpl.this.executePullRequestImmediately(pullRequest);
  19.                     } else {
  20.                         firstMsgOffset = pullResult.getMsgFoundList().get(0).getQueueOffset();
  22.                         DefaultMQPushConsumerImpl.this.getConsumerStatsManager().incPullTPS(pullRequest.getConsumerGroup(),
  23.                             pullRequest.getMessageQueue().getTopic(), pullResult.getMsgFoundList().size());
  25.                         boolean dispatchToConsume = processQueue.putMessage(pullResult.getMsgFoundList());
  26.                         DefaultMQPushConsumerImpl.this.consumeMessageService.submitConsumeRequest(
  27.                             pullResult.getMsgFoundList(),
  28.                             processQueue,
  29.                             pullRequest.getMessageQueue(),
  30.                             dispatchToConsume);
  32.                         if (DefaultMQPushConsumerImpl.this.defaultMQPushConsumer.getPullInterval() > 0) {
  33.                             DefaultMQPushConsumerImpl.this.executePullRequestLater(pullRequest,
  34.                                 DefaultMQPushConsumerImpl.this.defaultMQPushConsumer.getPullInterval());
  35.                         } else {
  36.                             DefaultMQPushConsumerImpl.this.executePullRequestImmediately(pullRequest);
  37.                         }
  38.                     }
  40.                     if (pullResult.getNextBeginOffset() < prevRequestOffset
  41.                         || firstMsgOffset < prevRequestOffset) {
  42.                         log.warn("[BUG] pull message result maybe data wrong, nextBeginOffset: {} firstMsgOffset: {} prevRequestOffset: {}",
  43.                             pullResult.getNextBeginOffset(),
  44.                             firstMsgOffset,
  45.                             prevRequestOffset);
  46.                     }
  48.                     break;
  49.                 caseNO_NEW_MSG:
  50.                 caseNO_MATCHED_MSG:
  51.                     pullRequest.setNextOffset(pullResult.getNextBeginOffset());
  53.                     DefaultMQPushConsumerImpl.this.correctTagsOffset(pullRequest);
  55.                     DefaultMQPushConsumerImpl.this.executePullRequestImmediately(pullRequest);
  56.                     break;
  57.                 caseOFFSET_ILLEGAL:
  58.                     log.warn("the pull request offset illegal, {} {}",
  59.                         pullRequest.toString(), pullResult.toString());
  60.                     pullRequest.setNextOffset(pullResult.getNextBeginOffset());
  62.                     pullRequest.getProcessQueue().setDropped(true);
  63.                     DefaultMQPushConsumerImpl.this.executeTaskLater(new Runnable() {
  65.                         @Override
  66.                         public void run() {
  67.                             try {
  68.                                 DefaultMQPushConsumerImpl.this.offsetStore.updateOffset(pullRequest.getMessageQueue(),
  69.                                     pullRequest.getNextOffset(), false);
  71.                                 DefaultMQPushConsumerImpl.this.offsetStore.persist(pullRequest.getMessageQueue());
  73.                                 DefaultMQPushConsumerImpl.this.rebalanceImpl.removeProcessQueue(pullRequest.getMessageQueue());
  75.                                 log.warn("fix the pull request offset, {}", pullRequest);
  76.                             } catch (Throwable e) {
  77.                                 log.error("executeTaskLater Exception", e);
  78.                             }
  79.                         }
  80.                     }, 10000);
  81.                     break;
  82.                 default:
  83.                     break;
  84.             }
  85.         }
  86.     }
  88.     @Override
  89.     public void onException(Throwable e) {
  90.         if (!pullRequest.getMessageQueue().getTopic().startsWith(MixAll.RETRY_GROUP_TOPIC_PREFIX)) {
  91.             log.warn("execute the pull request exception", e);
  92.         }
  94.         DefaultMQPushConsumerImpl.this.executePullRequestLater(pullRequest, pullTimeDelayMillsWhenException);
  95.     }
  96. };

第 11 步：设置系统标记

FLAG_COMMIT_OFFSET: 消费进度 大于0

FLAG_SUSPEND: 拉取消息时支持线程挂起

FLAG_SUBSCRIPTION: 消息过滤机制表达式

FLAG_CLASS_FILTER: 消息过滤机制是否为类过滤

  1. int sysFlag = PullSysFlag.buildSysFlag(
  2.     commitOffsetEnable, // commitOffset
  3.     true, // suspend
  4.     subExpression != null, // subscription
  5.     classFilter // class filter
  6. );

第 12 步：调用 broker 拉取消息

  1. // 每一个参数的含义如下
  2. this.pullAPIWrapper.pullKernelImpl(
  3.     pullRequest.getMessageQueue(), // 要拉取的消息队列
  4.     subExpression, // 消息过滤表达式
  5.     subscriptionData.getExpressionType(), // 过滤表达式类型
  6.     subscriptionData.getSubVersion(), // 时间戳
  7.     pullRequest.getNextOffset(), // 消息拉取的开始偏移量
  8.     this.defaultMQPushConsumer.getPullBatchSize(), // 拉取消息的数量 默认32条
  9.     sysFlag, // 系统标记
  10.     commitOffsetValue, // 消费的偏移量
  11.     BROKER_SUSPEND_MAX_TIME_MILLIS,  // 允许broker挂起的时间 默认15s
  12.     CONSUMER_TIMEOUT_MILLIS_WHEN_SUSPEND, // 允许的超时时间 默认30s
  13.     CommunicationMode.ASYNC, // 默认为异步拉取
  14.     pullCallback // 拉取消息之后的回调
  15. );