RocketMQ中Producer的啟動源碼分析

RocketMQ中通過DefaultMQProducer創建Producer DefaultMQProducer定義如下：其中defaultMQProducerImpl成員是Producer的具體實現，其餘的一些成員是對一些參數的設置：createTopicKey：是一個Topic值，在創建時使用 ...

RocketMQ中通過DefaultMQProducer創建Producer

DefaultMQProducer定義如下：

 1 public class DefaultMQProducer extends ClientConfig implements MQProducer {
 2     protected final transient DefaultMQProducerImpl defaultMQProducerImpl;
 3     
 4     private String createTopicKey = MixAll.AUTO_CREATE_TOPIC_KEY_TOPIC = "TBW102";
 5     
 6     private volatile int defaultTopicQueueNums = 4;
 7     
 8     private int sendMsgTimeout = 3000;
 9     
10     private int compressMsgBodyOverHowmuch = 1024 * 4;
11     
12     private int retryTimesWhenSendFailed = 2;
13     
14     private int retryTimesWhenSendAsyncFailed = 2;
15     
16     private boolean retryAnotherBrokerWhenNotStoreOK = false;
17     
18     private int maxMessageSize = 1024 * 1024 * 4; // 4M
19 }

其中defaultMQProducerImpl成員是Producer的具體實現，其餘的一些成員是對一些參數的設置：
createTopicKey：是一個Topic值，在創建時使用，後面會說明
defaultTopicQueueNums ：預設的Topic隊列個數
sendMsgTimeout：發送消息超時時間
compressMsgBodyOverHowmuch：消息容量限制，超過需要進行壓縮
retryTimesWhenSendFailed：同步消息發送失敗的允許重發次數
retryTimesWhenSendAsyncFailed：非同步消息發送失敗的允許重發次數
retryAnotherBrokerWhenNotStoreOK：是否允許發送給Broker失敗後，重新選擇Broker發送
maxMessageSize：消息最大大小
這些屬性可以通過DefaultMQProducer提供的get、set方法進行相應操作

常用的構造方法如下：

 1 public DefaultMQProducer() {
 2     this(MixAll.DEFAULT_PRODUCER_GROUP, null);
 3 }
 4 
 5 public DefaultMQProducer(final String producerGroup) {
 6     this(producerGroup, null);
 7 }
 8 
 9 public DefaultMQProducer(final String producerGroup, RPCHook rpcHook) {
10     this.producerGroup = producerGroup;
11     defaultMQProducerImpl = new DefaultMQProducerImpl(this, rpcHook);
12 }

DefaultMQProducer繼承自ClientConfig，首先會設置ClientConfig提供的更底層的參數配置：

 1 public class ClientConfig {
 2     public static final String SEND_MESSAGE_WITH_VIP_CHANNEL_PROPERTY = "com.rocketmq.sendMessageWithVIPChannel";
 3     
 4     private String namesrvAddr = System.getProperty(MixAll.NAMESRV_ADDR_PROPERTY, System.getenv(MixAll.NAMESRV_ADDR_ENV));
 5     
 6     private String clientIP = RemotingUtil.getLocalAddress();
 7     
 8     private String instanceName = System.getProperty("rocketmq.client.name", "DEFAULT");
 9     
10     private int clientCallbackExecutorThreads = Runtime.getRuntime().availableProcessors();
11     
12     private int pollNameServerInterval = 1000 * 30;
13     
14     private int heartbeatBrokerInterval = 1000 * 30;
15     
16     private boolean vipChannelEnabled = Boolean.parseBoolean(System.getProperty(SEND_MESSAGE_WITH_VIP_CHANNEL_PROPERTY, "true"));
17 }

其中namesrvAddr是非常重要的成員，其保存著名稱伺服器（Name Server）的地址，在一開始構造時會根據系統屬性進行設置，若是沒有設置系統屬性就是null，則需要在後面通過set方法進行設置
clientIP：Producer端的本地IP
instanceName：Producer的實例名稱
pollNameServerInterval ：輪詢NameServer的時間間隔
heartbeatBrokerInterval ：向Broker發送心跳包的時間間隔
SEND_MESSAGE_WITH_VIP_CHANNEL_PROPERTY 和vipChannelEnabled：決定是否使用VIP通道，即高優先順序

回到DefaultMQProducer的構造方法，其會創建DefaultMQProducerImpl實例

 1 private final Random random = new Random();
 2 private final ConcurrentMap<String/* topic */, TopicPublishInfo> topicPublishInfoTable =
 3         new ConcurrentHashMap<String, TopicPublishInfo>();
 4 private final ArrayList<SendMessageHook> sendMessageHookList = new ArrayList<SendMessageHook>();
 5 private final RPCHook rpcHook;
 6 protected BlockingQueue<Runnable> checkRequestQueue;
 7 protected ExecutorService checkExecutor;
 8 private ServiceState serviceState = ServiceState.CREATE_JUST;
 9 private MQClientInstance mQClientFactory;
10 private ArrayList<CheckForbiddenHook> checkForbiddenHookList = new ArrayList<CheckForbiddenHook>();
11 private int zipCompressLevel = Integer.parseInt(System.getProperty(MixAll.MESSAGE_COMPRESS_LEVEL, "5"));
12 private MQFaultStrategy mqFaultStrategy = new MQFaultStrategy();
13 private final BlockingQueue<Runnable> asyncSenderThreadPoolQueue;
14 private final ExecutorService defaultAsyncSenderExecutor;
15 private ExecutorService asyncSenderExecutor;
16 
17 public DefaultMQProducerImpl(final DefaultMQProducer defaultMQProducer, RPCHook rpcHook) {
18     this.defaultMQProducer = defaultMQProducer;
19     this.rpcHook = rpcHook;
20 
21     this.asyncSenderThreadPoolQueue = new LinkedBlockingQueue<Runnable>(50000);
22     this.defaultAsyncSenderExecutor = new ThreadPoolExecutor(
23         Runtime.getRuntime().availableProcessors(),
24         Runtime.getRuntime().availableProcessors(),
25         1000 * 60,
26         TimeUnit.MILLISECONDS,
27         this.asyncSenderThreadPoolQueue,
28         new ThreadFactory() {
29             private AtomicInteger threadIndex = new AtomicInteger(0);
30 
31             @Override
32             public Thread newThread(Runnable r) {
33                 return new Thread(r, "AsyncSenderExecutor_" + this.threadIndex.incrementAndGet());
34             }
35         });
36 }

在構造方法中會創建一個線程池，用來處理非同步消息的發送
其中有一個topicPublishInfoTable成員很重要，是一個map，保存了不同top和消息隊列之間的映射，在後面詳細介紹

DefaultMQProducer創建完成後，接著來看DefaultMQProducer的start方法：

 1 public void start() throws MQClientException {
 2     this.defaultMQProducerImpl.start();
 3     if (null != traceDispatcher) {
 4         try {
 5             traceDispatcher.start(this.getNamesrvAddr());
 6         } catch (MQClientException e) {
 7             log.warn("trace dispatcher start failed ", e);
 8         }
 9     }
10 }

首先交給了defaultMQProducerImpl的start方法去處理

defaultMQProducerImpl的start方法：

 1 public void start() throws MQClientException {
 2     this.start(true);
 3 }
 4 
 5 public void start(final boolean startFactory) throws MQClientException {
 6     switch (this.serviceState) {
 7         case CREATE_JUST:
 8             this.serviceState = ServiceState.START_FAILED;
 9 
10             this.checkConfig();
11 
12             if (!this.defaultMQProducer.getProducerGroup().equals(MixAll.CLIENT_INNER_PRODUCER_GROUP)) {
13                 this.defaultMQProducer.changeInstanceNameToPID();
14             }
15 
16             this.mQClientFactory = MQClientManager.getInstance().getAndCreateMQClientInstance(this.defaultMQProducer, rpcHook);
17 
18             boolean registerOK = mQClientFactory.registerProducer(this.defaultMQProducer.getProducerGroup(), this);
19             if (!registerOK) {
20                 this.serviceState = ServiceState.CREATE_JUST;
21                 throw new MQClientException("The producer group[" + this.defaultMQProducer.getProducerGroup()
22                     + "] has been created before, specify another name please." + FAQUrl.suggestTodo(FAQUrl.GROUP_NAME_DUPLICATE_URL),
23                     null);
24             }
25 
26             this.topicPublishInfoTable.put(this.defaultMQProducer.getCreateTopicKey(), new TopicPublishInfo());
27 
28             if (startFactory) {
29                 mQClientFactory.start();
30             }
31 
32             log.info("the producer [{}] start OK. sendMessageWithVIPChannel={}", this.defaultMQProducer.getProducerGroup(),
33                 this.defaultMQProducer.isSendMessageWithVIPChannel());
34             this.serviceState = ServiceState.RUNNING;
35             break;
36         case RUNNING:
37         case START_FAILED:
38         case SHUTDOWN_ALREADY:
39             throw new MQClientException("The producer service state not OK, maybe started once, "
40                 + this.serviceState
41                 + FAQUrl.suggestTodo(FAQUrl.CLIENT_SERVICE_NOT_OK),
42                 null);
43         default:
44             break;
45     }
46 
47     this.mQClientFactory.sendHeartbeatToAllBrokerWithLock();
48 }

