rocketmq源码剖析之producer和consumer的启动关闭

浅浅的花香味﹌ 2023-06-09 10:05 8阅读 0赞

`producer`在`rocketmq`的作用是消息的生产者，`consumer`在`rocketmq`的作用是消息的消费者，它的生命周期是跟项目相关的，即是由使用者控制的。而为什么要将这两个角色的启动关闭流程放在一起剖析呢？是因为他们都是MQ的客户端，在启动和关闭的行为上，有着很多共同的地方。接下来便将会来仔细探究其启动和关闭的过程。

## Producer ##

### DefaultMQProducer ###

#### DefaultMQProducer的启动 ####

`producer`在启动的时候会做一系列的内部初始化，其启动的源码如下所示：

public void start() throws MQClientException { 
            this.setProducerGroup(withNamespace(this.producerGroup));
            this.defaultMQProducerImpl.start();
            if (null != traceDispatcher) { 
                try { 
                    traceDispatcher.start(this.getNamesrvAddr(), this.getAccessChannel());
                } catch (MQClientException e) { 
                    log.warn("trace dispatcher start failed ", e);
                }
            }
        }

从简短的代码可以看见其做的事情是对`producer`的`group`的设置、核心初始化、`trace`的初始化。接下来便仔细说明下核心初始化过程，其余是其和`consumer`所共有的，会在公共行为部分进行剖析。

##### `producer`核心初始化 #####

`producer`启动初始化化部分核心逻辑如下，第一次启动的时候它会执行`CREATE_JUST`分支：

public void start(final boolean startFactory) throws MQClientException { 
            switch (this.serviceState) { 
                case CREATE_JUST:
                    this.serviceState = ServiceState.START_FAILED;
                    // 校验group参数
                    this.checkConfig();
    
                    if (!this.defaultMQProducer.getProducerGroup().equals(MixAll.CLIENT_INNER_PRODUCER_GROUP)) { 
                        // 设置producer的实例名
                        this.defaultMQProducer.changeInstanceNameToPID();
                    }
                    // 创建MQClient实例
                    this.mQClientFactory = MQClientManager.getInstance().getAndCreateMQClientInstance(this.defaultMQProducer, rpcHook);
                    // 向MQ客户端的producer表中注册
                    boolean registerOK = mQClientFactory.registerProducer(this.defaultMQProducer.getProducerGroup(), this);
                    if (!registerOK) { 
                        this.serviceState = ServiceState.CREATE_JUST;
                        throw new MQClientException("The producer group[" + this.defaultMQProducer.getProducerGroup()
                            + "] has been created before, specify another name please." + FAQUrl.suggestTodo(FAQUrl.GROUP_NAME_DUPLICATE_URL),
                            null);
                    }
                    // 向当前的producer的topic表中注册一个用于测试或demo用的topic “TBW102”
                    this.topicPublishInfoTable.put(this.defaultMQProducer.getCreateTopicKey(), new TopicPublishInfo());
    
                    if (startFactory) { 
                        // 启动MQClient
                        mQClientFactory.start();
                    }
    
                    log.info("the producer [{}] start OK. sendMessageWithVIPChannel={}", this.defaultMQProducer.getProducerGroup(),
                        this.defaultMQProducer.isSendMessageWithVIPChannel());
                    this.serviceState = ServiceState.RUNNING;
                    break;
                case RUNNING:
                case START_FAILED:
                case SHUTDOWN_ALREADY:
                    throw new MQClientException("The producer service state not OK, maybe started once, "
                        + this.serviceState
                        + FAQUrl.suggestTodo(FAQUrl.CLIENT_SERVICE_NOT_OK),
                        null);
                default:
                    break;
            }
            // broker的心跳机制发送
            this.mQClientFactory.sendHeartbeatToAllBrokerWithLock();
        }

浏览过这段代码，便可以很容易看出其中的核心逻辑在于`MQClient`的启动，以及`producer`与`broker`之间的心跳触发，这两部分是与`consumer`所共有的，会在公共行为部分进行剖析。

#### DefaultMQProducer的关闭 ####

`producer`的关闭行为是`producer`从`RUNNING`状态到`SHUTDOWN_ALREADY`状态的过程，主要做了三件事：

*  从注册表中清除了本身实例；
 *  关闭了线程池资源；
 *  清楚了client的资源；

核心代码如下所示：

