RocketMQ源码15- consumer启动流程

落日映苍穹つ 2023-10-08 23:50 70阅读 0赞

> 本系列[RocketMQ4.8注释github地址][RocketMQ4.8_github]，希望对大家有所帮助，要是觉得可以的话麻烦给点一下Star哈

`org.apache.rocketmq.example.quickstart.Consumer`:

public class Consumer {
    
        public static void main(String[] args) throws InterruptedException, MQClientException {
    
            // todo
            DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("please_rename_unique_group_name_4");
    
            consumer.setNamesrvAddr("127.0.0.1:9876");
           
            consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
    
            consumer.subscribe("TopicTest", "*");
    
            // 注册监听器，监听消息
            consumer.registerMessageListener(new MessageListenerConcurrently() {
    
                @Override
                public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
                    ConsumeConcurrentlyContext context) {
                    // 这里获得了消息
                    System.out.printf("%s Receive New Messages: %s %n", Thread.currentThread().getName(), msgs);
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
            });
            // todo 启动
            consumer.start();
    
            System.out.printf("Consumer Started.%n");
        }
    }
    复制代码

`consumer`使用起来还是挺简单的，先是创建了一个`DefaultMQPushConsumer`对象，然后配置了一些属性，比较关键的就是注册消息监听器（在这个监听器里会获取消息），之后就调用`start()`方法启动`consumer`.

接下来我们就来分析这块的消费过程。

## 1. 构造方法：`DefaultMQPushConsumer` ##

`consumer`的处理类为`DefaultMQPushConsumer`，我们先来看看`DefaultMQPushConsumer`的属性值和构造方法：

public class DefaultMQPushConsumer extends ClientConfig implements MQPushConsumer {
    
        private final InternalLogger log = ClientLogger.getLog();
    
        protected final transient DefaultMQPushConsumerImpl defaultMQPushConsumerImpl;
    
        // 消费者所属组
        private String consumerGroup;
        // 消息消费模式，分为集群模式、广播模式，默认为集群模式
        private MessageModel messageModel = MessageModel.CLUSTERING;
        // 第一次消费时指定消费策略
        private ConsumeFromWhere consumeFromWhere = ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET;
    
        private String consumeTimestamp = UtilAll.timeMillisToHumanString3(System.currentTimeMillis() - (1000 * 60 * 30));
    
        // 集群模式下消息队列的负载策略
        private AllocateMessageQueueStrategy allocateMessageQueueStrategy;
    
        /**
         * Subscription relationship 订阅信息
         */
        private Map<String /* topic */, String /* sub expression */> subscription = new HashMap<String, String>();
    
        /**
         * Message listener 消息业务监听器
         */
        private MessageListener messageListener;
    
        /**
         * Offset Storage 消息消费进度存储器
         */
        private OffsetStore offsetStore;
    
        /**
         * Minimum consumer thread number 消费者最小线程数
         */
        private int consumeThreadMin = 20;
    
        /**
         * Max consumer thread number  消费者最大线程数，因为消费
         * 者线程池使用无界队列，所以此参数不生效
         */
        private int consumeThreadMax = 20;
    
        /**
         * Threshold for dynamic adjustment of the number of thread pool
         */
        private long adjustThreadPoolNumsThreshold = 100000;
    
        /**
         * Concurrently max span offset.it has no effect on sequential consumption
         * 并发消息消费时处理队列最大跨度，默认2000，表示如果消息处理队列中偏移量最大的消息
         * 与偏移量最小的消息的跨度超过2000，则延迟50ms后再拉取消息
         */
        private int consumeConcurrentlyMaxSpan = 2000;
    
        /**
         * Flow control threshold on queue level, each message queue will cache at most 1000 messages by default,
         * Consider the {@code pullBatchSize}, the instantaneous value may exceed the limit
         * 默认1000，表示每1000
         * 次流控后打印流控日志
         */
        private int pullThresholdForQueue = 1000;
        private int pullThresholdSizeForTopic = -1;
    
