深圳幻海软件技术有限公司 欢迎您!

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

2023-02-28

大家好,我是君哥。今天来分享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 方法进行了装饰,这个装饰类主要增加了下面功能:

  1. 获取 Broker 地址。
  2. 检查 RocketMQ 版本。
  3. 如果 Broker 是从节点,把 sysFlag 标记偏移量的位改为 0,(偏移量 0x1)。
  4. 封装请求 header。
  5. 获取 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 客户端主要包括如下步骤:

  1. 把 serviceState 改为 START_FAIL ED。
  2. 初始化 Netty channel。
  3. 启动定时任务,包括定时获取 Name Server 地址、从 Name Server 更新 Topic 路由信息、清理过期的 Broker、向 Broker 发送心跳、持久化 offset、定时调整线程池的数量(源码里面这个并没有实现逻辑)。
  4. 启动拉取消息的线程,拉取线程的逻辑是从请求队列中不停地取出 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 类图: