RocketMQ-消息发送源码分析(一)生产者启动流程

在介绍Mq生产者启动流程之前，先抛出几个个问题，带着问题去分析源码

1、消息生产者启动时具体做了什么？
2、消息生产者启动时和NameServer，Broker之间建立了什么关系？

先从启动入库着手，消息生产者启动入库方法在org.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl#start(boolean)中，直接看代码

启动步骤分为六步，下面逐个说明

1. 第一步，判断生产者producerGroup是否合法，没啥好说的，均为group相关参数校验

2. 第二步，将默认的生产者实例名(“DEFAULT”)称替换为进程pid，代码如下
   

   为什么要把实例名instanceName换成进程pid呢？在第三步中做说明

3. 第三步，通过MQClientManager创建MQClientInstance实例，MQClientManager是一个饿汉单例模式的实现，factorytable的Map中缓存了MQClientInstance的信息，key是ClientId，ClientId怎么来的下面有分析
   

   在看下是如何创建MQClientInstance的
   
   关键点在于buildMQClientId方法，再看下具体实现，可以看到clientId的格式是：ip@instanceName@unitName，unitName不是必须的，默认就是ip@instanceName
   

现在明白为什么第二步中instanceName要设为端口号了么？

通常一个机器上会部署多个MQ相关应用，假设存在A、B两个MQ应用，生产者启动时如果clientId都等于ip@instanceName(默认为”DEFAULT“)，则两个应用的clientId是一样的，会造成混淆，所以每个JVM进程对应一个ClientId，同时一个ClientId又对应一个MQClientInstance

所以:如果一个应用中存在多个Producer的话，都是复用一个MQClientInstance的，因为其ClientId都是一样的

那么问题又来了，提了这么多遍的MQClientInstance，MQClientInstance究竟是干嘛的呢？简单看下其属性，大家应该就应该可以联想到其功能，生产者通过MQClientInstance拉取路由信息并缓存于topicRouteTable中，生产者与broker建立连接并缓存broker信息，消费者通过其中的pullMessageService实现消息拉取，通过rebalanceService实现消息负载，还有ScheduledExecutorService，说明肯定有各种定时器的启动

4. 第四步：将当前生产者对象注册到MQClientInstance中，就是将当前producer信息存储于MQClientInstance的producerTable中，一个group只能注册一次
   
5. 第五步：调用MQClientInstance.start()方法启动MQClientInstance，看下start方法的具体实现

启动时先调用http请求拉取nameServer地址。

再启动启动MQClientAPIImpl，MQClientAPIImpl是封装了Netty请求的一个API调用实现类，其中实现了路由信息拉取、消息发送，topic创建等Netty请求，是一个统一的Netty请求调用类。

消息拉取及负载实现不在本篇讨论范围，后续再做研究、分享

启动各种定时器，看下都有哪些定时器

MQClientInstance的启动到定时器启动这也结束了，开头提到的两个问题也有了答案

消息生产者启动时具体做了什么？

1、使用进程id替换instanceName
2、创建MQClientInstance(对应clientId为：ip@pid)
3、启动MQClientInstance，MQClientInstance启动时又启动了Netty的请求处理类mQClientAPIImpl，同时启动定时拉取NameServer线程，定时拉取路由信息线程，定时发送broker心跳请求线程，定时持久化消费offset线程等定时器，也启动了消息拉取、负载等service

消息生成者启动时和NameServer，Broker之间建立了什么关系？
1、生产者启动时与某一个nameServer建立长连接，并定时从该nameServer拉取topic对应路由信息
2、生产者启动时与所有Broker建立长连接，并定时发送心跳请求，定时清除本地无效broker信息