        /**
         * Message pull Interval
         * 推模式下拉取任务的间隔时间，默
         * 认一次拉取任务完成后继续拉取
         */
        private long pullInterval = 0;
    
        /**
         * Batch consumption size 消息并发消费时一次
         * 消费消息的条数，通俗点说，就是每次传入
         * MessageListener#consumeMessage中的消息条数
         */
        private int consumeMessageBatchMaxSize = 1;
    
        /**
         * Batch pull size 每次消息拉取的条数，默认32条
         */
        private int pullBatchSize = 32;
    
        /**
         * Whether update subscription relationship when every pull
         * 是否每次拉取消息都更
         * 新订阅信息，默认为false
         */
        private boolean postSubscriptionWhenPull = false;
    
        private boolean unitMode = false;
        /**
         *
         * 最大消费重试次数。如果消息消费次数超过maxReconsumeTimes还未成功，则将该消息转移到一个失败
         * 队列，等待被删除
         */
        private int maxReconsumeTimes = -1;
    
        /**
         * 延迟将该队列的消
         * 息提交到消费者线程的等待时间，默认延迟1s
         */
        private long suspendCurrentQueueTimeMillis = 1000;
    
        /**
         * Maximum amount of time in minutes a message may block the consuming thread.
         * 消息消费超时时间，默认为15，
         * 单位为分钟
         */
        private long consumeTimeout = 15;
        ...
        
        public DefaultMQPushConsumer(final String consumerGroup) {
            // 指定了队列分配策略 AllocateMessageQueueAveragely
            this(null, consumerGroup, null, new AllocateMessageQueueAveragely());
        }
    
        ...
        public DefaultMQPushConsumer(final String namespace, final String consumerGroup, RPCHook rpcHook,
            AllocateMessageQueueStrategy allocateMessageQueueStrategy) {
            this.consumerGroup = consumerGroup;
            this.namespace = namespace;
            this.allocateMessageQueueStrategy = allocateMessageQueueStrategy;
            defaultMQPushConsumerImpl = new DefaultMQPushConsumerImpl(this, rpcHook);
        }
    }
    复制代码

在构造方法中，就只是做了一些成员变量的赋值操作，比较关键的是分配消息队列的策略：`allocateMessageQueueStrategy`，如果指定，默认就使用`AllocateMessageQueueAveragely`，即从各队列平均获取消息。

## 2.客户端start()方法都干了什么？ ##

首先，暴露给开发者的 `DefaultMQPushConsumer`是一个外观类，真正工作的是其内部的`DefaultMQPushConsumerImpl`,所以我们看下`DefaultMQPushConsumerImpl#start()`的逻辑