this.mQClientFactory.unregisterProducer(this.defaultMQProducer.getProducerGroup());
    this.defaultAsyncSenderExecutor.shutdown();
    if (shutdownFactory) { 
        this.mQClientFactory.shutdown();
    }

### TransactionMQProducer ###

#### TransactionMQProducer的启动 ####

事务生产者`TransactionMQProducer`是默认的生产者的一个增强，它在启动方面多了一步操作：

public void start() throws MQClientException { 
            this.defaultMQProducerImpl.initTransactionEnv();
            super.start();
        }

可以看出多的这一步则是在事务方面的一个初始化，即事务检查的线程池初始化，其源码如下所示：

public void initTransactionEnv() { 
            TransactionMQProducer producer = (TransactionMQProducer) this.defaultMQProducer;
            if (producer.getExecutorService() != null) { 
                this.checkExecutor = producer.getExecutorService();
            } else { 
                this.checkRequestQueue = new LinkedBlockingQueue<Runnable>(producer.getCheckRequestHoldMax());
                this.checkExecutor = new ThreadPoolExecutor(
                    producer.getCheckThreadPoolMinSize(),
                    producer.getCheckThreadPoolMaxSize(),
                    1000 * 60,
                    TimeUnit.MILLISECONDS,
                    this.checkRequestQueue);
            }
        }

#### TransactionMQProducer的关闭 ####

同样，事务生产者在关闭的时候也要多一步关闭事务事务检查线程池资源的操作，具体就不在详细说明。

## Consumer ##

### PushConsumer ###

#### PushConsumer的启动 ####

`push`模式的消费者同生产者一样，在启动的时候依然是先包装`group`，然后在执行核心流程，最后在根据情况初始化`trace`。其源码如下：

public void start() throws MQClientException { 
            setConsumerGroup(NamespaceUtil.wrapNamespace(this.getNamespace(), this.consumerGroup));
            this.defaultMQPushConsumerImpl.start();
            if (null != traceDispatcher) { 
                try { 
                    traceDispatcher.start(this.getNamesrvAddr(), this.getAccessChannel());
                } catch (MQClientException e) { 
                    log.warn("trace dispatcher start failed ", e);
                }
            }
        }

接下来便仔细说明下核心初始化过程，其余是其和`producer`所共有的，会在公共行为部分进行剖析。

##### `PushConsumer`核心初始化 #####

这里直接附上核心的`start`过程源码。其一样的走`CREATE_JUST`分支，如下所示：

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;
                    // 校验group参数 
                    this.checkConfig();
                    // 进行拷贝注册（拷贝topic，监听器等）
                    this.copySubscription();
    
                    if (this.defaultMQPushConsumer.getMessageModel() == MessageModel.CLUSTERING) { 
                        // 如果是集群模式则根据PID改变实例名
                        this.defaultMQPushConsumer.changeInstanceNameToPID();
                    }
                    // 创建MQClient实例
                    this.mQClientFactory = MQClientManager.getInstance().getAndCreateMQClientInstance(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);
    
                    // pull-api包装
                    this.pullAPIWrapper = new PullAPIWrapper(
                        mQClientFactory,
                        this.defaultMQPushConsumer.getConsumerGroup(), isUnitMode());
                    // 注册消息过滤钩子
                    this.pullAPIWrapper.registerFilterMessageHook(filterMessageHookList);
    
                    // 复制offset-store
                    if (this.defaultMQPushConsumer.getOffsetStore() != null) { 
                        this.offsetStore = this.defaultMQPushConsumer.getOffsetStore();
                    } else { 
                        switch (this.defaultMQPushConsumer.getMessageModel()) { 
                            case BROADCASTING:
                                // 广播模式设置为本地文件存储模式
                                this.offsetStore = new LocalFileOffsetStore(this.mQClientFactory, this.defaultMQPushConsumer.getConsumerGroup());
                                break;
                            case CLUSTERING:
                                // 集群模式将偏移两设置为远程broker存储
                                this.offsetStore = new RemoteBrokerOffsetStore(this.mQClientFactory, this.defaultMQPushConsumer.getConsumerGroup());
                                break;
                            default:
                                break;
                        }
                        this.defaultMQPushConsumer.setOffsetStore(this.offsetStore);
                    }
                    // 加载消费偏移情况
                    this.offsetStore.load();
    
                    if (this.getMessageListenerInner() instanceof MessageListenerOrderly) { 
                        // 有序消费监听器
                        this.consumeOrderly = true;
                        this.consumeMessageService =
                            new ConsumeMessageOrderlyService(this, (MessageListenerOrderly) this.getMessageListenerInner());
                    } else if (this.getMessageListenerInner() instanceof MessageListenerConcurrently) { 
                        // 并发消费监听器
                        this.consumeOrderly = false;
                        this.consumeMessageService =
                            new ConsumeMessageConcurrentlyService(this, (MessageListenerConcurrently) this.getMessageListenerInner());
                    }
    
                    // 启动消费消息的服务（这里的start其实是启动了一个定时任务用于清理消费超时的消息）
                    this.consumeMessageService.start();
    
                    //向MQ客户端的consumer表中注册
                    boolean registerOK = mQClientFactory.registerConsumer(this.defaultMQPushConsumer.getConsumerGroup(), this);
                    if (!registerOK) { 
                        this.serviceState = ServiceState.CREATE_JUST;
                        this.consumeMessageService.shutdown();
                        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);
                    }
                    // mqclient启动
                    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注册数据
            this.updateTopicSubscribeInfoWhenSubscriptionChanged();
            // 检测topic所在的broker是否能够连接成功
            this.mQClientFactory.checkClientInBroker();
            // 触发想broker的心跳
            this.mQClientFactory.sendHeartbeatToAllBrokerWithLock();
            // 直接触发一次平衡操作
            this.mQClientFactory.rebalanceImmediately();
        }

这里来看一下消费偏移的load情况，消费偏移的加载分从远程`broker`和本地文件加载，远程由`RemoteBrokerOffsetStore`来实现，本地由`LocalFileOffsetStore`来实现，具体使用哪儿种取决于是集群模式还是广播模式，者同样是影响消费偏移是持久化到本地还是远程的问题。

远程的加载其实没有做任何事情，源码如下所示：

@Override
        public void load() { 
        }

从本地文件加载则是会根据是否从文件中读取到内容决定，具体源码如下：

public void load() throws MQClientException { 
            // 从本地文件读取内容到包装器
            OffsetSerializeWrapper offsetSerializeWrapper = this.readLocalOffset();
            if (offsetSerializeWrapper != null && offsetSerializeWrapper.getOffsetTable() != null) { 
                // 设置偏移量表
                offsetTable.putAll(offsetSerializeWrapper.getOffsetTable());
                // 打印日志
                for (MessageQueue mq : offsetSerializeWrapper.getOffsetTable().keySet()) { 
                    AtomicLong offset = offsetSerializeWrapper.getOffsetTable().get(mq);
                    log.info("load consumer's offset, {} {} {}",
                        this.groupName,
                        mq,
                        offset.get());
                }
            }
        }

#### PushConsumer的关闭 ####

`push`模式`consumer`在关闭的时候清理资源的情况如下所示：

public synchronized void shutdown() { 
            switch (this.serviceState) { 
                case CREATE_JUST:
                    break;
                case RUNNING:
                    // 关闭消费消息服务
                    this.consumeMessageService.shutdown();
                    // 持久化消息的消费情况
                    this.persistConsumerOffset();
                    // 从注册表移除当前的消费者
                    this.mQClientFactory.unregisterConsumer(this.defaultMQPushConsumer.getConsumerGroup());
                    // 关闭MQClient资源
                    this.mQClientFactory.shutdown();
                    log.info("the consumer [{}] shutdown OK", this.defaultMQPushConsumer.getConsumerGroup());
                    this.rebalanceImpl.destroy();
                    this.serviceState = ServiceState.SHUTDOWN_ALREADY;
                    break;
                case SHUTDOWN_ALREADY:
                    break;
                default:
                    break;
            }
        }

### PullConsumer ###

#### PullConsumer的启动 ####

`pull`模式的消费者在启动的时候则只有包装`group`和执行核心流程，其源码如下：

public void start() throws MQClientException { 
            this.setConsumerGroup(NamespaceUtil.wrapNamespace(this.getNamespace(), this.consumerGroup));
            this.defaultMQPullConsumerImpl.start();
        }

接下来便仔细说明下核心初始化过程，包装`group`部分会在公共行为部分进行剖析。

##### PullConsumer核心初始化 #####

这里直接附上核心的start过程源码，其一样的走CREATE\_JUST分支，如下所示：

public synchronized void start() throws MQClientException { 
            switch (this.serviceState) { 
                case CREATE_JUST:
                    this.serviceState = ServiceState.START_FAILED;
                    // 相关配置检查
                    this.checkConfig();
                    // 进行拷贝注册
                    this.copySubscription();
                
                    if (this.defaultMQPullConsumer.getMessageModel() == MessageModel.CLUSTERING) { 
                        // 如果是集群模式则更改实例名
                        this.defaultMQPullConsumer.changeInstanceNameToPID();
                    }
                    // 获取MQ客户端实例
                    this.mQClientFactory = MQClientManager.getInstance().getAndCreateMQClientInstance(this.defaultMQPullConsumer, this.rpcHook);
    
                    // 设置pull模式的均衡策略
                    this.rebalanceImpl.setConsumerGroup(this.defaultMQPullConsumer.getConsumerGroup());
                    this.rebalanceImpl.setMessageModel(this.defaultMQPullConsumer.getMessageModel());
                    this.rebalanceImpl.setAllocateMessageQueueStrategy(this.defaultMQPullConsumer.getAllocateMessageQueueStrategy());
                    this.rebalanceImpl.setmQClientFactory(this.mQClientFactory);
    
                    // 构造pullAPI的包装器
                    this.pullAPIWrapper = new PullAPIWrapper(
                        mQClientFactory,
                        this.defaultMQPullConsumer.getConsumerGroup(), isUnitMode());
                    this.pullAPIWrapper.registerFilterMessageHook(filterMessageHookList);
    
                    // pull模式的消息消费偏移存储
                    if (this.defaultMQPullConsumer.getOffsetStore() != null) { 
                        this.offsetStore = this.defaultMQPullConsumer.getOffsetStore();
                    } else { 
                        switch (this.defaultMQPullConsumer.getMessageModel()) { 
                            case BROADCASTING:
                                this.offsetStore = new LocalFileOffsetStore(this.mQClientFactory, this.defaultMQPullConsumer.getConsumerGroup());
                                break;
                            case CLUSTERING:
                                this.offsetStore = new RemoteBrokerOffsetStore(this.mQClientFactory, this.defaultMQPullConsumer.getConsumerGroup());
                                break;
                            default:
                                break;
                        }
                        this.defaultMQPullConsumer.setOffsetStore(this.offsetStore);
                    }
    
                    // 加载消费偏移情况
                    this.offsetStore.load();
    
                    // 注册pull消费者
                    boolean registerOK = mQClientFactory.registerConsumer(this.defaultMQPullConsumer.getConsumerGroup(), this);
                    if (!registerOK) { 
                        this.serviceState = ServiceState.CREATE_JUST;
    
                        throw new MQClientException("The consumer group[" + this.defaultMQPullConsumer.getConsumerGroup()
                            + "] has been created before, specify another name please." + FAQUrl.suggestTodo(FAQUrl.GROUP_NAME_DUPLICATE_URL),
                            null);
                    }
    
                    // MQClient的启动
                    mQClientFactory.start();
                    log.info("the consumer [{}] start OK", this.defaultMQPullConsumer.getConsumerGroup());
                    this.serviceState = ServiceState.RUNNING;
                    break;
                case RUNNING:
                case START_FAILED:
                case SHUTDOWN_ALREADY:
                    throw new MQClientException("The PullConsumer service state not OK, maybe started once, "
                        + this.serviceState
                        + FAQUrl.suggestTodo(FAQUrl.CLIENT_SERVICE_NOT_OK),
                        null);
                default:
                    break;
            }
    
        }

走览玩这段源码，可以看见，pull模式和push模式的消费者核心启动流程中，唯一的区别就是push模式下会有注册消息消费的监听器功能和最后检测连接操作，从这便可以看出，其实push模式是在pull模式上做了enhance。

#### PullConsumer的关闭 ####

`pull`模式`consumer`在关闭的时候清理资源的情况如下所示：

public synchronized void shutdown() { 
            switch (this.serviceState) { 
                case CREATE_JUST:
                    break;
                case RUNNING:
                    // 持久化消费偏移
                    this.persistConsumerOffset();
                    // 将消费者移除注册表
                    this.mQClientFactory.unregisterConsumer(this.defaultMQPullConsumer.getConsumerGroup());
                    // 关闭MQClient，清理资源
                    this.mQClientFactory.shutdown();
                    log.info("the consumer [{}] shutdown OK", this.defaultMQPullConsumer.getConsumerGroup());
                    this.serviceState = ServiceState.SHUTDOWN_ALREADY;
                    break;
                case SHUTDOWN_ALREADY:
                    break;
                default:
                    break;
            }
        }

## 公共行为 ##

### 包装设置 `group` ###

