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

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();
    }
}

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

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);
    }
}

//RebalanceService 类
public void run() {
    while (!this.isStopped()) {
        //waitInterval 默认 20s，可以配置
        this.waitForRunning(waitInterval);
        this.mqClientFactory.doRebalance();
    }
}

//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);
}

//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");
}

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);
    }
}

//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();
}