public synchronized void start() throws MQClientException {
        switch (this.serviceState) {
            case CREATE_JUST:
                log.info("the consumer [{}] start beginning. messageModel={}, isUnitMode={}", this.defaultMQPushConsumer.getConsumerGroup(),
                    this.defaultMQPushConsumer.getMessageModel(), this.defaultMQPushConsumer.isUnitMode());
                this.serviceState = ServiceState.START_FAILED;
    
                //TODO:检查配置，比如消费者组是否为空，消费模式是否为空，订阅信息是否为空等等
                this.checkConfig();
    
                // TODO 构建主题订阅信息SubscriptionData并加入RebalanceImpl的订阅消息中
                this.copySubscription();
    
                if (this.defaultMQPushConsumer.getMessageModel() == MessageModel.CLUSTERING) {
                    this.defaultMQPushConsumer.changeInstanceNameToPID();
                }
    
                // todo 初始化MQClientInstance、RebalanceImpl（消息重新负载实现类）等
                this.mQClientFactory = MQClientManager.getInstance().getOrCreateMQClientInstance(this.defaultMQPushConsumer, this.rpcHook);
    
                // 设置负载均衡相关属性
                this.rebalanceImpl.setConsumerGroup(this.defaultMQPushConsumer.getConsumerGroup());
                this.rebalanceImpl.setMessageModel(this.defaultMQPushConsumer.getMessageModel());
                this.rebalanceImpl.setAllocateMessageQueueStrategy(this.defaultMQPushConsumer.getAllocateMessageQueueStrategy());
                this.rebalanceImpl.setmQClientFactory(this.mQClientFactory);
    
                //TODO: 创建拉取对象的核心类，后面去broker拉取消息时会看到
                this.pullAPIWrapper = new PullAPIWrapper(
                    mQClientFactory,
                    this.defaultMQPushConsumer.getConsumerGroup(), isUnitMode());
                this.pullAPIWrapper.registerFilterMessageHook(filterMessageHookList);
    
                // todo 初始化消息进度。如果消息消费采用集群模式，那么消
                //息进度存储在Broker上，如果采用广播模式，那么消息消费进度存储
                //在消费端
                if (this.defaultMQPushConsumer.getOffsetStore() != null) {
                    this.offsetStore = this.defaultMQPushConsumer.getOffsetStore();
                } else {
                    switch (this.defaultMQPushConsumer.getMessageModel()) {
                        //TODO: 广播模式创建LocalFileOffsetStore，保存offset到本地文件中
                        case BROADCASTING:
                            this.offsetStore = new LocalFileOffsetStore(this.mQClientFactory, this.defaultMQPushConsumer.getConsumerGroup());
                            break;
                        //TODO:集群模式创建 RemoteBrokerOffsetStore，保存offset到broker文件中
                        case CLUSTERING:
                            this.offsetStore = new RemoteBrokerOffsetStore(this.mQClientFactory, this.defaultMQPushConsumer.getConsumerGroup());
                            break;
                        default:
                            break;
                    }
                    this.defaultMQPushConsumer.setOffsetStore(this.offsetStore);
                }
                // 加载消费信息的偏移量
                this.offsetStore.load();
    
                // todo 如果是顺序消费，创建消费端消费线程服务。
                //ConsumeMessageService主要负责消息消费，在内部维护一个线程池
                if (this.getMessageListenerInner() instanceof MessageListenerOrderly) {
                    this.consumeOrderly = true;
                    //TODO:创建顺序消费消息服务类
                    this.consumeMessageService =
                        new ConsumeMessageOrderlyService(this, (MessageListenerOrderly) this.getMessageListenerInner());
                } else if (this.getMessageListenerInner() instanceof MessageListenerConcurrently) {
                    this.consumeOrderly = false;
                    //TODO:创建其他消费消息服务类，其内部维护了一个线程池，后面会用到
                    this.consumeMessageService =
                        new ConsumeMessageConcurrentlyService(this, (MessageListenerConcurrently) this.getMessageListenerInner());
                }
    
                this.consumeMessageService.start();
    
                // todo 向MQClientInstance注册消费者并启动MQClientInstance，JVM中的所有消费者、生产者持有同一个MQClientInstance，MQClientInstance只会启动一次
                boolean registerOK = mQClientFactory.registerConsumer(this.defaultMQPushConsumer.getConsumerGroup(), this);
                if (!registerOK) {
                    this.serviceState = ServiceState.CREATE_JUST;
                    this.consumeMessageService.shutdown(defaultMQPushConsumer.getAwaitTerminationMillisWhenShutdown());
                    throw new MQClientException("The consumer group[" + this.defaultMQPushConsumer.getConsumerGroup()
                        + "] has been created before, specify another name please." + FAQUrl.suggestTodo(FAQUrl.GROUP_NAME_DUPLICATE_URL),
                        null);
                }
    
                // todo 启动客户端实例，这个方法非常重要，内部做了很多事情
                mQClientFactory.start();
                log.info("the consumer [{}] start OK.", this.defaultMQPushConsumer.getConsumerGroup());
                this.serviceState = ServiceState.RUNNING;
                break;
            case RUNNING:
            case START_FAILED:
            case SHUTDOWN_ALREADY:
                throw new MQClientException("The PushConsumer service state not OK, maybe started once, "
                    + this.serviceState
                    + FAQUrl.suggestTodo(FAQUrl.CLIENT_SERVICE_NOT_OK),
                    null);
            default:
                break;
        }
    
        // 更新 topic 的信息，从nameServer获取数据
        this.updateTopicSubscribeInfoWhenSubscriptionChanged();
        this.mQClientFactory.checkClientInBroker();
        // todo 发送心跳，发送到所有的broker
        this.mQClientFactory.sendHeartbeatToAllBrokerWithLock();
        // todo 负载均衡 立即重平衡
        this.mQClientFactory.rebalanceImmediately();
    }
    复制代码