对于`group`的设置，主要是在给予的`group`参数基础上加了一层名字空间域`namespace`，而在包装的过程中首先是要获取`namespace`的值，源码如下：

public String getNamespace() { 
            if (StringUtils.isNotEmpty(namespace)) { 
                return namespace;
            }
            if (StringUtils.isNotEmpty(this.namesrvAddr)) { 
                if (NameServerAddressUtils.validateInstanceEndpoint(namesrvAddr)) { 
                    return NameServerAddressUtils.parseInstanceIdFromEndpoint(namesrvAddr);
                }
            }
            return namespace;
        }

这段源码来自`ClientConfig`，可以看见其是先尝试获取`client`中配置的值，如果值不存在的话便尝试从实例的`Endpoint`中获取，从端点中获取是通过匹正则来正则来判断的，如果匹配成功，则从`InstanceEndpoint`中截取`namespace`得值，正则表达式如下所示：

^http://MQ_INST_\w+_\w+\..*

只要没有做配置，那便获取不到值，因为不存在默认值。在执行完毕获取`namespace`值的操作后，便是真正的来包装`group`的值了。源码如下所示：

public static String wrapNamespace(String namespace, String resourceWithOutNamespace) { 
            // 如果namespace为空或本身的group值为空
            if (StringUtils.isEmpty(namespace) || StringUtils.isEmpty(resourceWithOutNamespace)) { 
                return resourceWithOutNamespace;
            }
    
            // 如果是系统的group或者已经包装过namespace
            if (isSystemResource(resourceWithOutNamespace) || isAlreadyWithNamespace(resourceWithOutNamespace, namespace)) { 
                return resourceWithOutNamespace;
            }
    
            String resourceWithoutRetryAndDLQ = withOutRetryAndDLQ(resourceWithOutNamespace);
            StringBuffer strBuffer = new StringBuffer();
    
            // RetryTopic
            if (isRetryTopic(resourceWithOutNamespace)) { 
                strBuffer.append(MixAll.RETRY_GROUP_TOPIC_PREFIX);
            }
    
            // DLQTopic
            if (isDLQTopic(resourceWithOutNamespace)) { 
                strBuffer.append(MixAll.DLQ_GROUP_TOPIC_PREFIX);
            }
    
            return strBuffer.append(namespace).append(NAMESPACE_SEPARATOR).append(resourceWithoutRetryAndDLQ).toString();
    
        }

### 消息轨迹功能开启 ###

消息轨迹功能是异步的，其的核心也是一个`producer`，具体初始化话过程如下：

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

### `MQClient`启动 ###

这里直接来看`MQClient`的启动，如果是第一次启动则走`CREATE_JUST`分支，核心源码如下：

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
                        if (null == this.clientConfig.getNamesrvAddr()) { 
                            this.mQClientAPIImpl.fetchNameServerAddr();
                        }
                        // Start request-response channel
                        this.mQClientAPIImpl.start();
                        // Start various schedule tasks
                        this.startScheduledTask();
                        // Start pull service
                        this.pullMessageService.start();
                        // Start rebalance service
                        this.rebalanceService.start();
                        // Start push service
                        this.defaultMQProducer.getDefaultMQProducerImpl().start(false);
                        log.info("the client factory [{}] start OK", this.clientId);
                        this.serviceState = ServiceState.RUNNING;
                        break;
                    case RUNNING:
                        break;
                    case SHUTDOWN_ALREADY:
                        break;
                    case START_FAILED:
                        throw new MQClientException("The Factory object[" + this.getClientId() + "] has been created before, and failed.", null);
                    default:
                        break;
                }
            }
        }

可以看见，其主要是做了如下6件事：

*  再次检查`namesrv`地址
 *  打开client的交互信道
 *  启动pull-message服务
 *  启动push-message服务
 *  启动均衡负载服务
 *  启动一系列定时任务

其中最核心的便是启动的这些定时任务，具体定时任务如下：

*  定时刷新`namesrv`地址；本地的`topic`路由；
 *  定时处理`broker`（触发心跳，清理下线的）；
 *  定时持久化消费者消费位置；
 *  定时处理消费者的线程池；

