五张图带你理解 RocketMQ 消费者启动过程

2023-02-28

this 模式 pull

大家好，我是君哥。今天来分享RocketMQ中一个关键的知识点，消费者的启动过程。多数消息队列中，消费者和Broker通信的方式有两种，PUSH模式和PULL模式：PUSH模式：Broker主动把消息推送给订阅的消费者。PULL模式：消费者主动从Broker拉取消息。注意，RocketMQ并没有真正

大家好，我是君哥。

今天来分享 RocketMQ 中一个关键的知识点，消费者的启动过程。

多数消息队列中，消费者和 Broker 通信的方式有两种，PUSH 模式和 PULL 模式：

PUSH 模式：Broker 主动把消息推送给订阅的消费者。
PULL模式：消费者主动从 Broker 拉取消息。注意，RocketMQ 并没有真正实现 PUSH 模式， RocketMQ 中的 PUSH 。模式，本质上也是 PULL 模式，只是消费端封装了轮询过程，相当于开启一个定时线程不停地从 Broker 拉取消息，拉取到消息后唤醒本地业务线程来处理。本文讲解 PULL 模式的启动过程。涉及到到的启动过程如下图：

首先看下面这张图：

图中可以看出，消费者需要注册到 Name Server，拉取消息的时候可以从 Broker 主节点拉取，也可以从 Broker 从节点拉取。

在 RocketMQ 的源码中，拉模式有两个消费者相关的类，其中 DefaultMQPullCons umer 类已经被废弃，官方推荐使用 Defau ltLitePullConsumer 类。下面代码来自官方示例：

public static void main(String[] args) throws Exception {
    DefaultLitePullConsumer litePullConsumer = new DefaultLitePullConsumer("lite_pull_consumer_test");
    litePullConsumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
    litePullConsumer.subscribe("TopicTest", "*");
    //启动方法
    litePullConsumer.start();
    try {
        while (running) {
            //这里可以看到，PULL 模式下消费者需要业务代码主动去拉取消息
            List<MessageExt> messageExts = litePullConsumer.poll();
            System.out.printf("%s%n", messageExts);
        }
    } finally {
        litePullConsumer.shutdown();
    }
}1.
2.
3.
4.
5.
6.
7.
8.
9.
10.
11.
12.
13.
14.
15.
16.

上面代码中消费者属于消费组 lite_pull _consumer_test，订阅了【TopicTest 】这个 Topic 下的所有 tag。下面一起看一下启动方法。下图是消费者启动过程中类调用关系图，图中心的 pullRequestQueu e 是核心，pull 请求会先发送到这个队列，然后循环地拉取处理。

检查启动配置

消费者启动时首先会检查配置，检查的配置项如下：

消费组名称是否合法。包括校验项包括【非空】、【长度小于等于255】、符合正则表达式【^[%|a-zA-Z0-9_-]+$】、【不等于 “DEFAULT_CO NSUMER”】。
消息模式不能是空，包括集群和广播两种模式。
MessageQueue 负载策略不能是空，包括：平均分配策略、循环分配策略、自定义分配策略、按照机房平均分配策略、按照机房就近分配策略、一致性 HASH 策略。
长轮询模式下，消费者连接挂起时间不小于长轮询模式下 Broker 挂起时间，Broker 挂起时间默认 20s，官方不建议修改。

这部分源代码见 DefaultLitePullConsum erImpl#checkConfig。

修改消费者实例名称

如果是集群模式，实例名称改为【进程 ID + “ #” + 系统时间(纳秒 )】，代码如下：

