简述RocketMQ消息拉取过程【一】-阿里云开发者社区

前言

相信每一个使用RocketMQ的开发工程师都想了解一下消息是如何被拉到消费端的，消费者在消息拉取过程中都做了什么操作？这些疑问将在接下来的这篇文章中给大家一一解开；

DefaultMQPushConsumer拉消息

RocketMQ有两种类型的消费者，一种是DefaultLitePullConsumer主动拉的模式，另一种是DefaultMQPushConsumer被动接收的模式，我们先来介绍DefaultMQPushConsumer是如何拉消息的。我们需要从MQClientInstance类作为入口，找到pullMessageService，它才是负责从Broker拉取消息的；

启动拉消息线程

在消费者调用start()方法时，我们最终可以发现MQClientInstance.start()方法也被调用了，在里面我们可以发现这样一段代码：

// Start pull service
this.pullMessageService.start();
复制代码

最终我们发现pullMessageService其实是ServiceThread的子类，ServiceThread又实现了Runnable，说白了pullMessageService就是一个任务，pullMessageService.start()源码我们可以看一下：

public void start() {
        log.info("Try to start service thread:{} started:{} lastThread:{}", getServiceName(), started.get(), thread);
        if (!started.compareAndSet(false, true)) {
            return;
        }
        stopped = false;
        // 把pullMessageService作为线程任务
        this.thread = new Thread(this, getServiceName());
        this.thread.setDaemon(isDaemon);
        this.thread.start();
    }
复制代码

这里面就是启动了一个线程，然后这个线程的任务就是pullMessageService本身，所以我们需要重点关注run()方法的实现：

@Override
    public void run() {
        log.info(this.getServiceName() + " service started");
        while (!this.isStopped()) {
            try {
                // 从阻塞队列中取MessageRequest
                MessageRequest messageRequest = this.messageRequestQueue.take();
                // 根据消息请求模式拉消息
                if (messageRequest.getMessageRequestMode() == MessageRequestMode.POP) {
                    this.popMessage((PopRequest)messageRequest);
                } else {
                    this.pullMessage((PullRequest)messageRequest);
                }
            } catch (InterruptedException ignored) {
            } catch (Exception e) {
                log.error("Pull Message Service Run Method exception", e);
            }
        }
        log.info(this.getServiceName() + " service end");
    }
复制代码

维护消息请求队列

在DefaultMQPushConsumer中，当重平衡任务启动后，每个消费者实例将分配到对应的MessageQueue，然后就会创建对应的MessageRequest放进messageRequestQueue中，相关代码可查看RebalanceImpl.updateMessageQueueAssignment()方法，下面贴出相关的代码片段：

boolean allMQLocked = true;
List<PullRequest> pullRequestList = new ArrayList<PullRequest>();
for (MessageQueue mq : mq2PushAssignment.keySet()) {
    if (!this.processQueueTable.containsKey(mq)) {
        if (isOrder && !this.lock(mq)) {
            log.warn("doRebalance, {}, add a new mq failed, {}, because lock failed", consumerGroup, mq);
            allMQLocked = false;
            continue;
        }
        this.removeDirtyOffset(mq);
        ProcessQueue pq = createProcessQueue();
        pq.setLocked(true);
        long nextOffset = -1L;
        try {
            nextOffset = this.computePullFromWhereWithException(mq);
        } catch (Exception e) {
            log.info("doRebalance, {}, compute offset failed, {}", consumerGroup, mq);
            continue;
        }
        if (nextOffset >= 0) {
            ProcessQueue pre = this.processQueueTable.putIfAbsent(mq, pq);
            if (pre != null) {
                log.info("doRebalance, {}, mq already exists, {}", consumerGroup, mq);
            } else {
                log.info("doRebalance, {}, add a new mq, {}", consumerGroup, mq);
                PullRequest pullRequest = new PullRequest();
                pullRequest.setConsumerGroup(consumerGroup);
                pullRequest.setNextOffset(nextOffset);
                pullRequest.setMessageQueue(mq);
                pullRequest.setProcessQueue(pq);
                pullRequestList.add(pullRequest);
                changed = true;
            }
        } else {
            log.warn("doRebalance, {}, add new mq failed, {}", consumerGroup, mq);
        }
    }
}
if (!allMQLocked) {
    mQClientFactory.rebalanceLater(500);
}
this.dispatchPullRequest(pullRequestList, 500);
复制代码

这一段代码就是根据重平衡后生成了pullRequest并放进了messageRequestQueue请求队列中，同样的popRequest也是和上面代码类似：