在一開始DefaultMQProducerImpl實例化的時候，serviceState初始化為CREATE_JUST狀態，這是一個枚舉值，一共有如下幾種狀態：

1 public enum ServiceState {
2     CREATE_JUST,
3     RUNNING,
4     SHUTDOWN_ALREADY,
5     START_FAILED;
6 
7     private ServiceState() {
8     }
9 }

這幾個狀態值很容易理解，在後面MQClientInstance中還會使用到

回到start方法，根據serviceState進行判斷，只有當是CREATE_JUST狀態時正常執行，防止在其他狀態下錯誤調用start

直接看到CREATE_JUST的case部分：

 1 this.serviceState = ServiceState.START_FAILED;
 2 
 3 this.checkConfig();
 4 
 5 if (!this.defaultMQProducer.getProducerGroup().equals(MixAll.CLIENT_INNER_PRODUCER_GROUP)) {
 6 this.defaultMQProducer.changeInstanceNameToPID();
 7 }
 8 
 9 this.mQClientFactory = MQClientManager.getInstance().getAndCreateMQClientInstance(this.defaultMQProducer, rpcHook);
10 
11 boolean registerOK = mQClientFactory.registerProducer(this.defaultMQProducer.getProducerGroup(), this);
12 if (!registerOK) {
13 this.serviceState = ServiceState.CREATE_JUST;
14 throw new MQClientException("The producer group[" + this.defaultMQProducer.getProducerGroup()
15 + "] has been created before, specify another name please." + FAQUrl.suggestTodo(FAQUrl.GROUP_NAME_DUPLICATE_URL),
16 null);
17 }
18 
19 this.topicPublishInfoTable.put(this.defaultMQProducer.getCreateTopicKey(), new TopicPublishInfo());
20 
21 if (startFactory) {
22 mQClientFactory.start();
23 }
24 
25 log.info("the producer [{}] start OK. sendMessageWithVIPChannel={}", this.defaultMQProducer.getProducerGroup(),
26 this.defaultMQProducer.isSendMessageWithVIPChannel());
27 this.serviceState = ServiceState.RUNNING;
28 break;

首先更改serviceState狀態為START_FAILED，防止中途的失敗

checkConfig方法是用來進行ProducerGroup命名檢查：

 1 private void checkConfig() throws MQClientException {
 2     Validators.checkGroup(this.defaultMQProducer.getProducerGroup());
 3 
 4     if (null == this.defaultMQProducer.getProducerGroup()) {
 5         throw new MQClientException("producerGroup is null", null);
 6     }
 7 
 8     if (this.defaultMQProducer.getProducerGroup().equals(MixAll.DEFAULT_PRODUCER_GROUP)) {
 9         throw new MQClientException("producerGroup can not equal " + MixAll.DEFAULT_PRODUCER_GROUP + ", please specify another one.",
10             null);
11     }
12 }

主要是檢查命名的合法性，以及防止和預設的producerGroup生產者組名DEFAULT_PRODUCER_GROUP產生衝突

1 public static final String DEFAULT_PRODUCER_GROUP = "DEFAULT_PRODUCER";

接下來實例化mQClientFactory，這其實是生產者客戶端的實例，其中MQClientManager採用單例模式，getInstance是獲取MQClientManager的單例，根據ClientConfig的類型，通過getAndCreateMQClientInstance方法實例化不同屬性的生產者客戶端

