生產(chǎn)端-初始化

源碼版本 4.6.0

先看一個簡單消息發(fā)送的例子：

public static void main(String[] args) throws MQClientException, InterruptedException {

        DefaultMQProducer producer = new DefaultMQProducer("ProducerGroupName");
        producer.start();

        for (int i = 0; i < 128; i++)
            try {
                {
                    Message msg = new Message("TopicTest",
                        "TagA",
                        "OrderID188",
                        "Hello world".getBytes(RemotingHelper.DEFAULT_CHARSET));
                    SendResult sendResult = producer.send(msg);
                    System.out.printf("%s%n", sendResult);
                }

            } catch (Exception e) {
                e.printStackTrace();
            }

        producer.shutdown();
    }

在進行消息發(fā)送，即producer.send(msg)之前，需要啟動Producer，可以猜想下在啟動Prodducer中完成了消息發(fā)送端初始化操作，本文就是對初始化進行分析。

org.apache.rocketmq.client.producer.DefaultMQProducer#start

public void start() throws MQClientException {
        // 設置生產(chǎn)者組
        this.setProducerGroup(withNamespace(this.producerGroup));
        // 核心初始化方法
        this.defaultMQProducerImpl.start();
        if (null != traceDispatcher) {
            try {
                // 消息軌跡相關(guān)
                traceDispatcher.start(this.getNamesrvAddr(), this.getAccessChannel());
            } catch (MQClientException e) {
                log.warn("trace dispatcher start failed ", e);
            }
        }
    }

org.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl#start(boolean)

public void start(final boolean startFactory) throws MQClientException {
        switch (this.serviceState) {
            case CREATE_JUST:
                this.serviceState = ServiceState.START_FAILED;
                // 檢查生產(chǎn)者組名是否符合規(guī)范，不為空且不為默認組名 DEFAULT_PRODUCER    
                this.checkConfig();

                // 更換生產(chǎn)者實例名稱，這個待①說明
                if (!this.defaultMQProducer.getProducerGroup().equals(MixAll.CLIENT_INNER_PRODUCER_GROUP)) {
                    this.defaultMQProducer.changeInstanceNameToPID();
                }

                // 獲取MQ客戶端工廠，注意這個MQClientManager是單例模式的，②補充
                this.mQClientFactory = MQClientManager.getInstance().getOrCreateMQClientInstance(this.defaultMQProducer, rpcHook);

                // 注冊該生產(chǎn)者，③處說明
                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);
                }
                
                // 如果注冊成功，則加入自動創(chuàng)建主題的內(nèi)置Topic
                this.topicPublishInfoTable.put(this.defaultMQProducer.getCreateTopicKey(), new TopicPublishInfo());

                // 啟動客戶端 ④處補充
                if (startFactory) {
                    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;
        }

        // 開始發(fā)送心跳 ⑤
        this.mQClientFactory.sendHeartbeatToAllBrokerWithLock();

        // 掃描過期請求 ⑥
        this.timer.scheduleAtFixedRate(new TimerTask() {
            @Override
            public void run() {
                try {
                    RequestFutureTable.scanExpiredRequest();
                } catch (Throwable e) {
                    log.error("scan RequestFutureTable exception", e);
                }
            }
        }, 1000 * 3, 1000);
    }
①處說明：
public void changeInstanceNameToPID() {
      if (this.instanceName.equals("DEFAULT")) {
          this.instanceName = String.valueOf(UtilAll.getPid());
      }
  }

在未設置系統(tǒng)參數(shù)rocketmq.client.name的時候，默認instanceName為DEFAULT，如果未進行設置，則設置為進程ID，即啟動JVM進程的ID

為啥這么干？我的思考點主要是以下兩點：

  1. 保證同一個JVM中，獲取的mQClientFactory只有一份，獲取mQClientFactory的參數(shù)是以Instance拼接的字符串，如果Instance保持一致，就可保證在同一個JVM中，只會創(chuàng)建一個客戶端工廠。這個有什么好處，首先mQClientFactory中包含了網(wǎng)絡組件，定時任務組件，消息拉取組件等，如果都是依據(jù)創(chuàng)建一個實例就獲取一個新的實例工廠，那么在JVM中可能存在多套相同的功能組件，這樣即造成了資源浪費，也可能使得一些內(nèi)部任務執(zhí)行錯亂。
  2. 不同JVM中的生產(chǎn)者實例能區(qū)別開