List<PopRequest> popRequestList = new ArrayList<PopRequest>();
for (MessageQueue mq : mq2PopAssignment.keySet()) {
    if (!this.popProcessQueueTable.containsKey(mq)) {
        PopProcessQueue pq = createPopProcessQueue();
        PopProcessQueue pre = this.popProcessQueueTable.putIfAbsent(mq, pq);
        if (pre != null) {
            log.info("doRebalance, {}, mq pop already exists, {}", consumerGroup, mq);
        } else {
            log.info("doRebalance, {}, add a new pop mq, {}", consumerGroup, mq);
            PopRequest popRequest = new PopRequest();
            popRequest.setTopic(topic);
            popRequest.setConsumerGroup(consumerGroup);
            popRequest.setMessageQueue(mq);
            popRequest.setPopProcessQueue(pq);
            popRequest.setInitMode(getConsumeInitMode());
            popRequestList.add(popRequest);
            changed = true;
        }
    }
}
this.dispatchPopPullRequest(popRequestList, 500);
复制代码

最终都是调用dispatchPullRequest()和dispatchPopPullRequest()把请求放进阻塞队列中：

@Override
    public void dispatchPullRequest(final List<PullRequest> pullRequestList, final long delay) {
        for (PullRequest pullRequest : pullRequestList) {
            if (delay <= 0) {
               this.defaultMQPushConsumerImpl.executePullRequestImmediately(pullRequest);
            } else {
                this.defaultMQPushConsumerImpl.executePullRequestLater(pullRequest, delay);
            }
        }
    }
    @Override
    public void dispatchPopPullRequest(final List<PopRequest> pullRequestList, final long delay) {
        for (PopRequest pullRequest : pullRequestList) {
            if (delay <= 0) {
                this.defaultMQPushConsumerImpl.executePopPullRequestImmediately(pullRequest);
            } else {
                this.defaultMQPushConsumerImpl.executePopPullRequestLater(pullRequest, delay);
            }
        }
    }
复制代码

我们从这里就可以知道，DefaultMQPushConsumer拉消息的请求是通过重平衡来第一次触发的；重平衡会触发拉消息的请求产生，并存放到阻塞队列中；另外pullMessageService将会持续从阻塞队列中取出请求去真正地从Broker拉消息；

从Broker拉消息

以PullRequest作为示例，我们来看一下源码：

private void pullMessage(final PullRequest pullRequest) {
        final MQConsumerInner consumer = this.mQClientFactory.selectConsumer(pullRequest.getConsumerGroup());
        if (consumer != null) {
            DefaultMQPushConsumerImpl impl = (DefaultMQPushConsumerImpl) consumer;
            // 最终还是要去DefaultMQPushConsumerImpl中调用pullMessage()方法
            impl.pullMessage(pullRequest);
        } else {
            log.warn("No matched consumer for the PullRequest {}, drop it", pullRequest);
        }
    }
复制代码

所以我们下一步进入核心代码，真正实现从Broker拉消息，在处理之前会做一系列的判断：

1.状态检测

if (processQueue.isDropped()) {
            log.info("the pull request[{}] is dropped.", pullRequest.toString());
            return;
        }
复制代码

如果在重平衡后，processQueue已经被移除了，那么就不再去拉消息了；

try {
            this.makeSureStateOK();
        } catch (MQClientException e) {
            log.warn("pullMessage exception, consumer state not ok", e);
            this.executePullRequestLater(pullRequest, pullTimeDelayMillsWhenException);
            return;
        }
复制代码

如果当前消费者不在运行状态，那么把这个pullRequest重新放回阻塞队列，延迟3秒再处理；

if (this.isPause()) {
            log.warn("consumer was paused, execute pull request later. instanceName={}, group={}", this.defaultMQPushConsumer.getInstanceName(), this.defaultMQPushConsumer.getConsumerGroup());
            this.executePullRequestLater(pullRequest, PULL_TIME_DELAY_MILLS_WHEN_SUSPEND);
            return;
        }
复制代码

2.流量控制

如果这个消费者被暂停了，那么把这个任务放回阻塞队列，延迟1秒再处理；

if (cachedMessageCount > this.defaultMQPushConsumer.getPullThresholdForQueue()) {
            this.executePullRequestLater(pullRequest, PULL_TIME_DELAY_MILLS_WHEN_FLOW_CONTROL);
            if ((queueFlowControlTimes++ % 1000) == 0) {
                log.warn(
                    "the cached message count exceeds the threshold {}, so do flow control, minOffset={}, maxOffset={}, count={}, size={} MiB, pullRequest={}, flowControlTimes={}",
                    this.defaultMQPushConsumer.getPullThresholdForQueue(), processQueue.getMsgTreeMap().firstKey(), processQueue.getMsgTreeMap().lastKey(), cachedMessageCount, cachedMessageSizeInMiB, pullRequest, queueFlowControlTimes);
            }
            return;
        }
复制代码

如果processQueue中未消费的消息数量大于PullThresholdForQueue阈值，那么触发流控，这个pullRequest将重新放回阻塞队列中，延迟50毫秒再处理；

