摘要：所以基于目前的設計，建議關閉自動創建的功能，然后根據消息量的大小，手動創建。如果發送消息，返回結果超時，這種超時不會進行重試了如果是方法本身耗時超過，還未來得及調用發送消息，此時的超時也不會重試。

先來看下producer核心的類設計，如下圖：

1、核心發布消息的類DefaultMQProducer，繼承自MQProducer接口，此接口定義了一系列發送消息的方法，如普通消息，順序消息，延時消息等，最終進行網絡通信會交給MQClientAPIImpl處理。

2、rocketmq從4.1.3版本開始又支持了事務消息，由TransactionMQProducer類提供（之后會有專門的文章進行詳細解讀事務消息）

我們看到DefaultMQProducer繼承了一個客戶端的公共配置類ClientConfig（與consumer公用），其實就是一個普通的javaBean，既可以代碼中設置屬性，也可以集成spring來配置

producer獨有的配置：

我們在創建producer時必須要指定一個group，這里有兩個作用：

生產者一般會是集群部署的，group用來標識一類生產者，相同group的生產者一般要有相同的發送邏輯。

在發送事務消息時，當事務消息異常，broker端來回查事務狀態時，需要知道是由哪類生產者發送的事務消息，生產端會根據group名稱來查找對應的producer來執行相應的回查邏輯。

簡單說明下整個啟動流程：

1、首先在DefaultMQProducerImpl中會做一些參數校驗，如group是否合法；然后會創建MQClientInstance實例，此實例包含網絡連接、線程資源等，相同的clientId會共享此實例，所以通過MQClientManager來管理。

2、核心的啟動流程在MQClientInstance類中，如果nameserver地址沒有配置的話，會先通過靜態的http服務器地址去抓取nameserver的地址；再則啟動netty客戶端。

3、啟動一些定時任務，跟producer有關的如下幾個：

如果producer沒有配置nameserver地址，啟動定時抓取nameserver的地址的定時任務，任務延時10s開始，每隔2分支執行一次。

輪詢nameserver定時任務，主要是定時更新topic的路由信息，任務延時10ms開始，每隔30s執行一次。

清除下線的broker和向broker發送心跳，任務延時1s執行，每隔30s執行一次

RocketMQ 有多種配置方式可以令客戶端找到 NameServer, 然后通過 NameServer 再找到 Broker，分別如下，
優先級由高到低，高優優先級會覆蓋低優先級

1、代碼中指定 Name Server 地址

producer.setNamesrvAddr("192.168.0.1:9876;192.168.0.2:9876");

2、啟動參數指定

-Drocketmq.namesrv.addr=192.168.0.1:9876;192.168.0.2:9876

3、環境變量指定 Name Server 地址

export NAMESRV_ADDR=192.168.0.1:9876;192.168.0.2:9876

4、HTTP 靜態服務器尋址（默認）

如果以上三種都沒有設置name server的地址，客戶端啟動后先會訪問一個靜態http服務器獲取name server的地址，然后會啟動一個定時任務訪問這個靜態 HTTP 服務器，地址如下：

http://jmenv.tbsite.net:8080/rocketmq/nsaddr

這是默認的地址，當然你也可以更改，做如下設置：

代碼：

System.setProperty("rocketmq.namesrv.domain","localhost");
System.setProperty("rocketmq.namesrv.domain.subgroup"，"nameServer")

或者啟動參數指定：

-Drocketmq.namesrv.domain=localhost
-Drocketmq.namesrv.domain.subgroup=nameServer

以上設置后http服務器地址就變成：

http://localhsot:8080/rocketmq/nameServer

這個 URL 的返回內容格式如下：

192.168.0.1:9876;192.168.0.2:9876

客戶端每隔 2 分鐘訪問一次這個 HTTP 服務器，并更新本地的 Name Server 地址。

推薦使用 HTTP 靜態服務器尋址方式，好處是客戶端部署簡單，且 Name Server 集群可以熱升級。

1、broker在啟動的時候通過參數autoCreateTopicEnable設置是否自動創建topic，默認為true，此時會創建一個名為TBW102(4.3版本已經改名為AUTO_CREATE_TOPIC_KEY)的topic（參見類TopicConfigManager），broker在向namesrv注冊時會把默認的topic注冊上去。如果設置false，則不會注冊。

2、producer在發送消息時會在本地獲取路由信息，第一次發送的話本地肯定沒有，就會去namesrv獲取，如果此時namesrv也沒有，則會獲取TBW102的topic信息（參見DefaultMQProducerImpl.tryToFindTopicPublishInfo），以此為模板創建topic，然后選擇topic下的一臺broker發，broker創建后，會通過心跳注冊到namesrv上。

3、如果autoCreateTopicEnable設置false的話，producer發送消息會報找不到路由的異常，此時必須手動創建topic。

建議autoCreateTopicEnable設置false，基于以上第二步，自動創建topic后，以后所有該TOPIC的消息，都將發送到剛才選擇的這臺broke上，達不到負載均衡的目的。所以基于目前RocketMQ的設計，建議關閉自動創建TOPIC的功能，然后根據消息量的大小，手動創建TOPIC。

可以通過管理工具mqadmin來手動創建topic

sh mqadmin updateTopic -c [集群名稱] -n [nameserver地址] -t [topic名稱] -w [寫隊列數] -r [讀隊列數]

手動創建了Topic后，producer就可以輪詢的發送到不同的broker了。

這里講一下自動創建的topic的隊列數如何設置，首先broker創建的模板topic=AUTO_CREATE_TOPIC_KEY的隊列是8,參見類TopicConfigManager：

public TopicConfigManager(BrokerController brokerController) { 
    //省略無關代碼
    if (this.brokerController.getBrokerConfig().isAutoCreateTopicEnable()) {
        String topic = MixAll.AUTO_CREATE_TOPIC_KEY_TOPIC;
        TopicConfig topicConfig = new TopicConfig(topic);
        this.systemTopicList.add(topic);
        topicConfig.setReadQueueNums(this.brokerController.getBrokerConfig()
                                     .getDefaultTopicQueueNums());
        topicConfig.setWriteQueueNums(this.brokerController.getBrokerConfig()
                                      .getDefaultTopicQueueNums());
        int perm = PermName.PERM_INHERIT | PermName.PERM_READ | PermName.PERM_WRITE;
        topicConfig.setPerm(perm);
        this.topicConfigTable.put(topicConfig.getTopicName(), topicConfig);
    }
     //省略無關代碼
}

BrokerConfig：

private int defaultTopicQueueNums = 8;

DefaultMQProducer端默認知道要創建的topic的隊列數是4

private volatile int defaultTopicQueueNums = 4;

在MQClientInstance類的方法updateTopicRouteInfoFromNameServer中有這樣一段邏輯：

public boolean updateTopicRouteInfoFromNameServer(final String topic, boolean isDefault,
        DefaultMQProducer defaultMQProducer) {
    //省略無關代碼
    for (QueueData data : topicRouteData.getQueueDatas()) {
        int queueNums = Math.min(defaultMQProducer.getDefaultTopicQueueNums(), data.getReadQueueNums());
        data.setReadQueueNums(queueNums);
        data.setWriteQueueNums(queueNums);
    }
    //省略無關代碼
 }

創建隊列是取兩者最小的一個，也就是4，所以要設置topic的隊列數量，很明顯了設置broker的defaultTopicQueueNums的值和DefaultMQProducer的defaultTopicQueueNums值就可以了。這是自動創建Topic時隊列數的設置方法，上面也提到生成環境一般不會開啟自動創建Topic的功能，可以通過上面的手動創建Topic的指令來設置讀寫隊列數。你可能注意到了Topic下有讀寫隊兩個隊列數，分別代表上面意思呢？讀寫隊列其實是個邏輯概念，一個broker下topic的總隊列數是以寫隊列為準，而讀隊列意思是允許多少隊列可以被消費者消費，也就是說讀多寫少的情況下，沒有問題，隊列都可以被消費掉，如果寫多讀少的話，那么就會存在隊列不會被消費的情況。