//ClientConfig类
public void changeInstanceNameToPID() {
    if (this.instanceName.equals("DEFAULT")) {
        this.instanceName = UtilAll.getPid() + "#" + System.nanoTime();
    }1.
2.
3.
4.
5.

初始化 MQ 客户端

创建一个 MQClientInstance 实例，然后把消费者注册到 MQClientInstance。

private void initMQClientFactory() throws MQClientException {
    this.mQClientFactory = MQClientManager.getInstance().getOrCreateMQClientInstance(this.defaultLitePullConsumer, this.rpcHook);
    boolean registerOK = mQClientFactory.registerConsumer(this.defaultLitePullConsumer.getConsumerGroup(), this);
    if (!registerOK) {
        this.serviceState = ServiceState.CREATE_JUST;
        throw new MQClientException("The consumer group[" + this.defaultLitePullConsumer.getConsumerGroup()
            + "] has been created before, specify another name please." + FAQUrl.suggestTodo(FAQUrl.GROUP_NAME_DUPLICATE_URL),
            null);
    }
}1.
2.
3.
4.
5.
6.
7.
8.
9.
10.

初始化负载均衡器

对 RebalanceLitePullImpl 实例初始化，给下面的参数赋值：

消费者名称。
消息模型。
MessageQueue 负载均衡策略。
MQ 客户端，上节中初始化的 MQClientInstance 实例。

负载均衡线程启动后，默认每 20s 做一次负载均衡，见如下代码：

//RebalanceService 类
public void run() {
    while (!this.isStopped()) {
        //waitInterval 默认 20s，可以配置
        this.waitForRunning(waitInterval);
        this.mqClientFactory.doRebalance();
    }
}1.
2.
3.
4.
5.
6.
7.
8.

初始化 Wrapper

PullAPIWrapper 这个 Wrapper 类是 MQ-ClientInstance 类的 Wrapper 类，类中 pullKernelImpl 方法对 MQClientInstance 类中的 pullMessage 方法进行了装饰，这个装饰类主要增加了下面功能：

获取 Broker 地址。
检查 RocketMQ 版本。
如果 Broker 是从节点，把 sysFlag 标记偏移量的位改为 0,(偏移量 0x1)。
封装请求 header。
获取 filterServer 地址(如果消费者是通过 filterServer 从 Broker 拉取消息，这里随机获取一个 filterServer 地址)。

代码如下：

//PullAPIWrapper 
public PullResult pullKernelImpl(
    //省略所有参数
) throws MQClientException, RemotingException, MQBrokerException, InterruptedException {
    //1.获取 Broker 地址
    FindBrokerResult findBrokerResult =
        this.mQClientFactory.findBrokerAddressInSubscribe(mq.getBrokerName(),
                                                          this.recalculatePullFromWhichNode(mq), false);
    //省略从 Name sever 更新本地 Broker 缓存逻辑

    if (findBrokerResult != null) {
        {
            //2.检查 RocketMQ 版本
            if (!ExpressionType.isTagType(expressionType)
                && findBrokerResult.getBrokerVersion() < MQVersion.Version.V4_1_0_SNAPSHOT.ordinal()) {
                throw new MQClientException("The broker[" + mq.getBrokerName() + ", "
                                            + findBrokerResult.getBrokerVersion() + "] does not upgrade to support for filter message by " + expressionType, null);
            }
        }
        int sysFlagInner = sysFlag;

        if (findBrokerResult.isSlave()) {
            //3.把偏移量的位改为 0,（偏移量 0x1）
            sysFlagInner = PullSysFlag.clearCommitOffsetFlag(sysFlagInner);
        }
        //4.封装请求 header
        PullMessageRequestHeader  = new PullMessageRequestHeader();
        //省略封装 requestHeader

        String brokerAddr = findBrokerResult.getBrokerAddr();
        if (PullSysFlag.hasClassFilterFlag(sysFlagInner)) {
            //5.获取 filterServer 地址
            brokerAddr = computePullFromWhichFilterServer(mq.getTopic(), brokerAddr);
        }

        PullResult pullResult = this.mQClientFactory.getMQClientAPIImpl().pullMessage(
            brokerAddr,
            requestHeader,
            timeoutMillis,
            communicationMode,
            pullCallback);

        return pullResult;
    }

    throw new MQClientException("The broker[" + mq.getBrokerName() + "] not exist", null);
}1.
2.
3.
4.
5.
6.
7.
8.
9.
10.
11.
12.
13.
14.
15.
16.
17.
18.
19.
20.
21.
22.
23.
24.
25.
26.
27.
28.
29.
30.
31.
32.
33.
34.
35.
36.
37.
38.
39.
40.
41.
42.
43.
44.
45.
46.
47.

初始化 offset 存储器

offset 存储器的 UML 类图如下：

有两个实现类分别对应集群模式和广播模式，本文讨论的集群模式的实现类是 RemoteBrokerOffsetStore。offset 可以存储在本地或者远端服务器。

启动 MQ 客户端

启动 MQ 客户端主要包括如下步骤：

把 serviceState 改为 START_FAIL ED。
初始化 Netty channel。
启动定时任务，包括定时获取 Name Server 地址、从 Name Server 更新 Topic 路由信息、清理过期的 Broker、向 Broker 发送心跳、持久化 offset、定时调整线程池的数量(源码里面这个并没有实现逻辑)。
启动拉取消息的线程，拉取线程的逻辑是从请求队列中不停地取出 pull 请求，然后将请求发送到 Broker 进行拉取消息，代码如下：

//PullMessageService类
public void run() {
    log.info(this.getServiceName() + " service started");

    while (!this.isStopped()) {
        try {
            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");
}1.
2.
3.
4.
5.
6.
7.
8.
9.
10.
11.
12.
13.
14.
15.
16.

从下面的代码可以看出，PULL 拉取消息最终使用了 DefaultMQPushConsumer Impl，所以 PULL 模式和 PUSH 模式拉取消息的逻辑是一样的。

private void pullMessage(final PullRequest pullRequest) {
    final MQConsumerInner consumer = this.mQClientFactory.selectConsumer(pullRequest.getConsumerGroup());
    if (consumer != null) {
        DefaultMQPushConsumerImpl impl = (DefaultMQPushConsumerImpl) consumer;
        impl.pullMessage(pullRequest);
    } else {
        log.warn("No matched consumer for the PullRequest {}, drop it", pullRequest);
    }
}1.
2.
3.
4.
5.
6.
7.
8.
9.

5.启动 MessageQueue 负载均衡线程。

6.启动生产者线程;7.把 serviceState 改为 Running。

7.源码参考 MQClientInstance#start。

启动定时任务

这个定时任务默认每 30s 执行一次，用于监听每个 Topic 下的 MessageQueue 是否发生变化。代码见 startScheduleTask 方法。

启动轨迹消息

轨迹消息主要用于跟踪消息发送、消息消费的轨迹，用于记录详细日志。代码如下：

//AsyncTraceDispatcher 类
public void start(String nameSrvAddr, AccessChannel accessChannel) throws MQClientException {
    if (isStarted.compareAndSet(false, true)) {
        traceProducer.setNamesrvAddr(nameSrvAddr);
        traceProducer.setInstanceName(TRACE_INSTANCE_NAME + "_" + nameSrvAddr);
        traceProducer.start();
    }
    this.accessChannel = accessChannel;
    this.worker = new Thread(new AsyncRunnable(), "MQ-AsyncTraceDispatcher-Thread-" + dispatcherId);
    this.worker.setDaemon(true);
    this.worker.start();
    this.registerShutDownHook();
}1.
2.
3.
4.
5.
6.
7.
8.
9.
10.
11.
12.
13.

这里不详细展开了，后面再单独讨论。

总结

本文通过源码分析讲解了 RocketMQ 中 PULL 模式下的消费者启动过程，在生产上使用比较多的还是 PUSH 模式，PULL 模式拉取消息的方法跟 PUSH 模式一样，不同的是 PULL 模式需要应用程序进行拉取动作，可以通过 PULL 模式的学习更容易的理解 PUSH 模式。最后，分析一个 PULL 模式启动过程涉及的 UML 类图：