if (cachedMessageSizeInMiB > this.defaultMQPushConsumer.getPullThresholdSizeForQueue()) {
            this.executePullRequestLater(pullRequest, PULL_TIME_DELAY_MILLS_WHEN_FLOW_CONTROL);
            if ((queueFlowControlTimes++ % 1000) == 0) {
                log.warn(
                    "the cached message size exceeds the threshold {} MiB, so do flow control, minOffset={}, maxOffset={}, count={}, size={} MiB, pullRequest={}, flowControlTimes={}",
                    this.defaultMQPushConsumer.getPullThresholdSizeForQueue(), processQueue.getMsgTreeMap().firstKey(), processQueue.getMsgTreeMap().lastKey(), cachedMessageCount, cachedMessageSizeInMiB, pullRequest, queueFlowControlTimes);
            }
            return;
        }
复制代码

如果processQueue中未消费的消息大小(cachedMessageSizeInMiB代表未消费消息的MB)大于PullThresholdSizeForQueue阈值，那么依然触发流控，延迟50豪秒放回阻塞队列中；

if (!this.consumeOrderly) {
            if (processQueue.getMaxSpan() > this.defaultMQPushConsumer.getConsumeConcurrentlyMaxSpan()) {
                this.executePullRequestLater(pullRequest, PULL_TIME_DELAY_MILLS_WHEN_FLOW_CONTROL);
                if ((queueMaxSpanFlowControlTimes++ % 1000) == 0) {
                    log.warn(
                        "the queue's messages, span too long, so do flow control, minOffset={}, maxOffset={}, maxSpan={}, pullRequest={}, flowControlTimes={}",
                        processQueue.getMsgTreeMap().firstKey(), processQueue.getMsgTreeMap().lastKey(), processQueue.getMaxSpan(),
                        pullRequest, queueMaxSpanFlowControlTimes);
                }
                return;
            }
        }
复制代码

并发模式下，如果最大消费点位与最小消费点位间的差距超出了ConsumeConcurrentlyMaxSpan阈值，默认是2000，也会触发流控，延迟50豪秒放回阻塞队列中；

顺序消息

如果是顺序消息，那么需要先上锁再重新计算一下消费点位，如果锁竞争失败，那么重新放回阻塞队列，并延迟3秒钟：

if (processQueue.isLocked()) {
                if (!pullRequest.isPreviouslyLocked()) {
                    long offset = -1L;
                    try {
                       // 重新计算消费点位
                        offset = this.rebalanceImpl.computePullFromWhereWithException(pullRequest.getMessageQueue());
                        if (offset < 0) {
                            throw new MQClientException(ResponseCode.SYSTEM_ERROR, "Unexpected offset " + offset);
                        }
                    } catch (Exception e) {
                        this.executePullRequestLater(pullRequest, pullTimeDelayMillsWhenException);
                        log.error("Failed to compute pull offset, pullResult: {}", pullRequest, e);
                        return;
                    }
                    boolean brokerBusy = offset < pullRequest.getNextOffset();
                    log.info("the first time to pull message, so fix offset from broker. pullRequest: {} NewOffset: {} brokerBusy: {}",
                        pullRequest, offset, brokerBusy);
                    if (brokerBusy) {
                        log.info("[NOTIFYME]the first time to pull message, but pull request offset larger than broker consume offset. pullRequest: {} NewOffset: {}",
                            pullRequest, offset);
                    }
                    pullRequest.setPreviouslyLocked(true);
                    pullRequest.setNextOffset(offset);
                }
            } else {
               // 锁竞争失败后，延迟3秒放回阻塞队列
                this.executePullRequestLater(pullRequest, pullTimeDelayMillsWhenException);
                log.info("pull message later because not locked in broker, {}", pullRequest);
                return;
            }
复制代码

检查topic设置

在即将发送前还会检查是否设置了subscriptionData，其实就是对应的tag，默认情况下是*，如果没有设置的话，同样会延迟3秒后放回阻塞队列：

final SubscriptionData subscriptionData = this.rebalanceImpl.getSubscriptionInner().get(pullRequest.getMessageQueue().getTopic());
        if (null == subscriptionData) {
            this.executePullRequestLater(pullRequest, pullTimeDelayMillsWhenException);
            log.warn("find the consumer's subscription failed, {}", pullRequest);
            return;
        }
复制代码

构建PullCallback

下一步就是构建PullCallback，但是可以先不看这个，这个属于消息拿到后的回调出来，里面有两个方法，一个是onSuccess(PullResult pullResult)，一个是onException(Throwable e)；如无论消息获取是否成功，依然还会创建新的pullRequest放进阻塞队列中，这样就保证了消息拉取的持续性；

发送请求拉取消息

最终通过一系列的判断后，最后一步就是发出拉取消息的请求：

try {
            this.pullAPIWrapper.pullKernelImpl(
                pullRequest.getMessageQueue(),
                subExpression,
                subscriptionData.getExpressionType(),
                subscriptionData.getSubVersion(),
                pullRequest.getNextOffset(),
                this.defaultMQPushConsumer.getPullBatchSize(),
                this.defaultMQPushConsumer.getPullBatchSizeInBytes(),
                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);
        }
复制代码