前面我們講到了如何獲取topic的路由信息，如何創建topic的隊列數，一個topic下有多個隊列，又可以分布在不同的broker上面，所以topic的總隊列數應該是所有broker上的topic下隊列數的總和。

備注：如果手動在每個broker上分別創建topic的話，相同topic在不同broker上的隊列數可以不一樣。

那么問題來了，在發送消息時根據怎么樣的策略來選擇一個隊列發送呢？rocketmq提供了一個MQFaultStrategy策略類來負責選擇隊列，這里會有一個參數sendLatencyFaultEnable是否開啟延遲故障，

該值默認為false,在不開啟的情況下，相同線程發送消息是輪詢topic下的所有隊列，不同線程發送是隨機的，核心代碼如下：

public MessageQueue selectOneMessageQueue(final TopicPublishInfo tpInfo, final String lastBrokerName) {
    if (this.sendLatencyFaultEnable) {
        //省略不必要的代碼......
    }
    return tpInfo.selectOneMessageQueue(lastBrokerName);
}
//以上代碼邏輯參見類MQFaultStrategy.selectOneMessageQueue
public MessageQueue selectOneMessageQueue(final String lastBrokerName) {
    if (lastBrokerName == null) {
        return selectOneMessageQueue();
    } else {
        //省略不必要的代碼......
    }
}
public MessageQueue selectOneMessageQueue() {
    int index = this.sendWhichQueue.getAndIncrement();
    int pos = Math.abs(index) % this.messageQueueList.size();
    if (pos < 0)
        pos = 0;
    return this.messageQueueList.get(pos);
}
//以上代碼邏輯參見類TopicPublishInfo
public int getAndIncrement() {
    Integer index = this.threadLocalIndex.get();//ThreadLocal中獲取
    if (null == index) {//為空，隨機生成一個
        index = Math.abs(random.nextInt());
        if (index < 0)
            index = 0;
        this.threadLocalIndex.set(index);
    }
    index = Math.abs(index + 1);
    if (index < 0)
        index = 0;
    this.threadLocalIndex.set(index);
    return index;
}
//以上代碼參見類ThreadLocalIndex

每次獲取index的時候都是從本地線程變量ThreadLocal中獲取，沒有的情況下就是隨機生成一個，加1取絕對值后返回，再對隊列列表的長度取模，所以在同一線程中，會輪訓的從隊列列表獲取隊列。而如果是不同線程的話，index是隨機生成的，所以就是隨機從隊列列表中獲取。如下圖所示：

可以看到選擇隊列方法的入參有一個lastBrokerName的入參，此參數的目的是在發送消息失敗的情況下，producer會重試再次發送，而再次發送選擇的隊列需要另選一個broker，lastBrokerName就是要過濾掉失敗的broker，選擇下一個broker的隊列進行發送消息。

開啟延遲故障，每當發送完一次消息，不管成功還是失敗，都會把此次存儲消息的broker給保存下來，記錄故障情況下此broker需要延長多長時間才能再次發送，目前看到在代碼里面寫死了，故障下30s之內是不能再向此broker發送消息了。

producer的send方法本身支持內部重試，重試邏輯如下：

1、最大重試次數默認2次，可以通過參數retryTimesWhenSendFailed設置

2、發送失敗，則輪詢到下一個broker，如果此時只有一個broker在線呢？那就會輪訓這個broker下的其他隊列。

3、這個方法的總耗時時間不超過 sendMsgTimeout 設置的值，默認為3s。

如果發送消息，broker返回結果超時，這種超時不會進行重試了；如果是方法本身耗時超過sendMsgTimeout ，還未來得及調用發送消息，此時的超時也不會重試。

以上策略其實也很難保證同步發送消息一定成功，如果應用要保證消息不丟失，最好先把消息存儲到db，后臺啟線程定時重試，確保消息一定存儲到broker。