### 2.1 检查配置 ###

1.  校验`GroupName`是否为空;校验`GroupName`是否等于`DEFAULT_CONSUMER`(等于的话直接抛出异常)
2.  校验消费模式:集群/广播
3.  校验ConsumeFromWhere
4.  校验开始消费的指定时间
5.  校验AllocateMessageQueueStrategy
6.  校验订阅关系
7.  校验是否注册消息监听
8.  校验消费线程数,`consumeThreadMin`和 `consumeThreadMax` 默认值都是20，取值区间都是 \[1, 1000\]
9.  校验本地队列缓存消息的最大数，默认是1000，取值范围是\[1, 1024\], 主要是做流控用的
10. 校验拉取消息的时间间隔，`pullInterval`参数，默认是不存在间隔，取值范围是\[0, 65535\]。当消费速度比生产速度快，可以设置这个参数，避免花费大概率从broker拉取空消息
11. 校验单次拉取的最大消息数，`consumeMessageBatchMaxSize` 参数，默认是1，取值范围是\[1, 1024\]
12. 校验单次消费的最大消息数, `pullBatchSize` 参数，默认是32，取值范围是\[1, 1024\]。

### 2.2 拷贝订阅关系 ###

将订阅关系设置到重平衡服务类`RebalanceImpl`中，订阅`重试主题`消息：

//TODO:key=topic, value=订阅数据（就是tag信息）
    protected final ConcurrentMap<String /* topic */, SubscriptionData> subscriptionInner =
        new ConcurrentHashMap<String, SubscriptionData>();
    复制代码

### 2.3 创建客户端实例 `MQClientInstance` ###

// todo 初始化MQClientInstance、RebalanceImpl（消息重新负载实现类）等
    this.mQClientFactory = MQClientManager.getInstance().getOrCreateMQClientInstance(this.defaultMQPushConsumer, this.rpcHook);
    复制代码

> 这个对象在前面分析Producer启动时也看到了，因为`Prodcuer/Consumer` 都是客户端，所以都会根据这个实例来创建对象。 我们看下这个类的结构：

public class MQClientInstance {
        
            //TODO:生产者表，producer启动时创建一个新的MQClientInstance实例对象，将生产者信息注册到这里。生产者实例对象中消费者信息是空
            private final ConcurrentMap<String/* group */, MQProducerInner> producerTable = new ConcurrentHashMap<String, MQProducerInner>();
    
            //TODO:消费者表，consumer启动时创建一个新的MQClientInstance实例对象，将消费者信息注册到这里。消费者实例对象中生产者信息是空
            private final ConcurrentMap<String/* group */, MQConsumerInner> consumerTable = new ConcurrentHashMap<String, MQConsumerInner>();
    
            //TODO:topic路由信息，producer和consumer都会使用
            private final ConcurrentMap<String/* Topic */, TopicRouteData> topicRouteTable = new ConcurrentHashMap<String, TopicRouteData>();
    
            //TODO:broker信息，producer和consumer都会用到
            private final ConcurrentMap<String/* Broker Name */, HashMap<Long/* brokerId */, String/* address */>> brokerAddrTable =
                new ConcurrentHashMap<String, HashMap<Long, String>>();
            
