RocketMQ源码解析之客户端启动时定时任务

org.apache.rocketmq.client.impl.factory.MQClientInstance#startScheduledTask 客户端启动后会启动5个定时任务

/**
 * 定时任务：该任务会在一次任务执行完毕后的间隔时间才会执行下一次任务
 */
private final ScheduledExecutorService scheduledExecutorService = Executors.newSingleThreadScheduledExecutor(new ThreadFactory() {
    @Override
    public Thread newThread(Runnable r) {
        return new Thread(r, "MQClientFactoryScheduledThread");
    }
});

/**
 * 启动5个定时任务
 * (1)若Namesrv不存在，则调用fetchNameServerAddr来获取,2min执行一次
 * (2)定时更新Topic所对应的路由信息，默认轮训时间30s
 * (3)定时清除离线的Broker，并向当前在线的Broker发送心跳包，默认间隔时间30s
 * (4)定时持久化消费者队列的消费进度
 * (5)定时调整消费者端的线程池的大小
 */
private void startScheduledTask() {
    /**
     * 定时任务1: 若Namesrv不存在，则调用fetchNameServerAddr来获取,2min执行一次
     */
    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);
    }

    /**
     * 定时任务2：定时更新Topic所对应的路由信息，默认轮训时间30s
     */
    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);

    /**
     * 定时任务3：定时清除离线的Broker，并向当前在线的Broker发送心跳包，默认间隔时间30s
     */
    this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {

        @Override
        public void run() {
            try {
                /** 清除离线的Broker */
                MQClientInstance.this.cleanOfflineBroker();
                /** 向所有在线的Broker定时发送心跳包  */
                MQClientInstance.this.sendHeartbeatToAllBrokerWithLock();
            } catch (Exception e) {
                log.error("ScheduledTask sendHeartbeatToAllBroker exception", e);
            }
        }
    }, 1000, this.clientConfig.getHeartbeatBrokerInterval(), TimeUnit.MILLISECONDS);

    /**
     * 定时任务4：定时持久化消费者队列的消费进度
     */
    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);

    /**
     * 定时任务5：定时调整消费者端的线程池的大小
     */
    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);
}

启动5个定时任务：

序号	任务作用
定时任务1	若Namesrv不存在，则调用fetchNameServerAddr来获取,2min执行一次，详细解析见fetchNameServerAddr
定时任务2	更新Topic所对应的路由信息，默认轮训时间pollNameServerInterval是30s，详细解析见updateTopicRouteInfoFromNameServer
定时任务3	清除离线的Broker，并向当前在线的Broker发送心跳包，默认间隔时间30s
定时任务4	持久化消费者队列的消费进度
定时任务5	调整消费者端的线程池的大小

MQClientAPIImpl#fetchNameServerAddr

public String fetchNameServerAddr() {
    try {
        /** 先利用GET请求获取名称服务地址，若是获取到了，则判断是否需要更新名称服务列表以及原来的nameSrvAddr */
        String addrs = this.topAddressing.fetchNSAddr();
        if (addrs != null) {
            if (!addrs.equals(this.nameSrvAddr)) {
                log.info("name server address changed, old=" + this.nameSrvAddr + ", new=" + addrs);
                /** 更新原来的nameSrvAddr */
                this.updateNameServerAddressList(addrs);
                this.nameSrvAddr = addrs;
                return nameSrvAddr;
            }
        }
    } catch (Exception e) {
        log.error("fetchNameServerAddr Exception", e);
    }
    return nameSrvAddr;
}

• 第行是利用java的httpGet 方法从一个服务器上获取GET请求，可通过配置rocketmq.namesrv.domain 来决定获取的websock地址
• 第行是更新原来的nameSrvAddr 地址，名称服务端可以是集群节点，所有会利用冒号 : 进行字符串分割，得到所有的Namesrv地址

MQClientInstance#updateTopicRouteInfoFromNameServer

/**
 * 定时更新Topic所对应的路由信息
 */
public void updateTopicRouteInfoFromNameServer() {
    /** 将所有Consumer和Producer的Topic封装在topicList */
    Set<String> topicList = new HashSet<String>();

    // Consumer
    {
        Iterator<Entry<String, MQConsumerInner>> it = this.consumerTable.entrySet().iterator();
        while (it.hasNext()) {
            Entry<String, MQConsumerInner> entry = it.next();
            MQConsumerInner impl = entry.getValue();
            if (impl != null) {
                Set<SubscriptionData> subList = impl.subscriptions();
                if (subList != null) {
                    for (SubscriptionData subData : subList) {
                        topicList.add(subData.getTopic());
                    }
                }
            }
        }
    }

    // Producer
    {
        Iterator<Entry<String, MQProducerInner>> it = this.producerTable.entrySet().iterator();
        while (it.hasNext()) {
            Entry<String, MQProducerInner> entry = it.next();
            MQProducerInner impl = entry.getValue();
            if (impl != null) {
                Set<String> lst = impl.getPublishTopicList();
                topicList.addAll(lst);
            }
        }
    }

    for (String topic : topicList) {
        // 更新topic 路由信息
        this.updateTopicRouteInfoFromNameServer(topic);
    }
}