RocketMQ源碼：Producer啓動分析

    本文主要分析RocketMQ中Producer的啓動過程。

    RocketMQ的版本爲：4.2.0 release。

一.時序圖

    根據源碼，把Producer啓動過程的時序圖畫了一遍：

 

二.源碼分析

    1 start() ：DefaultMQProducer啓動。

    DefaultMQProducer主要功能都是在DefaultMQProducerImpl中實現的。相似的，DefaultMQPushConsumer的大部分功能也在DefaultMQPushConsumerImpl中實現：

//DefaultMQProducer#start
    public void start() throws MQClientException {
        this.defaultMQProducerImpl.start();
    }

 

    1.1 checkConfig：檢查producerGroup是否合法。

// DefaultMQProducerImpl#checkConfig
    private void checkConfig() throws MQClientException {
        Validators.checkGroup(this.defaultMQProducer.getProducerGroup());
        if (null == this.defaultMQProducer.getProducerGroup()) {
            throw new MQClientException("producerGroup is null", null);
        }
        if (this.defaultMQProducer.getProducerGroup().equals(MixAll.DEFAULT_PRODUCER_GROUP)) {// 不能等於"DEFAULT_PRODUCER"
            throw new MQClientException("producerGroup can not equal " + MixAll.DEFAULT_PRODUCER_GROUP + ", please specify another one.",
                null);
        }
    }
    // Validators#checkGroup
    public static void checkGroup(String group) throws MQClientException {
        if (UtilAll.isBlank(group)) {// 不能爲空
            throw new MQClientException("the specified group is blank", null);
        }
        if (!regularExpressionMatcher(group, PATTERN)) {
            throw new MQClientException(String.format(
                "the specified group[%s] contains illegal characters, allowing only %s", group,
                VALID_PATTERN_STR), null);
        }
        if (group.length() > CHARACTER_MAX_LENGTH) {// 長度不能大於255
            throw new MQClientException("the specified group is longer than group max length 255.", null);
        }
    }

 

    1.2 getAndCreateMQClientInstance：獲取MQClientInstance。

//MQClientManager#getAndCreateMQClientInstance
    public MQClientInstance getAndCreateMQClientInstance(final ClientConfig clientConfig, RPCHook rpcHook) {
        String clientId = clientConfig.buildMQClientId();// 構建該Producer的ClientID，等於IP地址@instanceName
        MQClientInstance instance = this.factoryTable.get(clientId);
        if (null == instance) {// 若是當前客戶端不在mq客戶端實例集合中,則建立一個實例並加入
            instance =
                new MQClientInstance(clientConfig.cloneClientConfig(),
                    this.factoryIndexGenerator.getAndIncrement(), clientId, rpcHook);
            MQClientInstance prev = this.factoryTable.putIfAbsent(clientId, instance);
            if (prev != null) {// 說明一個IP客戶端下面的應用，只有在啓動多個進程的狀況下才會建立多個MQClientInstance對象
                instance = prev;
                log.warn("Returned Previous MQClientInstance for clientId:[{}]", clientId);
            } else {
                log.info("Created new MQClientInstance for clientId:[{}]", clientId);
            }
        }
        return instance;
    }

 

    1.3 registerProducer：註冊Producer。

// MQClientInstance#registerProducer
    public boolean registerProducer(final String group, final DefaultMQProducerImpl producer) {
        if (null == group || null == producer) {
            return false;
        }
        MQProducerInner prev = this.producerTable.putIfAbsent(group, producer);// 若是沒有添加過，就往producerTable中加入當前的Producer
        if (prev != null) {
            log.warn("the producer group[{}] exist already.", group);
            return false;
        }
        return true;
    }

 

    1.4 MQClientInstance#start 啓動mQClientFactory。

// DefaultMQProducerImpl#start(true)
    if (startFactory) {
        mQClientFactory.start();
    }
    // 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
                    if (null == this.clientConfig.getNamesrvAddr()) {
                        this.mQClientAPIImpl.fetchNameServerAddr();// 獲取nameService地址
                    }
                    // Start request-response channel
                    this.mQClientAPIImpl.start();// 對象負責底層消息通訊，獲取nameService地址
                    // 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);// 再次調用DefaultMQProducerImpl.start()，注意傳參爲false。此時ServiceState仍是 START_FAILED 只調用了一次心跳服務 this.mQClientFactory.sendHeartbeatToAllBrokerWithLock()
                    log.info("the client factory [{}] start OK", this.clientId);
                    this.serviceState = ServiceState.RUNNING;// 改變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;
            }
        }
    }

 

    1.4.1 MQClientAPIImpl#start 負責底層消息通訊，啓動客戶端對象。

// MQClientAPIImpl#start
    public void start() {
        this.remotingClient.start();// RemotingClient是RocketMQ封裝了Netty網絡通訊的客戶端
    }

 

    1.4.2 MQClientInstance#startScheduledTask 啓動各類定時任務。

// MQClientInstance#startScheduledTask
    private void startScheduledTask() {
        if (null == this.clientConfig.getNamesrvAddr()) {
            this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
                @Override
                public void run() {
                    try {
                        MQClientInstance.this.mQClientAPIImpl.fetchNameServerAddr();// 更新NameServer地址
                    } 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();// 從nameService更新Topic路由信息
                } 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();// 清理掛掉的broker
                    MQClientInstance.this.sendHeartbeatToAllBrokerWithLock();// 向broker發送心跳信息
                } 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();// 持久化consumerOffset，保存消費者的Offset
                } 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);
    }

 

    1.5 MQClientInstance#sendHeartbeatToAllBrokerWithLock 向全部的Broker發送心跳信息。

// MQClientInstance#sendHeartbeatToAllBrokerWithLock
    public void sendHeartbeatToAllBrokerWithLock() {
        if (this.lockHeartbeat.tryLock()) {
            try {
                this.sendHeartbeatToAllBroker();// 向全部在MQClientInstance.brokerAddrTable列表中的Broker發送心跳消息
                this.uploadFilterClassSource();// 向Filter過濾服務器發送REGISTER_MESSAGE_FILTER_CLASS請求碼，更新過濾服務器中的Filterclass文件
            } catch (final Exception e) {
                log.error("sendHeartbeatToAllBroker exception", e);
            } finally {
                this.lockHeartbeat.unlock();
            }
        } else {
            log.warn("lock heartBeat, but failed.");
        }
    }

 

    上面就是RocketMQ中Producer的啓動過程，上面分析了主要的幾處地方，若是想了解啓動過程當中的詳細代碼，能夠從Github上面clone代碼到本地，試着調試和分析。附上地址：4.2.0 release。