            //TODO......
            
            
           /**
            * TODO:构造器
            */
           public MQClientInstance(ClientConfig clientConfig, int instanceIndex, String clientId, RPCHook rpcHook) {
            //TODO:客户端处理器,比如在集群消费模式下，有新的消费者加入，则通知消费者客户端重平衡,是给消费者用的(分析生产者时，我们直接忽略了它）
            this.clientRemotingProcessor = new ClientRemotingProcessor(this);
            
            //TODO:它的内部会创建netty客户端对象（NettyRemotingClient），用于和broker通信
            this.mQClientAPIImpl = new MQClientAPIImpl(this.nettyClientConfig, this.clientRemotingProcessor, rpcHook, clientConfig);
    
            //TODO.......
    
            //TODO:拉取消息的服务，和消费者相关
            this.pullMessageService = new PullMessageService(this);
    
            //TODO:重平衡服务，和消费者相关
            this.rebalanceService = new RebalanceService(this);
            
            //TODO:other......
        }
    }        
    复制代码

所谓的"客户端"，实际上是在`MQClientAPIImpl`对象的内部的`NettyRemotingClient`：

public class MQClientAPIImpl {
        //TODO:.....
        
        private final RemotingClient remotingClient;
        private final TopAddressing topAddressing;
        private final ClientRemotingProcessor clientRemotingProcessor;
        private String nameSrvAddr = null;
        private ClientConfig clientConfig;
    
