Producer隨機與一個NameServer建立長連線，從NameServer獲取topic的最新佇列情況。Producer會向提供topic服務的master建立長連線，且定時向master傳送心跳。
傳送訊息demo:

// 構造Producer
DefaultMQProducer producer = new DefaultMQProducer("ProducerGroupName");
// 初始化Producer，整個應用生命週期內，只需要初始化1次
producer.start();
// 構造Message
Message msg = new Message("TopicTest1",// topic
 

                        "TagA",// tag：給訊息打標籤,用於區分一類訊息，可為null
                        "OrderID188",// key：自定義Key，可以用於去重，可為null
                        ("Hello MetaQ").getBytes());// body：訊息內容
// 傳送訊息並返回結果
SendResult sendResult = producer.send(msg);
// 清理資源，關閉網路連線，登出自己
producer.shutdown();

一：start()
在整個應用生命週期內，生產者只需要呼叫一次start方法

• 如果沒有指定namesrv地址，將會自動定址
• 啟動定時任務：從namsrv更新topic路由資訊、清理已經掛掉的broker、向所有線上的broker master傳送心跳…
• 啟動負載均衡的服務：producer根據roundbin方式輪詢topic下的所有佇列來實現傳送方的負載均衡。

二：DefaultMQProducer
send的內部實現原理虛擬碼如下：

private SendResult sendDefaultImpl(Message msg,......) {
    // 檢查Producer的狀態是否是RUNNING
    this.makeSureStateOK();
    // 檢查msg是否合法：是否為null、topic,body是否為空、body是否超長
 

    Validators.checkMessage(msg, this.defaultMQProducer);
    // 獲取topic路由資訊
    TopicPublishInfo topicPublishInfo = this.tryToFindTopicPublishInfo(msg.getTopic());
    // 從路由資訊中選擇一個訊息佇列
    MessageQueue mq = topicPublishInfo.selectOneMessageQueue(lastBrokerName);
    // 將訊息傳送到該佇列上去
    sendResult = this.sendKernelImpl(msg, mq, communicationMode, sendCallback, timeout);
}

1：producer初始化的時候會從namesrv獲取topic的路由資訊更新到本地快取，所以tryToFindTopicPublishInfo會先從本地快取取，如果沒取到再從namesrv獲取最新的路由資訊。

 private TopicPublishInfo tryToFindTopicPublishInfo(final String topic) {
        TopicPublishInfo topicPublishInfo = this.topicPublishInfoTable.get(topic);
        if (null == topicPublishInfo || !topicPublishInfo.ok()) {
            this.topicPublishInfoTable.putIfAbsent(topic, new TopicPublishInfo());
            this.mQClientFactory.updateTopicRouteInfoFromNameServer(topic);
            topicPublishInfo = this.topicPublishInfoTable.get(topic);
        }

        if (topicPublishInfo.isHaveTopicRouterInfo() || (topicPublishInfo != null &&    topicPublishInfo.ok())) {
            return topicPublishInfo;
        } else {
            this.mQClientFactory.updateTopicRouteInfoFromNameServer(topic, true, this.defaultMQProducer);
            topicPublishInfo = this.topicPublishInfoTable.get(topic);
            return topicPublishInfo;
        }
    }

2.傳送訊息時，當沒有指定佇列或佇列選擇器時，呼叫selectOneMessageQueue：使用輪詢方式，返回一個佇列。當指定佇列選擇器時，通過selector.select(topicPublishInfo.getMessageQueueList(), msg, arg);獲取佇列。
3.訊息傳送
CommunicationMode:
SYNC:同步（預設）,
ASYNC：非同步（callback）,
ONEWAY：單向,

private SendResult sendKernelImpl(final Message msg, 
                                      final MessageQueue mq, 
                                      final CommunicationMode communicationMode,
                                      ......){
     String brokerAddr = this.mQClientFactory.findBrokerAddressInPublish(mq.getBrokerName());
     SendMessageRequestHeader requestHeader = new SendMessageRequestHeader();
     switch (communicationMode) {
         case ASYNC:
                  sendResult = this.mQClientFactory.getMQClientAPIImpl().sendMessage(
                                brokerAddr, // 1
                                mq.getBrokerName(), // 2
                                msg, // 3
                                requestHeader, // 4
                                timeout, // 5
                                communicationMode, // 6
                                sendCallback, // 7
                                topicPublishInfo, // 8
                                this.mQClientFactory, // 9
                                this.defaultMQProducer.getRetryTimesWhenSendAsyncFailed(), //10
                                context, //
                                this);
                        break;
                    case ONEWAY:
                    case SYNC:
                       ....
                        break;

}