RocketMQ源码学习笔记：Producer启动流程

这是本人学习的总结，主要学习资料如下

• 马士兵教育
• rocketMq官方文档

1、Overview

这是发送信息的代码样例，

DefaultMQProducer producer = new DefaultMQProducer(PRODUCER_GROUP);
producer.setNamesrvAddr(DEFAULT_NAMESRVADDR);
producer.start();
for (int i = 0; i < MESSAGE_COUNT; i++) {try {Message msg = new Message(TOPIC, TAG, ("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET));SendResult sendResult = producer.send(msg);System.out.printf("%s%n", sendResult);} catch (Exception e) {e.printStackTrace();Thread.sleep(1000);}
}
producer.shutdown();

生产者启动最少需要两个信息，group和nameSrvAddr。启动的源码则是producer#start()中。

最终RocketMQ会创建MQClientInstance的实例，然后在调用MQClientInstance#start()完成生产者的启动。

1.1、创建MQClientInstance

1.1.1、检查

代码线索DefaultMQProducer#start() -> DefaultMQProducerImpl#start() -> DefaultMQProducerImpl#checkConfig()。

创建MQClientInstance前做前置检查，主要是检查group的格式，并且不能和系统的group重命。

1.1.1、MQClientInstance的ID

MQClientInstance由MQClientManager进行管理。MQClientManager整个JVM中只有一个实例，其内部用ConcurrentMap<String, MQClientManager>管理着所有的MQClientInstance，其中的String可以看成是每个MQClientInstance的id，下面通过源码查看id是如何组成的。

代码线索DefaultMQProducer#start() -> DefaultMQProducerImpl#start() -> MQClientManager#getInstance()#getOrCreateMQClientInstance() -> ClientConfig#buildMQClientId()。

public String buildMQClientId() {StringBuilder sb = new StringBuilder();sb.append(this.getClientIP());sb.append("@");sb.append(this.getInstanceName());if (!UtilAll.isBlank(this.unitName)) {sb.append("@");sb.append(this.unitName);}if (enableStreamRequestType) {sb.append("@");sb.append(RequestType.STREAM);}return sb.toString();
}

很明显，每个MQClientInstance的ID主要是由IP，instanceName和unitName组成，其中instanceName和unitName都可以设置。所以如果我们想要创建多个MQClientInstance使用的话，可以设置不同的instanceName和unitName。

1.2、MQClientInstance.start()

启动一些线程池，心跳服务。

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 serverif (null == this.clientConfig.getNamesrvAddr()) {this.mQClientAPIImpl.fetchNameServerAddr();}// Start request-response channel// NRC startthis.mQClientAPIImpl.start();// Start various schedule tasksthis.startScheduledTask();// Start pull servicethis.pullMessageService.start();// Start rebalance servicethis.rebalanceService.start();// Start push servicethis.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;}}
}