        public MQClientAPIImpl(final NettyClientConfig nettyClientConfig,
            final ClientRemotingProcessor clientRemotingProcessor,
            RPCHook rpcHook, final ClientConfig clientConfig) {
            this.clientConfig = clientConfig;
            topAddressing = new TopAddressing(MixAll.getWSAddr(), clientConfig.getUnitName());
             
            //TODO:所谓的客户端实际上就是这个，netty客户端 
            this.remotingClient = new NettyRemotingClient(nettyClientConfig, null);
            this.clientRemotingProcessor = clientRemotingProcessor;
    
            this.remotingClient.registerRPCHook(rpcHook);
           
            //TODO:注册处理器
            this.remotingClient.registerProcessor(RequestCode.CHECK_TRANSACTION_STATE, this.clientRemotingProcessor, null);
            this.remotingClient.registerProcessor(RequestCode.NOTIFY_CONSUMER_IDS_CHANGED, this.clientRemotingProcessor, null);
    
         //TODO:....注册其他处理器......
        }
    复制代码

这里我们关注下 `NettyRemotingClient`，其实前面分析Broker,Producer时，我们经常看到这个，因为RocketMQ是使用`netty`作为通信的。

> `NettyRemotingClient`：netty的客户端对象  
> `NettyRemotingServer`：netty的服务端对象

### 2.4 创建拉取消息的核心类 `PullAPIWrapper` ###

this.pullAPIWrapper = new PullAPIWrapper(
        mQClientFactory,
        this.defaultMQPushConsumer.getConsumerGroup(), isUnitMode());
    复制代码

### 2.5 创建 offset存储服务 `OffsetStore` ###

*  如果是广播模式(`BROADCASTING`), 则创建 `LocalFileOffsetStore` 对象，将消费者的offset存储到本地的，默认文件路径为当前用户主目录下的 `.rocketmq_offsets/clientId/{clientId}/clientId/{group}/Offsets.json`。其中clientId为当前消费者id，默认为`ip@default`,\{clientId\}为当前消费者id，默认为ip@default, clientId为当前消费者id，默认为ip@default,\{group\}为消费者组名称
 *  如果是集群模式(`CLUSTERING`), 则创建 `RemoteBrokerOffsetStore`对象，将消费者的offset存储到broker中，文件路径为当前用户主目录下的`store/config/consumerOffset.json`

### 2.6 创建消费消息的服务类 `ConsumeMessageService` ###

*  如果是顺序消费，则创建 `ConsumeMessageOrderlyService` 对象
 *  如果是其他消费，则创建 `ConsumeMessageConcurrentlyService` 对象，同时内部也会创建一个`ThreadPoolExecutor`线程池，这个线程池非常的重要，拉取到消息后会将消息提交到这个线程池中给消费者消费

### 2.7 将consumer注册到本地 ###

boolean registerOK = mQClientFactory.registerConsumer(this.defaultMQPushConsumer.getConsumerGroup(), this);
    复制代码

将消费者组信息添加到本地客户端实例`MQClientInstance`的 `consumerTable` Map中，key=groupName; value=`DefaultMQPushConsumerImpl`,就是消费者对象

### 2.8 启动客户端实例（重要） ###

mQClientFactory.start();
    复制代码

其内部做了很多事情,主要如下： org.apache.rocketmq.client.impl.factory.MQClientInstance\#start：

public void start() throws MQClientException {
    
        synchronized (this) {
            switch (this.serviceState) {
                case CREATE_JUST:
                    // 先设置 失败
                    this.serviceState = ServiceState.START_FAILED;
                    // If not specified,looking address from name server
                    // 判断namesrv 是否为null
                    if (null == this.clientConfig.getNamesrvAddr()) {
                        this.mQClientAPIImpl.fetchNameServerAddr();
                    }
                    // todo Start request-response channel
                    // 启动远程服务，这个方法只是装配了netty客户端相关配置
                    // todo 注意：1. 这里是netty客户端，2. 这里并没有创建连接
                    this.mQClientAPIImpl.start();
                    // Start various schedule tasks
                    // todo 开启任务调度
                    this.startScheduledTask();
                    // Start pull service
                    // todo 开启 拉取服务 仅对consumer启作用
                    this.pullMessageService.start();
                    // Start rebalance service
                    // todo 开启平衡服务 仅对consumer启作用
                    this.rebalanceService.start();
                    // Start push service
                    // todo 启用内部的 producer
                    this.defaultMQProducer.getDefaultMQProducerImpl().start(false);
                    log.info("the client factory [{}] start OK", this.clientId);
                    // 设置状态为running
                    this.serviceState = ServiceState.RUNNING;
                    break;
                case START_FAILED:
                    throw new MQClientException("The Factory object[" + this.getClientId() + "] has been created before, and failed.", null);
                default:
                    break;
            }
        }
    }
    复制代码

在`producer`的启动过程中，也会调用这个方法，前面我们已经分析过了一波了，这次我们在`consumer`的角度再来分析这个方法。

该方法所做的工作如下：

1.  获取 `nameServer` 的地址
2.  启动客户端的远程服务，这个方法会配置`netty`客户端
3.  启动定时任务
4.  启动拉取消息服务
5.  启动负载均衡服务

上面的`1`，`2`与`producer`的流程并无区别，就不再分析了，我们来看看接下来的任务

#### 2.8.1 启动远程netty客户端 ####

// Start request-response channel
    this.mQClientAPIImpl.start();
    复制代码
    复制代码

就是启动其内部的 `NettyRemotingClient`,用于和broker通信。

#### 2.8.2 启动各种定时任务 ####

// Start various schedule tasks
    this.startScheduledTask();
    复制代码

那么有哪些定时任务呢？继续往里看,罗列几个特别关注的

2.8.2.1 发送心跳到Broker

this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
    
        @Override
        public void run() {
            try {
                // todo 清除下线broker
                MQClientInstance.this.cleanOfflineBroker();
                // todo 发送心跳到所有broker上面
                MQClientInstance.this.sendHeartbeatToAllBrokerWithLock();
            } catch (Exception e) {
                log.error("ScheduledTask sendHeartbeatToAllBroker exception", e);
            }
        }
    }, 1000, this.clientConfig.getHeartbeatBrokerInterval(), TimeUnit.MILLISECONDS);
    复制代码

> 延迟1s执行，每隔30s发送一次心跳包

2.8.2.2 持久化消费者的 offset

// // 持久化消费者的消费偏移量，每5秒一次
    this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
    
        @Override
        public void run() {
            try {
                // todo 持久化consumer offset 可以放在本地文件，也可以推送到 broker
                MQClientInstance.this.persistAllConsumerOffset();
            } catch (Exception e) {
                log.error("ScheduledTask persistAllConsumerOffset exception", e);
            }
        }
    }, 1000 * 10, this.clientConfig.getPersistConsumerOffsetInterval(), TimeUnit.MILLISECONDS);
    复制代码