MQClientManager：

 1 public class MQClientManager {
 2     private final static InternalLogger log = ClientLogger.getLog();
 3     private static MQClientManager instance = new MQClientManager();
 4     private AtomicInteger factoryIndexGenerator = new AtomicInteger();
 5     private ConcurrentMap<String/* clientId */, MQClientInstance> factoryTable =
 6         new ConcurrentHashMap<String, MQClientInstance>();
 7 
 8     private MQClientManager() {
 9     }
10 
11     public static MQClientManager getInstance() {
12         return instance;
13     }
14 }

其中factoryTable是所有生產者客戶端實例的map緩存，factoryIndexGenerator 是創建的每個客戶端實例的流水號

getAndCreateMQClientInstance方法：

 1 public MQClientInstance getAndCreateMQClientInstance(final ClientConfig clientConfig, RPCHook rpcHook) {
 2     String clientId = clientConfig.buildMQClientId();
 3     MQClientInstance instance = this.factoryTable.get(clientId);
 4     if (null == instance) {
 5         instance =
 6             new MQClientInstance(clientConfig.cloneClientConfig(),
 7                 this.factoryIndexGenerator.getAndIncrement(), clientId, rpcHook);
 8         MQClientInstance prev = this.factoryTable.putIfAbsent(clientId, instance);
 9         if (prev != null) {
10             instance = prev;
11             log.warn("Returned Previous MQClientInstance for clientId:[{}]", clientId);
12         } else {
13             log.info("Created new MQClientInstance for clientId:[{}]", clientId);
14         }
15     }
16 
17     return instance;
18 }

首先通過buildMQClientId方法創建clientId：

 1 public String buildMQClientId() {
 2     StringBuilder sb = new StringBuilder();
 3     sb.append(this.getClientIP());
 4 
 5     sb.append("@");
 6     sb.append(this.getInstanceName());
 7     if (!UtilAll.isBlank(this.unitName)) {
 8         sb.append("@");
 9         sb.append(this.unitName);
10     }
11 
12     return sb.toString();
13 }

clientId主要由生產者客戶端的ip地址以及實例名稱，根據unitName的有無，附加unitName

通過生成的clientId，在factoryTable緩存中先去獲取是否創建過客戶端實例
若是沒有獲取到，就需要實例化一個MQClientInstance
這裡在實例化MQClientInstance時，並沒有直接傳入clientConfig，而是通過cloneClientConfig方法複製了一份，來保證安全性：

 1 public ClientConfig cloneClientConfig() {
 2     ClientConfig cc = new ClientConfig();
 3     cc.namesrvAddr = namesrvAddr;
 4     cc.clientIP = clientIP;
 5     cc.instanceName = instanceName;
 6     cc.clientCallbackExecutorThreads = clientCallbackExecutorThreads;
 7     cc.pollNameServerInterval = pollNameServerInterval;
 8     cc.heartbeatBrokerInterval = heartbeatBrokerInterval;
 9     cc.persistConsumerOffsetInterval = persistConsumerOffsetInterval;
10     cc.unitMode = unitMode;
11     cc.unitName = unitName;
12     cc.vipChannelEnabled = vipChannelEnabled;
13     cc.useTLS = useTLS;
14     cc.language = language;
15     return cc;
16 }

