简述RocketMQ消息拉取过程

妖狐艹你老母 2024-03-25 19:40 10阅读 0赞

## 前言 ##

相信每一个使用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.defaul