②處說明：

首先MQClientManager是單例的，也就是一個JVM中只會存在一個實例，接著看getOrCreateMQClientInstance方法，首先構(gòu)建實例ID：

String clientId = clientConfig.buildMQClientId();
            |
            |
            v
public String buildMQClientId() {
        StringBuilder sb = new StringBuilder();
        // 獲取客戶端IP
        sb.append(this.getClientIP());
        
        sb.append("@");
        // 拼裝InstanceName,一般情況下就是進程ID
        sb.append(this.getInstanceName());
        // 設置unitName 一般為空，可在Producer上設置
        if (!UtilAll.isBlank(this.unitName)) {
            sb.append("@");
            sb.append(this.unitName);
        }

        return sb.toString();
    }

總的來說clientId = IP + @ + instanceName + unitName,接著就拿這個clientId去緩存中尋找，如果沒有，就進行創(chuàng)建。
主要實例化的組件包含這幾個：

  • mQClientAPIImpl （Netty通訊組件）
  • pullMessageService （消息拉取組件）
  • rebalanceService （重平衡組件）
  • consumerStatsManager （消費信息統(tǒng)計組件）

順帶說一句，DefaultMQProducer和DefaultMQProducerImpl的關(guān)系，可以這么理解，兩者之間互相包含，DefaultMQProducer繼承了ClientConfig，更相當于一個實例自定義配置類的角色，DefaultMQProducerImpl實現(xiàn)MQProducerInner，消息發(fā)送主要邏輯是在這里面完成的。

③處說明：

注冊該主題及對應的生產(chǎn)者實例，也就是在Map中放入該數(shù)據(jù)，即：

MQProducerInner prev = this.producerTable.putIfAbsent(group, producer);
if (prev != null) {
    log.warn("the producer group[{}] exist already.", group);
     return false;
}

注意這里用的是putIfAbsent，如果生產(chǎn)者重復啟動，或者組名相同的生產(chǎn)者啟動，都會注冊失敗，觸發(fā)警告，并啟動失敗。

拋出的異常：

 new MQClientException("The producer group[" + this.defaultMQProducer.getProducerGroup()
                        + "] has been created before, specify another name please." + FAQUrl.suggestTodo(FAQUrl.GROUP_NAME_DUPLICATE_URL),
                        null);
④處說明：

獲取的客戶端實例啟動，這個是真正的啟動工作線程

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
                    // 如果Producer未設置nameServer地址，則進行遠端拉取
                    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
                    // 啟動生產(chǎn)客戶端
                    this.defaultMQProducer.getDefaultMQProducerImpl().start(false);
                    log.info("the client factory [{}] start OK", this.clientId);
                    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;
            }
        }
    }

4.1 注意當Producer為配置NameServer地址的時候，則進行遠端拉取，這個作用相當大，這個就讓線上環(huán)境動態(tài)對NameServer擴容，遷移成為可能

public String fetchNameServerAddr() {
        try {
            String addrs = this.topAddressing.fetchNSAddr();
            if (addrs != null) {
                if (!addrs.equals(this.nameSrvAddr)) {
                    log.info("name server address changed, old=" + this.nameSrvAddr + ", new=" + addrs);
                    this.updateNameServerAddressList(addrs);
                    this.nameSrvAddr = addrs;
                    return nameSrvAddr;
                }
            }
        } catch (Exception e) {
            log.error("fetchNameServerAddr Exception", e);
        }
        return nameSrvAddr;
    }

根據(jù)設置的NameServer路由拉取地址進行拉取，地址拼接如下：