創建MQClientInstance實例：

 1 public MQClientInstance(ClientConfig clientConfig, int instanceIndex, String clientId, RPCHook rpcHook) {
 2     this.clientConfig = clientConfig;
 3     this.instanceIndex = instanceIndex;
 4     this.nettyClientConfig = new NettyClientConfig();
 5     this.nettyClientConfig.setClientCallbackExecutorThreads(clientConfig.getClientCallbackExecutorThreads());
 6     this.nettyClientConfig.setUseTLS(clientConfig.isUseTLS());
 7     this.clientRemotingProcessor = new ClientRemotingProcessor(this);
 8     this.mQClientAPIImpl = new MQClientAPIImpl(this.nettyClientConfig, this.clientRemotingProcessor, rpcHook, clientConfig);
 9 
10     if (this.clientConfig.getNamesrvAddr() != null) {
11         this.mQClientAPIImpl.updateNameServerAddressList(this.clientConfig.getNamesrvAddr());
12         log.info("user specified name server address: {}", this.clientConfig.getNamesrvAddr());
13     }
14 
15     this.clientId = clientId;
16 
17     this.mQAdminImpl = new MQAdminImpl(this);
18 
19     this.pullMessageService = new PullMessageService(this);
20 
21     this.rebalanceService = new RebalanceService(this);
22 
23     this.defaultMQProducer = new DefaultMQProducer(MixAll.CLIENT_INNER_PRODUCER_GROUP);
24     this.defaultMQProducer.resetClientConfig(clientConfig);
25 
26     this.consumerStatsManager = new ConsumerStatsManager(this.scheduledExecutorService);
27 
28     log.info("Created a new client Instance, InstanceIndex:{}, ClientID:{}, ClientConfig:{}, ClientVersion:{}, SerializerType:{}",
29         this.instanceIndex,
30         this.clientId,
31         this.clientConfig,
32         MQVersion.getVersionDesc(MQVersion.CURRENT_VERSION), RemotingCommand.getSerializeTypeConfigInThisServer());
33 }

可以看到MQClientInstance的構造方法創建了很多東西，就不一一說明，主要說幾個重要的
其中nettyClientConfig，就很清楚的說明瞭RocketMQ通過Netty來進行網路之間的I/O，其保存了對Netty的一些配置
clientRemotingProcessor，用來進行消息的處理

mQClientAPIImpl則是一個非常重要的部分，直接實例化了一個MQClientAPIImpl對象：

 1 public MQClientAPIImpl(final NettyClientConfig nettyClientConfig,
 2     final ClientRemotingProcessor clientRemotingProcessor,
 3     RPCHook rpcHook, final ClientConfig clientConfig) {
 4     this.clientConfig = clientConfig;
 5     topAddressing = new TopAddressing(MixAll.getWSAddr(), clientConfig.getUnitName());
 6     this.remotingClient = new NettyRemotingClient(nettyClientConfig, null);
 7     this.clientRemotingProcessor = clientRemotingProcessor;
 8     
 9     this.remotingClient.registerRPCHook(rpcHook);
10     this.remotingClient.registerProcessor(RequestCode.CHECK_TRANSACTION_STATE, this.clientRemotingProcessor, null);
11     
12     this.remotingClient.registerProcessor(RequestCode.NOTIFY_CONSUMER_IDS_CHANGED, this.clientRemotingProcessor, null);
13     
14     this.remotingClient.registerProcessor(RequestCode.RESET_CONSUMER_CLIENT_OFFSET, this.clientRemotingProcessor, null);
15     
16     this.remotingClient.registerProcessor(RequestCode.GET_CONSUMER_STATUS_FROM_CLIENT, this.clientRemotingProcessor, null);
17     
18     this.remotingClient.registerProcessor(RequestCode.GET_CONSUMER_RUNNING_INFO, this.clientRemotingProcessor, null);
19     
20     this.remotingClient.registerProcessor(RequestCode.CONSUME_MESSAGE_DIRECTLY, this.clientRemotingProcessor, null);
21 }

可以看到在這個構造方法里，首先創建了一個TopAddressing，用於以後的名稱服務的定址，其預設地址是：

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

需要通過系統屬性來完成更改

