RocketMQ源码：consumer消费流程

梦里梦外; 2024-03-24 16:01 44阅读 0赞

## 1.拉取消息前的准备工作 ##

> 在去`broker`拉取消息前必须要做的一件很重要的操作：触发重平衡

前面我们分析了`Consumer`的启动流程，其中有两个服务类是特别关注的，一个是重平衡服务类`RebalanceService`，一个是拉取消息的服务类`PullMessageService`。

我们先看下拉取消息的服务类`PullMessageService`，他是一个`异步线程`，启动后将阻塞。

> `PullMessageService`对象是拉取消息的入口

public class PullMessageService extends ServiceThread {
    
           private final LinkedBlockingQueue<PullRequest> pullRequestQueue = new LinkedBlockingQueue<PullRequest>();
      
           //TODO:...省略其他代码.....
      
            @Override
            public void run() {
                log.info(this.getServiceName() + " service started");
    
                while (!this.isStopped()) {
                    try {
                        //TODO:从队列中获取 PullRequest, 刚开始肯定获取不到，那么我们就要看是什么时候将PullRequest放入队列中的呢？
                        PullRequest pullRequest = this.pullRequestQueue.take();
                        this.pullMessage(pullRequest);
                    } catch (InterruptedException ignored) {
                    } catch (Exception e) {
                        log.error("Pull Message Service Run Method exception", e);
                    }
                }
    
                log.info(this.getServiceName() + " service end");
            }
     }       
    复制代码

它要从队列`pullRequestQueue`中获取`PullRequest`对象，但是刚开始肯定获取不到，所以我们就要看是什么时候将`PullRequest`放入到队列中去的。

> 没错，这就要看`重平衡服务`了。

那么接下来我们就看下`重平衡服务`都做了什么？重平衡服务也是一个`异步线程服务`，我们就看下核心逻辑：

private void rebalanceByTopic(final String topic, final boolean isOrder) {
        switch (messageModel) {
            // 广播模式：不需要处理负载均衡，每个消费者都要消费，只需要更新负载信息
            case BROADCASTING: {
                // 更新负载均衡信息，这里传入的参数是mqSet，即所有队列
                Set<MessageQueue> mqSet = this.topicSubscribeInfoTable.get(topic);
                if (mqSet != null) {
                    boolean changed = this.updateProcessQueueTableInRebalance(topic, mqSet, isOrder);
                    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;
            }
            // todo 集群模式
            case CLUSTERING: {
                // 从主题订阅信息缓存表中获取主题的队列信息, 获取这个topic下的所有队列（默认是4个）
                Set<MessageQueue> mqSet = this.topicSubscribeInfoTable.get(topic);
                //发送请求从Broker中获取该消费组内当前所有的消费者客户端ID，主题的队
                //列可能分布在多个Broker上，那么请求该发往哪个Broker呢？
                //RocketeMQ从主题的路由信息表中随机选择一个Broker
                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);
                }
    
                // 如果mqSet、cidAll任意一个为空，则忽略本次消息队列负载
                if (mqSet != null && cidAll != null) {
                    List<MessageQueue> mqAll = new ArrayList<MessageQueue>();
                    mqAll.addAll(mqSet);
    
                    // 对cidAll、mqAll进行排序
                    // 这一步很重要，同一个消费组内看到的视图应保持一致，确保同一个消费队列不会被多个消费者分配
                    Collections.sort(mqAll);
                    Collections.sort(cidAll);
    
                    // 默认是平均分配策略
                    AllocateMessageQueueStrategy strategy = this.allocateMessageQueueStrategy;
    
                    List<MessageQueue> allocateResult = null;
                    try {
                        // todo 分配算法
                        allocateResult = strategy.allocate(
                            this.consumerGroup,
                            this.mQClientFactory.getClientId(),
                            mqAll,
                            cidAll);
                    } catch (Throwable e) {
                        log.error("AllocateMessageQueueStrategy.allocate Exception. allocateMessageQueueStrategyName={}", strategy.getName(),
                            e);
                        return;
                    }
    
                    //TODO:保存了当前消费者需要消费的队列
                    Set<MessageQueue> allocateResultSet = new HashSet<MessageQueue>();
                    if (allocateResult != null) {
                        allocateResultSet.addAll(allocateResult);
                    }
    
    
                    // todo 对比消息队列是否发生变化 更新负载均衡信息，传入参数是 allocateResultSet，即当前consumer分配到的队列
                    boolean changed = this.updateProcessQueueTableInRebalance(topic, allocateResultSet, isOrder);
                    if (changed) {
                        log.info(
                            "rebalanced result changed. allocateMessageQueueStrategyName={}, group={}, topic={}, clientId={}, mqAllSize={}, cidAllSize={}, rebalanceResultSize={}, rebalanceResultSet={}",
                            strategy.getName(), consumerGroup, topic, this.mQClientFactory.getClientId(), mqSet.size(), cidAll.size(),
                            allocateResultSet.size(), allocateResultSet);
                        this.messageQueueChanged(topic, mqSet, allocateResultSet);
                    }
                }
                break;
            }
            default:
                break;
        }
    }
    复制代码