private void startScheduledTask() { 
            if (null == this.clientConfig.getNamesrvAddr()) { 
                this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() { 
    
                    @Override
                    public void run() { 
                        try { 
                            MQClientInstance.this.mQClientAPIImpl.fetchNameServerAddr();
                        } catch (Exception e) { 
                            log.error("ScheduledTask fetchNameServerAddr exception", e);
                        }
                    }
                }, 1000 * 10, 1000 * 60 * 2, TimeUnit.MILLISECONDS);
            }
    
            this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() { 
    
                @Override
                public void run() { 
                    try { 
                        MQClientInstance.this.updateTopicRouteInfoFromNameServer();
                    } catch (Exception e) { 
                        log.error("ScheduledTask updateTopicRouteInfoFromNameServer exception", e);
                    }
                }
            }, 10, this.clientConfig.getPollNameServerInterval(), TimeUnit.MILLISECONDS);
    
            this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() { 
    
                @Override
                public void run() { 
                    try { 
                        MQClientInstance.this.cleanOfflineBroker();
                        MQClientInstance.this.sendHeartbeatToAllBrokerWithLock();
                    } catch (Exception e) { 
                        log.error("ScheduledTask sendHeartbeatToAllBroker exception", e);
                    }
                }
            }, 1000, this.clientConfig.getHeartbeatBrokerInterval(), TimeUnit.MILLISECONDS);
    
            this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() { 
    
                @Override
                public void run() { 
                    try { 
                        MQClientInstance.this.persistAllConsumerOffset();
                    } catch (Exception e) { 
                        log.error("ScheduledTask persistAllConsumerOffset exception", e);
                    }
                }
            }, 1000 * 10, this.clientConfig.getPersistConsumerOffsetInterval(), TimeUnit.MILLISECONDS);
    
            this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() { 
    
                @Override
                public void run() { 
                    try { 
                        MQClientInstance.this.adjustThreadPool();
                    } catch (Exception e) { 
                        log.error("ScheduledTask adjustThreadPool exception", e);
                    }
                }
            }, 1, 1, TimeUnit.MINUTES);
        }

### `broker`心跳触发 ###

在触发心跳的操作中，具体分为两个部分，如下所示：

// 触发心跳
    this.sendHeartbeatToAllBroker();
    // 上传加载消息过滤类
    this.uploadFilterClassSource();

上传消息过滤类资源是为了push模式下的消费者进行消息过滤用的，这里不做探究，直接来看启动触发心跳的核心代码，如下所示：

private void sendHeartbeatToAllBroker() { 
            final HeartbeatData heartbeatData = this.prepareHeartbeatData();
            final boolean producerEmpty = heartbeatData.getProducerDataSet().isEmpty();
            final boolean consumerEmpty = heartbeatData.getConsumerDataSet().isEmpty();
            if (producerEmpty && consumerEmpty) { 
                log.warn("sending heartbeat, but no consumer and no producer");
                return;
            }
    
            if (!this.brokerAddrTable.isEmpty()) { 
                long times = this.sendHeartbeatTimesTotal.getAndIncrement();
                Iterator<Entry<String, HashMap<Long, String>>> it = this.brokerAddrTable.entrySet().iterator();
                while (it.hasNext()) { 
                    Entry<String, HashMap<Long, String>> entry = it.next();
                    String brokerName = entry.getKey();
                    HashMap<Long, String> oneTable = entry.getValue();
                    if (oneTable != null) { 
                        for (Map.Entry<Long, String> entry1 : oneTable.entrySet()) { 
                            Long id = entry1.getKey();
                            String addr = entry1.getValue();
                            if (addr != null) { 
                                if (consumerEmpty) { 
                                    if (id != MixAll.MASTER_ID)
                                        continue;
                                }
    
                                try { 
                                    // 在触发心跳的时候获取broker的版本号
                                    int version = this.mQClientAPIImpl.sendHearbeat(addr, heartbeatData, 3000);
                                    if (!this.brokerVersionTable.containsKey(brokerName)) { 
                                        this.brokerVersionTable.put(brokerName, new HashMap<String, Integer>(4));
                                    }
                                    this.brokerVersionTable.get(brokerName).put(addr, version);
                                    if (times % 20 == 0) { 
                                        log.info("send heart beat to broker[{} {} {}] success", brokerName, id, addr);
                                        log.info(heartbeatData.toString());
                                    }
                                } catch (Exception e) { 
                                    if (this.isBrokerInNameServer(addr)) { 
                                        log.info("send heart beat to broker[{} {} {}] failed", brokerName, id, addr, e);
                                    } else { 
                                        log.info("send heart beat to broker[{} {} {}] exception, because the broker not up, forget it", brokerName,
                                            id, addr, e);
                                    }
                                }
                            }
                        }
                    }
                }
            }
        }