public static String getWSAddr() {
        String wsDomainName = System.getProperty("rocketmq.namesrv.domain", DEFAULT_NAMESRV_ADDR_LOOKUP);
        String wsDomainSubgroup = System.getProperty("rocketmq.namesrv.domain.subgroup", "nsaddr");
        String wsAddr = "http://" + wsDomainName + ":8080/rocketmq/" + wsDomainSubgroup;
        if (wsDomainName.indexOf(":") > 0) {
            wsAddr = "http://" + wsDomainName + "/rocketmq/" + wsDomainSubgroup;
        }
        return wsAddr;
    }

4.2 定時任務啟動

private void startScheduledTask() {
        if (null == this.clientConfig.getNamesrvAddr()) {
            this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {

                @Override
                public void run() {
                    try {
                       // 拉取NameSever地址 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 {
                // 更新Topic路由地址
                    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 {
                   // 清除下線Broker，發(fā)送心跳
                    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 {
                    // 調(diào)整線程池 空實現(xiàn)，沒啥用
                    MQClientInstance.this.adjustThreadPool();
                } catch (Exception e) {
                    log.error("ScheduledTask adjustThreadPool exception", e);
                }
            }
        }, 1, 1, TimeUnit.MINUTES);
    }

總結(jié)下：

  • 拉取NameServer地址，延時10S,頻率2min
  • 更新主題路由信息，延時10ms,頻率30S
  • 向Broker發(fā)送心跳，延時1S,頻率30S
  • 消費進度持久化，延時1S,頻率5S
  • 動態(tài)調(diào)整線程池，不起作用

值得注意的是，定時任務線程池是單線程無界隊列類型的，且用的FixedRate模式，實際的執(zhí)行頻率可能不是準確的，有興趣可以看下ScheduledExecutorService源碼

4.3 其余的組件啟動和消費相關(guān)，這里先不深入了

⑤處說明：

this.mQClientFactory.sendHeartbeatToAllBrokerWithLock();

啟動成功后開始發(fā)送心跳，心跳發(fā)送的過程是持有鎖的，個人感覺主要是避免心跳混亂，特殊用途暫時沒聯(lián)想到。

心跳發(fā)送主代碼：

 private void sendHeartbeatToAllBroker() {
        // 準備心跳發(fā)送包，主要是消費訂閱配置和生產(chǎn)者配置等信息，這個后續(xù)再詳細討論
        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;
        }
        // 獲取所有的Broker地址
        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 {
                                // 發(fā)送心跳，超時3S
                                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);
                                }
                            }
                        }
                    }
                }
            }
        }
    }

這塊值得注意的是：

  • 如果當前JVM中只有生產(chǎn)者實例，那么只向主節(jié)點發(fā)送心跳。
  • 如果當前JVM即存在生產(chǎn)者又存在消費者，那么就向所有節(jié)點發(fā)送心跳。這個和消息發(fā)送邏輯，消息消費邏輯有關(guān)，后期再談。

從這也可以學到，只要涉及到網(wǎng)絡請求，請加上超時，為了你的服務穩(wěn)定！

⑤處說明：

移除過期請求，這個requestFutureTable的填充涉及的API：

org.apache.rocketmq.client.producer.DefaultMQProducer#request(org.apache.rocketmq.common.message.Message, long)

從方法說明上看是發(fā)送消息，在等到該消息消費后再返回，提供異步和同步模式的API，改API在生產(chǎn)上沒實際用過，關(guān)于這個就不過多講解了。

最后提供一個總圖：


啟動流程-Producer.png
最后編輯于 ：
?著作權(quán)歸作者所有,轉(zhuǎn)載或內(nèi)容合作請聯(lián)系作者
【社區(qū)內(nèi)容提示】社區(qū)部分內(nèi)容疑似由AI輔助生成，瀏覽時請結(jié)合常識與多方信息審慎甄別。
平臺聲明：文章內(nèi)容（如有圖片或視頻亦包括在內(nèi)）由作者上傳并發(fā)布，文章內(nèi)容僅代表作者本人觀點，簡書系信息發(fā)布平臺，僅提供信息存儲服務。

相關(guān)閱讀更多精彩內(nèi)容

友情鏈接更多精彩內(nèi)容