> 延迟10s执行，每隔5s持久化一次offset  
> 这里的持久化是将本地Map中offset发送到broker中，然后broker中的定时任务写到文件中，完成真正的持久化，后面会看到。

#### 2.8.3 启动拉取消息的服务`PullMessageService`(重要） ####

// Start pull service
    this.pullMessageService.start();
    复制代码

他是一个异步线程，其核心逻辑是

public void run() {
        log.info(this.getServiceName() + " service started");
    
        // Stopped 声明为volatile，每执行一次业务逻辑，检测一下其运行状态，可以
        //通过其他线程将Stopped设置为true，从而停止该线程
        while (!this.isStopped()) {
            try {
                // 从pullRequestQueue中获取一个PullRequest消息拉取任务，
                //如果pullRequestQueue为空，则线程将阻塞，直到有拉取任务被放入
                PullRequest pullRequest = this.pullRequestQueue.take();
                // todo 调用pullMessage方法进行消息拉取
                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`对象，去拉取消息，这个逻辑后面再说。

> 刚开始肯定是阻塞的，我们要看什么时候往队列中放入值，以及放入值之后做什么

#### 2.8.4 启动重平衡服务 `RebalanceService` ####

// Start rebalance service
    this.rebalanceService.start();
    复制代码
    复制代码

它也是一个异步线程，其核心逻辑是

public void run() {
        log.info(this.getServiceName() + " service started");
    
    
        while (!this.isStopped()) {
            // todo 等待20s执行一次 内部使用了juc的CountDownLatch, 使得这里启动后仍然是阻塞的
            this.waitForRunning(waitInterval);
            // todo
            this.mqClientFactory.doRebalance();
        }
    
        log.info(this.getServiceName() + " service end");
    }
    复制代码

他是重平衡的核心逻辑，但是在启动时，由于使用了JUC的 `CountDownLatch`锁，使其不会立即重平衡，而是阻塞，什么时候出发重平衡呢？我们还是继续往后看

### 2.9 发送心跳到broker ###

this.updateTopicSubscribeInfoWhenSubscriptionChanged();
    this.mQClientFactory.checkClientInBroker();
    //TODO: 发送心跳到broker
    this.mQClientFactory.sendHeartbeatToAllBrokerWithLock();
    this.mQClientFactory.rebalanceImmediately();
    复制代码

是不是有些疑惑？前面的定时任务中不是已经启动了心跳服务吗，为什么这里还要启动呢？我也不清楚，猜测是因为担心网络问题导致没有及时发送给broker吧

### 2.10 立即启动重平衡服务`RebalanceService` ###

//TODO:立即启动重平衡服务
    this.mQClientFactory.rebalanceImmediately();
    复制代码

前面在 **2.8.4**的时候，启动了重平衡服务，但是因为 `CountDownLatch` 导致阻塞了，这里就是唤醒，可以执行重平衡的逻辑。

> 这里先不关注它的内部逻辑,后面在分析。

其实，到这里`Consumer`客户端启动流程就结束了，至于如何拉取消息，也是这些服务类的相互配合工作的，后面我们在分析中在仔细分析它是如何一步一步拉取消息并消费的。

## 3. 总结 ##

消费者启动过程相比生产者启动过程要复杂一些，会启动很多对象（实际上生产者也启动了，因为他们都是基于客户端实例`MQClientInstance`去创建的对象，只不过生产者并不会使用某些服务类）

简单总结下：

1.  创建客户端实例`MQClientInstance`（其内部要创建netty客户端对象`NettyRemotingClient`）
2.  启动客户端实例，其内部要启动netty客户端，消息拉取服务，以及重平衡服务，还有各种定时任务
3.  发送心跳到broker

好了，Consumer启动过程就分析到这里，接下来就分析下消费过程

[RocketMQ4.8_github]: https://link.juejin.cn/?target=https%3A%2F%2Fgithub.com%2Fhsfxuebao%2Frocketmq-all-4.8.0-source