接著創建了一個NettyRemotingClient，這個就是實實在在的Netty客戶端

 1 private final Bootstrap bootstrap = new Bootstrap();
 2 // 名稱服務列表
 3 private final AtomicReference<List<String>> namesrvAddrList = new AtomicReference<List<String>>();
 4 
 5 public NettyRemotingClient(final NettyClientConfig nettyClientConfig,
 6     final ChannelEventListener channelEventListener) {
 7     super(nettyClientConfig.getClientOnewaySemaphoreValue(), nettyClientConfig.getClientAsyncSemaphoreValue());
 8     this.nettyClientConfig = nettyClientConfig;
 9     this.channelEventListener = channelEventListener;
10 
11     int publicThreadNums = nettyClientConfig.getClientCallbackExecutorThreads();
12     if (publicThreadNums <= 0) {
13         publicThreadNums = 4;
14     }
15 
16     this.publicExecutor = Executors.newFixedThreadPool(publicThreadNums, new ThreadFactory() {
17         private AtomicInteger threadIndex = new AtomicInteger(0);
18 
19         @Override
20         public Thread newThread(Runnable r) {
21             return new Thread(r, "NettyClientPublicExecutor_" + this.threadIndex.incrementAndGet());
22         }
23     });
24 
25     this.eventLoopGroupWorker = new NioEventLoopGroup(1, new ThreadFactory() {
26         private AtomicInteger threadIndex = new AtomicInteger(0);
27 
28         @Override
29         public Thread newThread(Runnable r) {
30             return new Thread(r, String.format("NettyClientSelector_%d", this.threadIndex.incrementAndGet()));
31         }
32     });
33 
34     if (nettyClientConfig.isUseTLS()) {
35         try {
36             sslContext = TlsHelper.buildSslContext(true);
37             log.info("SSL enabled for client");
38         } catch (IOException e) {
39             log.error("Failed to create SSLContext", e);
40         } catch (CertificateException e) {
41             log.error("Failed to create SSLContext", e);
42             throw new RuntimeException("Failed to create SSLContext", e);
43         }
44     }
45 }

此時Netty的客戶端僅僅完成了對Bootstrap的初始化，以及對NioEventLoopGroup的設置和初始化

回到MQClientInstance的構造方法，在完成MQClientAPIImpl的創建後，會根據clientConfig的getNamesrvAddr判斷是否設置了namesrvAddr名稱服務地址，若是設置了，需要通過mQClientAPIImpl的updateNameServerAddressList方法，完成對名稱服務地址的更新：

MQClientAPIImpl的updateNameServerAddressList方法：

1 public void updateNameServerAddressList(final String addrs) {
2     String[] addrArray = addrs.split(";");
3     List<String> list = Arrays.asList(addrArray);
4     this.remotingClient.updateNameServerAddressList(list);
5 }

由於名稱服務可以是集群的方式，所以在這裡用“；”進行分割，得到所有的名稱服務地址，再由remotingClient進行更新，而此時的remotingClient也就是剛纔創建的NettyRemotingClient
NettyRemotingClient的updateNameServerAddressList方法：

 1 public void updateNameServerAddressList(List<String> addrs) {
 2     List<String> old = this.namesrvAddrList.get();
 3     boolean update = false;
 4 
 5     if (!addrs.isEmpty()) {
 6         if (null == old) {
 7             update = true;
 8         } else if (addrs.size() != old.size()) {
 9             update = true;
10         } else {
11             for (int i = 0; i < addrs.size() && !update; i++) {
12                 if (!old.contains(addrs.get(i))) {
13                     update = true;
14                 }
15             }
16         }
17 
18         if (update) {
19             Collections.shuffle(addrs);
20             log.info("name server address updated. NEW : {} , OLD: {}", addrs, old);
21             this.namesrvAddrList.set(addrs);
22         }
23     }
24 }

這裡邏輯比較簡單，完成了名稱服務列表的更新

回到MQClientInstance的構造方法，做完以上操作後，又在後面創建了MQAdminImpl、PullMessageService、RebalanceService、ConsumerStatsManager以及一個新的DefaultMQProducer，關於這幾個在後面出現時再介紹

回到MQClientManager的getAndCreateMQClientInstance方法，在完成MQClientInstance的創建後，將其放入緩存中

再回到DefaultMQProducerImpl的start方法，在創建完MQClientInstance後，調用registerProducer方法
MQClientInstance的registerProducer方法：

 1 public boolean registerProducer(final String group, final DefaultMQProducerImpl producer) {
 2     if (null == group || null == producer) {
 3         return false;
 4     }
 5 
 6     MQProducerInner prev = this.producerTable.putIfAbsent(group, producer);
 7     if (prev != null) {
 8         log.warn("the producer group[{}] exist already.", group);
 9         return false;
10     }
11 
12     return true;
13 }

在MQClientInstance初始化時，會創建producerTable 、consumerTable 、topicRouteTable 、brokerAddrTable 這幾個比較重要的map

1 private final ConcurrentMap<String/* group */, MQProducerInner> producerTable = new ConcurrentHashMap<String, MQProducerInner>();
2 private