然后继续看下`updateProcessQueueTableInRebalance(...)`方法的核心逻辑：

/**
     * 对比消息队列是否发生变化，主要思路是遍历当前负载队列集
     * 合，如果队列不在新分配队列的集合中，需要将该队列停止消费并保
     * 存消费进度；遍历已分配的队列，如果队列不在队列负载表中
     * （processQueueTable），则需要创建该队列拉取任务PullRequest，
     * 然后添加到PullMessageService线程的pullRequestQueue中，
     * PullMessageService才会继续拉取任务
     */
    private boolean updateProcessQueueTableInRebalance(final String topic, final Set<MessageQueue> mqSet,
        final boolean isOrder) {
        boolean changed = false;
    
        /**
         * ConcurrentMap〈MessageQueue, ProcessQueue〉
         * processQueueTable是当前消费者负载的消息队列缓存表，如果缓存表
         * 中的MessageQueue不包含在mqSet中，说明经过本次消息队列负载后，
         * 该mq被分配给其他消费者，需要暂停该消息队列消息的消费。方法是
         * 将ProccessQueue的状态设置为droped=true，该ProcessQueue中的消
         * 息将不会再被消费，调用removeUnnecessaryMessageQueue方法判断是
         * 否将MessageQueue、ProccessQueue从缓存表中移除。
         * removeUnnecessaryMessageQueue在RebalanceImple中定义为抽象方
         * 法。removeUnnecessaryMessageQueue方法主要用于持久化待移除
         * MessageQueue的消息消费进度。在推模式下，如果是集群模式并且是
         * 顺序消息消费，还需要先解锁队列
         */
        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;
                    }
                }
            }
        }
    
        /**
         * 遍历本次负载分配到的队列集合，如果
         * processQueueTable中没有包含该消息队列，表明这是本次新增加的消
         * 息队列，首先从内存中移除该消息队列的消费进度，然后从磁盘中读
         * 取该消息队列的消费进度，创建PullRequest对象。这里有一个关键，
         * 如果读取到的消费进度小于0，则需要校对消费进度。RocketMQ提供了
         * CONSUME_FROM_LAST_OFFSET、CONSUME_FROM_FIRST_OFFSET、
         * CONSUME_FROM_TIMESTAMP方式，在创建消费者时可以通过调用
         * DefaultMQPushConsumer#setConsumeFromWhere方法进行设置
         */
        List<PullRequest> pullRequestList = new ArrayList<PullRequest>();
        for (MessageQueue mq : mqSet) {
            if (!this.processQueueTable.containsKey(mq)) {
    
                /**
                 * 经过消息队列重新负载（分配）后，分配到新的消息队列时，首
                 * 先需要尝试向Broker发起锁定该消息队列的请求，如果返回加锁成
                 * 功，则创建该消息队列的拉取任务，否则跳过，等待其他消费者释放
                 * 该消息队列的锁，然后在下一次队列重新负载时再尝试加锁
                 */
                // 顺序消息
                if (isOrder && !this.lock(mq)) {
                    log.warn("doRebalance, {}, add a new mq failed, {}, because lock failed", consumerGroup, mq);
                    continue;
                }
    
                this.removeDirtyOffset(mq);
                ProcessQueue pq = new ProcessQueue();
                // todo PullRequest的nextOffset计算逻辑位于RebalancePushImpl#computePullFromWhere
                long nextOffset = this.computePullFromWhere(mq);
                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);
                }
            }
        }
    
        // todo 将PullRequest加入PullMessageService，以便唤醒PullMessageService线程
        this.dispatchPullRequest(pullRequestList);
    
        return changed;
    }
    复制代码

这里说一下，它会遍历`MessageQueue`集合，这里的`MessageQueue`集合是经过`分配策略`分配的结果(默认是`平均分配策略`),然后根据每个`MessageQueue`会构建`PullRequest`对象。

> 假如我有两个消费者A,B;并且是集群消费，然后队列数是4个，按照平均分配策略，那么A分配到的队列是2个(queueid=0,1)，而B也分配2个(queueid=2,3); 集群模式下，一个队列只能被同一个消费者组下的一个消费者去消费。 现在是只有一个消费者A，队列数是4个，所以当前消费者A根据分配策略会分配到4个queue,然后遍历属于消费者A的4个queue,为每个queue创建一个`PullRequest`,然后放入集合中。

然后我们继续看下分发`PullRequest`方法的逻辑： org.apache.rocketmq.client.impl.consumer.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 {}&