您的位置:首页 > 其它

RocketMQ中Producer的启动源码分析

2019-07-31 03:10 2715 查看

RocketMQ中通过DefaultMQProducer创建Producer

 

DefaultMQProducer定义如下:

public class DefaultMQProducer extends ClientConfig implements MQProducer {
protected final transient DefaultMQProducerImpl defaultMQProducerImpl;

private String createTopicKey = MixAll.AUTO_CREATE_TOPIC_KEY_TOPIC = "TBW102";

private volatile int defaultTopicQueueNums = 4;

private int sendMsgTimeout = 3000;

private int compressMsgBodyOverHowmuch = 1024 * 4;

private int retryTimesWhenSendFailed = 2;

private int retryTimesWhenSendAsyncFailed = 2;

private boolean retryAnotherBrokerWhenNotStoreOK = false;

private int maxMessageSize = 1024 * 1024 * 4; // 4M
}

其中defaultMQProducerImpl成员是Producer的具体实现,其余的一些成员是对一些参数的设置:
createTopicKey:是一个Topic值,在创建时使用,后面会说明
defaultTopicQueueNums :默认的Topic队列个数
sendMsgTimeout:发送消息超时时间
compressMsgBodyOverHowmuch:消息容量限制,超过需要进行压缩
retryTimesWhenSendFailed:同步消息发送失败的允许重发次数
retryTimesWhenSendAsyncFailed:异步消息发送失败的允许重发次数
retryAnotherBrokerWhenNotStoreOK:是否允许发送给Broker失败后,重新选择Broker发送
maxMessageSize:消息最大大小
这些属性可以通过DefaultMQProducer提供的get、set方法进行相应操作

常用的构造方法如下:

public DefaultMQProducer() {
this(MixAll.DEFAULT_PRODUCER_GROUP, null);
}

public DefaultMQProducer(final String producerGroup) {
this(producerGroup, null);
}

public DefaultMQProducer(final String producerGroup, RPCHook rpcHook) {
this.producerGroup = producerGroup;
defaultMQProducerImpl = new DefaultMQProducerImpl(this, rpcHook);
}


DefaultMQProducer继承自ClientConfig,首先会设置ClientConfig提供的更底层的参数配置:

public class ClientConfig {
public static final String SEND_MESSAGE_WITH_VIP_CHANNEL_PROPERTY = "com.rocketmq.sendMessageWithVIPChannel";

private String namesrvAddr = System.getProperty(MixAll.NAMESRV_ADDR_PROPERTY, System.getenv(MixAll.NAMESRV_ADDR_ENV));

private String clientIP = RemotingUtil.getLocalAddress();

private String instanceName = System.getProperty("rocketmq.client.name", "DEFAULT");

private int clientCallbackExecutorThreads = Runtime.getRuntime().availableProcessors();

private int pollNameServerInterval = 1000 * 30;

private int heartbeatBrokerInterval = 1000 * 30;

private boolean vipChannelEnabled = Boolean.parseBoolean(System.getProperty(SEND_MESSAGE_WITH_VIP_CHANNEL_PROPERTY, "true"));
}

其中namesrvAddr是非常重要的成员,其保存着名称服务器(Name Server)的地址,在一开始构造时会根据系统属性进行设置,若是没有设置系统属性就是null,则需要在后面通过set方法进行设置
clientIP:Producer端的本地IP
instanceName:Producer的实例名称
pollNameServerInterval :轮询NameServer的时间间隔
heartbeatBrokerInterval :向Broker发送心跳包的时间间隔
SEND_MESSAGE_WITH_VIP_CHANNEL_PROPERTY 和vipChannelEnabled:决定是否使用VIP通道,即高优先级

回到DefaultMQProducer的构造方法,其会创建DefaultMQProducerImpl实例

private final Random random = new Random();
private final ConcurrentMap<String/* topic */, TopicPublishInfo> topicPublishInfoTable =
new ConcurrentHashMap<String, TopicPublishInfo>();
private final ArrayList<SendMessageHook> sendMessageHookList = new ArrayList<SendMessageHook>();
private final RPCHook rpcHook;
protected BlockingQueue<Runnable> checkRequestQueue;
protected ExecutorService checkExecutor;
private ServiceState serviceState = ServiceState.CREATE_JUST;
private MQClientInstance mQClientFactory;
private ArrayList<CheckForbiddenHook> checkForbiddenHookList = new ArrayList<CheckForbiddenHook>();
private int zipCompressLevel = Integer.parseInt(System.getProperty(MixAll.MESSAGE_COMPRESS_LEVEL, "5"));
private MQFaultStrategy mqFaultStrategy = new MQFaultStrategy();
private final BlockingQueue<Runnable> asyncSenderThreadPoolQueue;
private final ExecutorService defaultAsyncSenderExecutor;
private ExecutorService asyncSenderExecutor;

public DefaultMQProducerImpl(final DefaultMQProducer defaultMQProducer, RPCHook rpcHook) {
this.defaultMQProducer = defaultMQProducer;
this.rpcHook = rpcHook;

this.asyncSenderThreadPoolQueue = new LinkedBlockingQueue<Runnable>(50000);
this.defaultAsyncSenderExecutor = new ThreadPoolExecutor(
Runtime.getRuntime().availableProcessors(),
Runtime.getRuntime().availableProcessors(),
1000 * 60,
TimeUnit.MILLISECONDS,
this.asyncSenderThreadPoolQueue,
new ThreadFactory() {
private AtomicInteger threadIndex = new AtomicInteger(0);

@Override
public Thread newThread(Runnable r) {
return new Thread(r, "AsyncSenderExecutor_" + this.threadIndex.incrementAndGet());
}
});
}

在构造方法中会创建一个线程池,用来处理异步消息的发送
其中有一个topicPublishInfoTable成员很重要,是一个map,保存了不同top和消息队列之间的映射,在后面详细介绍


DefaultMQProducer创建完成后,接着来看DefaultMQProducer的start方法:

public void start() throws MQClientException {
this.defaultMQProducerImpl.start();
if (null != traceDispatcher) {
try {
traceDispatcher.start(this.getNamesrvAddr());
} catch (MQClientException e) {
log.warn("trace dispatcher start failed ", e);
}
}
}

首先交给了defaultMQProducerImpl的start方法去处理

defaultMQProducerImpl的start方法:

public void start() throws MQClientException {
this.start(true);
}

public void start(final boolean startFactory) throws MQClientException {
switch (this.serviceState) {
case CREATE_JUST:
this.serviceState = ServiceState.START_FAILED;

this.checkConfig();

if (!this.defaultMQProducer.getProducerGroup().equals(MixAll.CLIENT_INNER_PRODUCER_GROUP)) {
this.defaultMQProducer.changeInstanceNameToPID();
}

this.mQClientFactory = MQClientManager.getInstance().getAndCreateMQClientInstance(this.defaultMQProducer, rpcHook);

boolean registerOK = mQClientFactory.registerProducer(this.defaultMQProducer.getProducerGroup(), this);
if (!registerOK) {
this.serviceState = ServiceState.CREATE_JUST;
throw new MQClientException("The producer group[" + this.defaultMQProducer.getProducerGroup()
+ "] has been created before, specify another name please." + FAQUrl.suggestTodo(FAQUrl.GROUP_NAME_DUPLICATE_URL),
null);
}

this.topicPublishInfoTable.put(this.defaultMQProducer.getCreateTopicKey(), new TopicPublishInfo());

if (startFactory) {
mQClientFactory.start();
}

log.info("the producer [{}] start OK. sendMessageWithVIPChannel={}", this.defaultMQProducer.getProducerGroup(),
this.defaultMQProducer.isSendMessageWithVIPChannel());
this.serviceState = ServiceState.RUNNING;
break;
case RUNNING:
case START_FAILED:
case SHUTDOWN_ALREADY:
throw new MQClientException("The producer service state not OK, maybe started once, "
+ this.serviceState
+ FAQUrl.suggestTodo(FAQUrl.CLIENT_SERVICE_NOT_OK),
null);
default:
break;
}

this.mQClientFactory.sendHeartbeatToAllBrokerWithLock();
}

在一开始DefaultMQProducerImpl实例化的时候,serviceState初始化为CREATE_JUST状态,这是一个枚举值,一共有如下几种状态:

public enum ServiceState {
CREATE_JUST,
RUNNING,
SHUTDOWN_ALREADY,
START_FAILED;

private ServiceState() {
}
}

这几个状态值很容易理解,在后面MQClientInstance中还会使用到

回到start方法,根据serviceState进行判断,只有当是CREATE_JUST状态时正常执行,防止在其他状态下错误调用start

直接看到CREATE_JUST的case部分:

this.serviceState = ServiceState.START_FAILED;

this.checkConfig();

if (!this.defaultMQProducer.getProducerGroup().equals(MixAll.CLIENT_INNER_PRODUCER_GROUP)) {
this.defaultMQProducer.changeInstanceNameToPID();
}

this.mQClientFactory = MQClientManager.getInstance().getAndCreateMQClientInstance(this.defaultMQProducer, rpcHook);

boolean registerOK = mQClientFactory.registerProducer(this.defaultMQProducer.getProducerGroup(), this);
if (!registerOK) {
this.serviceState = ServiceState.CREATE_JUST;
throw new MQClientException("The producer group[" + this.defaultMQProducer.getProducerGroup()
+ "] has been created before, specify another name please." + FAQUrl.suggestTodo(FAQUrl.GROUP_NAME_DUPLICATE_URL),
null);
}

this.topicPublishInfoTable.put(this.defaultMQProducer.getCreateTopicKey(), new TopicPublishInfo());

if (startFactory) {
mQClientFactory.start();
}

log.info("the producer [{}] start OK. sendMessageWithVIPChannel={}", this.defaultMQProducer.getProducerGroup(),
this.defaultMQProducer.isSendMessageWithVIPChannel());
this.serviceState = ServiceState.RUNNING;
break;

首先更改serviceState状态为START_FAILED,防止中途的失败

checkConfig方法是用来进行ProducerGroup命名检查:

private void checkConfig() throws MQClientException {
Validators.checkGroup(this.defaultMQProducer.getProducerGroup());

if (null == this.defaultMQProducer.getProducerGroup()) {
throw new MQClientException("producerGroup is null", null);
}

if (this.defaultMQProducer.getProducerGroup().equals(MixAll.DEFAULT_PRODUCER_GROUP)) {
throw new MQClientException("producerGroup can not equal " + MixAll.DEFAULT_PRODUCER_GROUP + ", please specify another one.",
null);
}
}

主要是检查命名的合法性,以及防止和默认的producerGroup生产者组名DEFAULT_PRODUCER_GROUP产生冲突

public static final String DEFAULT_PRODUCER_GROUP = "DEFAULT_PRODUCER";


接下来实例化mQClientFactory,这其实是生产者客户端的实例,其中MQClientManager采用单例模式,getInstance是获取MQClientManager的单例,根据ClientConfig的类型,通过getAndCreateMQClientInstance方法实例化不同属性的生产者客户端

MQClientManager:

public class MQClientManager {
private final static InternalLogger log = ClientLogger.getLog();
private static MQClientManager instance = new MQClientManager();
private AtomicInteger factoryIndexGenerator = new AtomicInteger();
private ConcurrentMap<String/* clientId */, MQClientInstance> factoryTable =
new ConcurrentHashMap<String, MQClientInstance>();

private MQClientManager() {
}

public static MQClientManager getInstance() {
return instance;
}
}

其中factoryTable是所有生产者客户端实例的map缓存,factoryIndexGenerator 是创建的每个客户端实例的流水号

getAndCreateMQClientInstance方法:

public MQClientInstance getAndCreateMQClientInstance(final ClientConfig clientConfig, RPCHook rpcHook) {
String clientId = clientConfig.buildMQClientId();
MQClientInstance instance = this.factoryTable.get(clientId);
if (null == instance) {
instance =
new MQClientInstance(clientConfig.cloneClientConfig(),
this.factoryIndexGenerator.getAndIncrement(), clientId, rpcHook);
MQClientInstance prev = this.factoryTable.putIfAbsent(clientId, instance);
if (prev != null) {
instance = prev;
log.warn("Returned Previous MQClientInstance for clientId:[{}]", clientId);
} else {
log.info("Created new MQClientInstance for clientId:[{}]", clientId);
}
}

return instance;
}

首先通过buildMQClientId方法创建clientId:

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);
}

return sb.toString();
}

clientId主要由生产者客户端的ip地址以及实例名称,根据unitName的有无,附加unitName

通过生成的clientId,在factoryTable缓存中先去获取是否创建过客户端实例
若是没有获取到,就需要实例化一个MQClientInstance
这里在实例化MQClientInstance时,并没有直接传入clientConfig,而是通过cloneClientConfig方法复制了一份,来保证安全性:

public ClientConfig cloneClientConfig() {
ClientConfig cc = new ClientConfig();
cc.namesrvAddr = namesrvAddr;
cc.clientIP = clientIP;
cc.instanceName = instanceName;
cc.clientCallbackExecutorThreads = clientCallbackExecutorThreads;
cc.pollNameServerInterval = pollNameServerInterval;
cc.heartbeatBrokerInterval = heartbeatBrokerInterval;
cc.persistConsumerOffsetInterval = persistConsumerOffsetInterval;
cc.unitMode = unitMode;
cc.unitName = unitName;
cc.vipChannelEnabled = vipChannelEnabled;
cc.useTLS = useTLS;
cc.language = language;
return cc;
}


创建MQClientInstance实例:

public MQClientInstance(ClientConfig clientConfig, int instanceIndex, String clientId, RPCHook rpcHook) {
this.clientConfig = clientConfig;
this.instanceIndex = instanceIndex;
this.nettyClientConfig = new NettyClientConfig();
this.nettyClientConfig.setClientCallbackExecutorThreads(clientConfig.getClientCallbackExecutorThreads());
this.nettyClientConfig.setUseTLS(clientConfig.isUseTLS());
this.clientRemotingProcessor = new ClientRemotingProcessor(this);
this.mQClientAPIImpl = new MQClientAPIImpl(this.nettyClientConfig, this.clientRemotingProcessor, rpcHook, clientConfig);

if (this.clientConfig.getNamesrvAddr() != null) {
this.mQClientAPIImpl.updateNameServerAddressList(this.clientConfig.getNamesrvAddr());
log.info("user specified name server address: {}", this.clientConfig.getNamesrvAddr());
}

this.clientId = clientId;

this.mQAdminImpl = new MQAdminImpl(this);

this.pullMessageService = new PullMessageService(this);

this.rebalanceService = new RebalanceService(this);

this.defaultMQProducer = new DefaultMQProducer(MixAll.CLIENT_INNER_PRODUCER_GROUP);
this.defaultMQProducer.resetClientConfig(clientConfig);

this.consumerStatsManager = new ConsumerStatsManager(this.scheduledExecutorService);

log.info("Created a new client Instance, InstanceIndex:{}, ClientID:{}, ClientConfig:{}, ClientVersion:{}, SerializerType:{}",
this.instanceIndex,
this.clientId,
this.clientConfig,
MQVersion.getVersionDesc(MQVersion.CURRENT_VERSION), RemotingCommand.getSerializeTypeConfigInThisServer());
}

可以看到MQClientInstance的构造方法创建了很多东西,就不一一说明,主要说几个重要的
其中nettyClientConfig,就很清楚的说明了RocketMQ通过Netty来进行网络之间的I/O,其保存了对Netty的一些配置
clientRemotingProcessor,用来进行消息的处理

mQClientAPIImpl则是一个非常重要的部分,直接实例化了一个MQClientAPIImpl对象:

public MQClientAPIImpl(final NettyClientConfig nettyClientConfig,
final ClientRemotingProcessor clientRemotingProcessor,
RPCHook rpcHook, final ClientConfig clientConfig) {
this.clientConfig = clientConfig;
topAddressing = new TopAddressing(MixAll.getWSAddr(), clientConfig.getUnitName());
this.remotingClient = new NettyRemotingClient(nettyClientConfig, null);
this.clientRemotingProcessor = clientRemotingProcessor;

this.remotingClient.registerRPCHook(rpcHook);
this.remotingClient.registerProcessor(RequestCode.CHECK_TRANSACTION_STATE, this.clientRemotingProcessor, null);

this.remotingClient.registerProcessor(RequestCode.NOTIFY_CONSUMER_IDS_CHANGED, this.clientRemotingProcessor, null);

this.remotingClient.registerProcessor(RequestCode.RESET_CONSUMER_CLIENT_OFFSET, this.clientRemotingProcessor, null);

this.remotingClient.registerProcessor(RequestCode.GET_CONSUMER_STATUS_FROM_CLIENT, this.clientRemotingProcessor, null);

this.remotingClient.registerProcessor(RequestCode.GET_CONSUMER_RUNNING_INFO, this.clientRemotingProcessor, null);

this.remotingClient.registerProcessor(RequestCode.CONSUME_MESSAGE_DIRECTLY, this.clientRemotingProcessor, null);
}

可以看到在这个构造方法里,首先创建了一个TopAddressing,用于以后的名称服务的寻址,其默认地址是:

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

需要通过系统属性来完成更改

接着创建了一个NettyRemotingClient,这个就是实实在在的Netty客户端

private final Bootstrap bootstrap = new Bootstrap();
// 名称服务列表
private final AtomicReference<List<String>> namesrvAddrList = new AtomicReference<List<String>>();

public NettyRemotingClient(final NettyClientConfig nettyClientConfig,
final ChannelEventListener channelEventListener) {
super(nettyClientConfig.getClientOnewaySemaphoreValue(), nettyClientConfig.getClientAsyncSemaphoreValue());
this.nettyClientConfig = nettyClientConfig;
this.channelEventListener = channelEventListener;

int publicThreadNums = nettyClientConfig.getClientCallbackExecutorThreads();
if (publicThreadNums <= 0) {
publicThreadNums = 4;
}

this.publicExecutor = Executors.newFixedThreadPool(publicThreadNums, new ThreadFactory() {
private AtomicInteger threadIndex = new AtomicInteger(0);

@Override
public Thread newThread(Runnable r) {
return new Thread(r, "NettyClientPublicExecutor_" + this.threadIndex.incrementAndGet());
}
});

this.eventLoopGroupWorker = new NioEventLoopGroup(1, new ThreadFactory() {
private AtomicInteger threadIndex = new AtomicInteger(0);

@Override
public Thread newThread(Runnable r) {
return new Thread(r, String.format("NettyClientSelector_%d", this.threadIndex.incrementAndGet()));
}
});

if (nettyClientConfig.isUseTLS()) {
try {
sslContext = TlsHelper.buildSslContext(true);
log.info("SSL enabled for client");
} catch (IOException e) {
log.error("Failed to create SSLContext", e);
} catch (CertificateException e) {
log.error("Failed to create SSLContext", e);
throw new RuntimeException("Failed to create SSLContext", e);
}
}
}

此时Netty的客户端仅仅完成了对Bootstrap的初始化,以及对NioEventLoopGroup的设置和初始化

回到MQClientInstance的构造方法,在完成MQClientAPIImpl的创建后,会根据clientConfig的getNamesrvAddr判断是否设置了namesrvAddr名称服务地址,若是设置了,需要通过mQClientAPIImpl的updateNameServerAddressList方法,完成对名称服务地址的更新:

MQClientAPIImpl的updateNameServerAddressList方法:

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

由于名称服务可以是集群的方式,所以在这里用“;”进行分割,得到所有的名称服务地址,再由remotingClient进行更新,而此时的remotingClient也就是刚才创建的NettyRemotingClient
NettyRemotingClient的updateNameServerAddressList方法:

public void updateNameServerAddressList(List<String> addrs) {
List<String> old = this.namesrvAddrList.get();
boolean update = false;

if (!addrs.isEmpty()) {
if (null == old) {
update = true;
} else if (addrs.size() != old.size()) {
update = true;
} else {
for (int i = 0; i < addrs.size() && !update; i++) {
if (!old.contains(addrs.get(i))) {
update = true;
}
}
}

if (update) {
Collections.shuffle(addrs);
log.info("name server address updated. NEW : {} , OLD: {}", addrs, old);
this.namesrvAddrList.set(addrs);
}
}
}

这里逻辑比较简单,完成了名称服务列表的更新

回到MQClientInstance的构造方法,做完以上操作后,又在后面创建了MQAdminImpl、PullMessageService、RebalanceService、ConsumerStatsManager以及一个新的DefaultMQProducer,关于这几个在后面出现时再介绍

回到MQClientManager的getAndCreateMQClientInstance方法,在完成MQClientInstance的创建后,将其放入缓存中

再回到DefaultMQProducerImpl的start方法,在创建完MQClientInstance后,调用registerProducer方法
MQClientInstance的registerProducer方法:

public boolean registerProducer(final String group, final DefaultMQProducerImpl producer) {
if (null == group || null == producer) {
return false;
}

MQProducerInner prev = this.producerTable.putIfAbsent(group, producer);
if (prev != null) {
log.warn("the producer group[{}] exist already.", group);
return false;
}

return true;
}

在MQClientInstance初始化时,会创建producerTable 、consumerTable 、topicRouteTable 、brokerAddrTable 这几个比较重要的map

private final ConcurrentMap<String/* group */, MQProducerInner> producerTable = new ConcurrentHashMap<String, MQProducerInner>();
private final ConcurrentMap<String/* group */, MQConsumerInner> consumerTable = new ConcurrentHashMap<String, MQConsumerInner>();
private final ConcurrentMap<String/* Topic */, TopicRouteData> topicRouteTable = new ConcurrentHashMap<String, TopicRouteData>();
private final ConcurrentMap<String/* Broker Name */, HashMap<Long/* brokerId */, String/* address */>> brokerAddrTable =
new ConcurrentHashMap<String, HashMap<Long, String>>();

其中MQProducerInner是接口,DefaultMQProducerImpl是其实现类,完成了以group组名称为键值的DefaultMQProducerImpl的关联
在这里就是根据group,进行DefaultMQProducerImpl的缓存,MQConsumerInner同理
topicRouteTable 则记录与Topic对应的Broker以及消息队列信息
brokerAddrTable则记录与Broker Name对应的Broker的地址列表

还是回到start方法,在完成registerProducer方法后,根据返回值registerOK,判断接下来的操作
若是失败,将serviceState置为CREATE_JUST,并报出异常,方便下一次的正常start

若是成功,则先需要向topicPublishInfoTable中添加一条键值为createTopicKey("TBW102")的TopicPublishInfo记录
TopicPublishInfo:

public class TopicPublishInfo {
private boolean orderTopic = false;
private boolean haveTopicRouterInfo = false;
private List<MessageQueue> messageQueueList = new ArrayList<MessageQueue>();
private volatile ThreadLocalIndex sendWhichQueue = new ThreadLocalIndex();
}

其中messageQueueList存放消息队列MessageQueue,sendWhichQueue 是用来获取sendWhichQueue中的下标,也就是当前所要发送的具体的消息队列

MessageQueue:

public class MessageQueue implements Comparable<MessageQueue>, Serializable {
private static final long serialVersionUID = 6191200464116433425L;
private String topic;
private String brokerName;
private int queueId;

public MessageQueue() {
}

public MessageQueue(String topic, String brokerName, int queueId) {
this.topic = topic;
this.brokerName = brokerName;
this.queueId = queueId;
}
}

可以看到这是一个简单的pojo,其封装了topic,brokerName以及queueId

ThreadLocalIndex :

public class ThreadLocalIndex {
private final ThreadLocal<Integer> threadLocalIndex = new ThreadLocal<Integer>();
private final Random random = new Random();

public int getAndIncrement() {
Integer index = this.threadLocalIndex.get();
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;
}

@Override
public String toString() {
return "ThreadLocalIndex{" +
"threadLocalIndex=" + threadLocalIndex.get() +
'}';
}
}

通过ThreadLocal,赋予每个线程一个随机值,后面会根据这个随机值通过和messageQueueList的length取余运算,选取一个MessageQueue ,进而选取一条真正的消息队列进行消息发送

再次回到DefaultMQProducerImpl的start方法,在完成createTopicKey的Topic的记录添加后,根据startFactory判断是否需要调用mQClientFactory的start方法,这里默认startFactory是true,就需要调用mQClientFactory的start方法:

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 server
if (null == this.clientConfig.getNamesrvAddr()) {
this.mQClientAPIImpl.fetchNameServerAddr();
}
// Start request-response channel
this.mQClientAPIImpl.start();
// Start various schedule tasks
this.startScheduledTask();
// Start pull service
this.pullMessageService.start();
// Start rebalance service
this.rebalanceService.start();
// Start push service
this.defaultMQProducer.getDefaultMQProducerImpl().start(false);
log.info("the client factory [{}] start OK", this.clientId);
this.serviceState = ServiceState.RUNNING;
break;
case RUNNING:
break;
case SHUTDOWN_ALREADY:
break;
case START_FAILED:
throw new MQClientException("The Factory object[" + this.getClientId() + "] has been created before, and failed.", null);
default:
break;
}
}
}

MQClientInstance在创建时其serviceState状态也是CREATE_JUST

这里首先检查名称服务地址是否设置,若是没有设置,则通过MQClientAPIImpl的fetchNameServerAddr方法,尝试自动获取名称服务
MQClientAPIImpl的fetchNameServerAddr方法:

public String fetchNameServerAddr() {
try {
String addrs = this.topAddressing.fetchNSAddr();
if (addrs != null) {
if (!addrs.equals(this.nameSrvAddr)) {
log.info("name server address changed, old=" + this.nameSrvAddr + ", new=" + addrs);
this.updateNameServerAddressList(addrs);
this.nameSrvAddr = addrs;
return nameSrvAddr;
}
}
} catch (Exception e) {
log.error("fetchNameServerAddr Exception", e);
}
return nameSrvAddr;
}

这里首先根据topAddressing的fetchNSAddr方法获取名称服务地址,若是获取到了,则判断是否需要更新名称服务列表以及原来的nameSrvAddr

topAddressing在前面说过,MQClientAPIImpl构造方法中,创建TopAddressing实例
TopAddressing的fetchNSAddr方法:

public final String fetchNSAddr() {
return fetchNSAddr(true, 3000);
}

public final String fetchNSAddr(boolean verbose, long timeoutMills) {
String url = this.wsAddr;
try {
if (!UtilAll.isBlank(this.unitName)) {
url = url + "-" + this.unitName + "?nofix=1";
}
HttpTinyClient.HttpResult result = HttpTinyClient.httpGet(url, null, null, "UTF-8", timeoutMills);
if (200 == result.code) {
String responseStr = result.content;
if (responseStr != null) {
return clearNewLine(responseStr);
} else {
log.error("fetch nameserver address is null");
}
} else {
log.error("fetch nameserver address failed. statusCode=" + result.code);
}
} catch (IOException e) {
if (verbose) {
log.error("fetch name server address exception", e);
}
}

if (verbose) {
String errorMsg =
"connect to " + url + " failed, maybe the domain name " + MixAll.getWSAddr() + " not bind in /etc/hosts";
errorMsg += FAQUrl.suggestTodo(FAQUrl.NAME_SERVER_ADDR_NOT_EXIST_URL);

log.warn(errorMsg);
}
return null;
}

首先根据wsAddr和unitName创建url,其中wsAddr在前面说过,默认是http://jmenv.tbsite.net:8080/rocketmq/nsaddr,需要通过系统属性来更改

然后通过HttpTinyClient的httpGet方法建立连接,进行GET请求,获取名称地址
HttpTinyClient的httpGet方法:

static public HttpResult httpGet(String url, List<String> headers, List<String> paramValues,
String encoding, long readTimeoutMs) throws IOException {
String encodedContent = encodingParams(paramValues, encoding);
url += (null == encodedContent) ? "" : ("?" + encodedContent);

HttpURLConnection conn = null;
try {
conn = (HttpURLConnection) new URL(url).openConnection();
conn.setRequestMethod("GET");
conn.setConnectTimeout((int) readTimeoutMs);
conn.setReadTimeout((int) readTimeoutMs);
setHeaders(conn, headers, encoding);

conn.connect();
int respCode = conn.getResponseCode();
String resp = null;

if (HttpURLConnection.HTTP_OK == respCode) {
resp = IOTinyUtils.toString(conn.getInputStream(), encoding);
} else {
resp = IOTinyUtils.toString(conn.getErrorStream(), encoding);
}
return new HttpResult(respCode, resp);
} finally {
if (conn != null) {
conn.disconnect();
}
}
}

这里就通过了JDK原生的HttpURLConnection ,完成了一次指定url的GET请求,返回请求数据,将请求到的数据以及状态码封装为HttpResult,返回给上一级调用,也就是TopAddressing的fetchNSAddr方法中,再调用clearNewLine方法,将状态码为200的数据处理(清除不必要的空客、换行、回车),得到名称地址,最后回到fetchNameServerAddr方法中,完成名称服务列表的更新,至此自动获取名称服务结束

回到MQClientInstance的start方法中:
在确定有名称服务的情况下,首先调用mQClientAPIImpl的start方法:
MQClientAPIImpl的start方法:

public void start() {
this.remotingClient.start();
}

这里实际上调用了前面所创建的Nettt客户端的start方法:
NettyRemotingClient的start方法:

public void start() {
this.defaultEventExecutorGroup = new DefaultEventExecutorGroup(
nettyClientConfig.getClientWorkerThreads(),
new ThreadFactory() {

private AtomicInteger threadIndex = new AtomicInteger(0);

@Override
public Thread newThread(Runnable r) {
return new Thread(r, "NettyClientWorkerThread_" + this.threadIndex.incrementAndGet());
}
});

Bootstrap handler = this.bootstrap.group(this.eventLoopGroupWorker).channel(NioSocketChannel.class)
.option(ChannelOption.TCP_NODELAY, true)
.option(ChannelOption.SO_KEEPALIVE, false)
.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, nettyClientConfig.getConnectTimeoutMillis())
.option(ChannelOption.SO_SNDBUF, nettyClientConfig.getClientSocketSndBufSize())
.option(ChannelOption.SO_RCVBUF, nettyClientConfig.getClientSocketRcvBufSize())
.handler(new ChannelInitializer<SocketChannel>() {
@Override
public void initChannel(SocketChannel ch) throws Exception {
ChannelPipeline pipeline = ch.pipeline();
if (nettyClientConfig.isUseTLS()) {
if (null != sslContext) {
pipeline.addFirst(defaultEventExecutorGroup, "sslHandler", sslContext.newHandler(ch.alloc()));
log.info("Prepend SSL handler");
} else {
log.warn("Connections are insecure as SSLContext is null!");
}
}
pipeline.addLast(
defaultEventExecutorGroup,
new NettyEncoder(),
new NettyDecoder(),
new IdleStateHandler(0, 0, nettyClientConfig.getClientChannelMaxIdleTimeSeconds()),
new NettyConnectManageHandler(),
new NettyClientHandler());
}
});

this.timer.scheduleAtFixedRate(new TimerTask() {
@Override
public void run() {
try {
NettyRemotingClient.this.scanResponseTable();
} catch (Throwable e) {
log.error("scanResponseTable exception", e);
}
}
}, 1000 * 3, 1000);

if (this.channelEventListener != null) {
this.nettyEventExecutor.start();
}
}

这里完成了Bootstrap对前面创建的EventLoopGroup以及handler的绑定


在完成mQClientAPIImpl的start方法后,调用startScheduledTask方法,启动定时任务
startScheduledTask方法:

private void startScheduledTask() {
if (null == this.clientConfig.getNamesrvAddr()) {
this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {

@Override
public void run() {
try {
MQClientInstance.this.mQClientAPIImpl.fetchNameServerAddr();
} catch (Exception e) {
log.error("ScheduledTask fetchNameServerAddr exception", e);
}
}
}, 1000 * 10, 1000 * 60 * 2, TimeUnit.MILLISECONDS);
}

this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {

@Override
public void run() {
try {
MQClientInstance.this.updateTopicRouteInfoFromNameServer();
} catch (Exception e) {
log.error("ScheduledTask updateTopicRouteInfoFromNameServer exception", e);
}
}
}, 10, this.clientConfig.getPollNameServerInterval(), TimeUnit.MILLISECONDS);

this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {

@Override
public void run() {
try {
MQClientInstance.this.cleanOfflineBroker();
MQClientInstance.this.sendHeartbeatToAllBrokerWithLock();
} catch (Exception e) {
log.error("ScheduledTask sendHeartbeatToAllBroker exception", e);
}
}
}, 1000, this.clientConfig.getHeartbeatBrokerInterval(), TimeUnit.MILLISECONDS);

this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {

@Override
public void run() {
try {
MQClientInstance.this.persistAllConsumerOffset();
} catch (Exception e) {
log.error("ScheduledTask persistAllConsumerOffset exception", e);
}
}
}, 1000 * 10, this.clientConfig.getPersistConsumerOffsetInterval(), TimeUnit.MILLISECONDS);

this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {

@Override
public void run() {
try {
MQClientInstance.this.adjustThreadPool();
} catch (Exception e) {
log.error("ScheduledTask adjustThreadPool exception", e);
}
}
}, 1, 1, TimeUnit.MINUTES);
}

可以看到,一共设置了五个定时任务

①若是名称服务地址namesrvAddr不存在,则调用前面的fetchNameServerAddr方法,定时更新名称服务

②通过updateTopicRouteInfoFromNameServer方法定时更新Topic所对应的路由信息:

public void updateTopicRouteInfoFromNameServer() {
Set<String> topicList = new HashSet<String>();

// Consumer
{
Iterator<Entry<String, MQConsumerInner>> it = this.consumerTable.entrySet().iterator();
while (it.hasNext()) {
Entry<String, MQConsumerInner> entry = it.next();
MQConsumerInner impl = entry.getValue();
if (impl != null) {
Set<SubscriptionData> subList = impl.subscriptions();
if (subList != null) {
for (SubscriptionData subData : subList) {
topicList.add(subData.getTopic());
}
}
}
}
}

// Producer
{
Iterator<Entry<String, MQProducerInner>> it = this.producerTable.entrySet().iterator();
while (it.hasNext()) {
Entry<String, MQProducerInner> entry = it.next();
MQProducerInner impl = entry.getValue();
if (impl != null) {
Set<String> lst = impl.getPublishTopicList();
topicList.addAll(lst);
}
}
}

for (String topic : topicList) {
this.updateTopicRouteInfoFromNameServer(topic);
}
}

将所有Consumer和Producer的Topic封装在topicList,交给updateTopicRouteInfoFromNameServer调用

updateTopicRouteInfoFromNameServer方法:

public boolean updateTopicRouteInfoFromNameServer(final String topic) {
return updateTopicRouteInfoFromNameServer(topic, false, null);
}

public boolean updateTopicRouteInfoFromNameServer(final String topic, boolean isDefault,
DefaultMQProducer defaultMQProducer) {
try {
if (this.lockNamesrv.tryLock(LOCK_TIMEOUT_MILLIS, TimeUnit.MILLISECONDS)) {
try {
TopicRouteData topicRouteData;
if (isDefault && defaultMQProducer != null) {
topicRouteData = this.mQClientAPIImpl.getDefaultTopicRouteInfoFromNameServer(defaultMQProducer.getCreateTopicKey(),
1000 * 3);
if (topicRouteData != null) {
for (QueueData data : topicRouteData.getQueueDatas()) {
int queueNums = Math.min(defaultMQProducer.getDefaultTopicQueueNums(), data.getReadQueueNums());
data.setReadQueueNums(queueNums);
data.setWriteQueueNums(queueNums);
}
}
} else {
topicRouteData = this.mQClientAPIImpl.getTopicRouteInfoFromNameServer(topic, 1000 * 3);
}
if (topicRouteData != null) {
TopicRouteData old = this.topicRouteTable.get(topic);
boolean changed = topicRouteDataIsChange(old, topicRouteData);
if (!changed) {
changed = this.isNeedUpdateTopicRouteInfo(topic);
} else {
log.info("the topic[{}] route info changed, old[{}] ,new[{}]", topic, old, topicRouteData);
}

if (changed) {
TopicRouteData cloneTopicRouteData = topicRouteData.cloneTopicRouteData();

for (BrokerData bd : topicRouteData.getBrokerDatas()) {
this.brokerAddrTable.put(bd.getBrokerName(), bd.getBrokerAddrs());
}

// Update Pub info
{
TopicPublishInfo publishInfo = topicRouteData2TopicPublishInfo(topic, topicRouteData);
publishInfo.setHaveTopicRouterInfo(true);
Iterator<Entry<String, MQProducerInner>> it = this.producerTable.entrySet().iterator();
while (it.hasNext()) {
Entry<String, MQProducerInner> entry = it.next();
MQProducerInner impl = entry.getValue();
if (impl != null) {
impl.updateTopicPublishInfo(topic, publishInfo);
}
}
}

// Update sub info
{
Set<MessageQueue> subscribeInfo = topicRouteData2TopicSubscribeInfo(topic, topicRouteData);
Iterator<Entry<String, MQConsumerInner>> it = this.consumerTable.entrySet().iterator();
while (it.hasNext()) {
Entry<String, MQConsumerInner> entry = it.next();
MQConsumerInner impl = entry.getValue();
if (impl != null) {
impl.updateTopicSubscribeInfo(topic, subscribeInfo);
}
}
}
log.info("topicRouteTable.put. Topic = {}, TopicRouteData[{}]", topic, cloneTopicRouteData);
this.topicRouteTable.put(topic, cloneTopicRouteData);
return true;
}
} else {
log.warn("updateTopicRouteInfoFromNameServer, getTopicRouteInfoFromNameServer return null, Topic: {}", topic);
}
} catch (Exception e) {
if (!topic.startsWith(MixAll.RETRY_GROUP_TOPIC_PREFIX) && !topic.equals(MixAll.AUTO_CREATE_TOPIC_KEY_TOPIC)) {
log.warn("updateTopicRouteInfoFromNameServer Exception", e);
}
} finally {
this.lockNamesrv.unlock();
}
} else {
log.warn("updateTopicRouteInfoFromNameServer tryLock timeout {}ms", LOCK_TIMEOUT_MILLIS);
}
} catch (InterruptedException e) {
log.warn("updateTopicRouteInfoFromNameServer Exception", e);
}

return false;
}

这里首先由mQClientAPIImpl的getTopicRouteInfoFromNameServer方法,从名称服务器上获取其Topic所对应的路由信息

其中Topic的路由信息由TopicRouteData进行封装:

public class TopicRouteData extends RemotingSerializable {
private String orderTopicConf;
private List<QueueData> queueDatas;
private List<BrokerData> brokerDatas;
private HashMap<String/* brokerAddr */, List<String>/* Filter Server */> filterServerTable;
}

QueueData:

public class QueueData implements Comparable<QueueData> {
private String brokerName;
private int readQueueNums;
private int writeQueueNums;
private int perm;
private int topicSynFlag;
}

BrokerData:

public class BrokerData implements Comparable<BrokerData> {
private String cluster;
private String brokerName;
private HashMap<Long/* brokerId */, String/* broker address */> brokerAddrs;
}


getTopicRouteInfoFromNameServer方法:

public TopicRouteData getTopicRouteInfoFromNameServer(final String topic, final long timeoutMillis)
throws RemotingException, MQClientException, InterruptedException {
return getTopicRouteInfoFromNameServer(topic, timeoutMillis, true);
}

public TopicRouteData getTopicRouteInfoFromNameServer(final String topic, final long timeoutMillis,
boolean allowTopicNotExist) throws MQClientException, InterruptedException, RemotingTimeoutException, RemotingSendRequestException, RemotingConnectException {
GetRouteInfoRequestHeader requestHeader = new GetRouteInfoRequestHeader();
requestHeader.setTopic(topic);

RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.GET_ROUTEINTO_BY_TOPIC, requestHeader);

RemotingCommand response = this.remotingClient.invokeSync(null, request, timeoutMillis);
assert response != null;
switch (response.getCode()) {
case ResponseCode.TOPIC_NOT_EXIST: {
if (allowTopicNotExist && !topic.equals(MixAll.AUTO_CREATE_TOPIC_KEY_TOPIC)) {
log.warn("get Topic [{}] RouteInfoFromNameServer is not exist value", topic);
}

break;
}
case ResponseCode.SUCCESS: {
byte[] body = response.getBody();
if (body != null) {
return TopicRouteData.decode(body, TopicRouteData.class);
}
}
default:
break;
}

throw new MQClientException(response.getCode(), response.getRemark());
}

这里主要通过remotingClient即Netty客户端的invokeSync方法向名称服务器发送封装好的request请求来获取response
通过名称服务器寻找与Topic相关的Broker有关路由信息,将这些信息作为response返回,在这里接收到进行处理,封装成TopicRouteData

在invokeSync方法中采用懒加载的方式,尝试获取已经建立好连接的Channel,若是没有,则需要通过bootstrap的connect方法先建立连接产生ChannelFuture,进而获取并缓存Channel

回到updateTopicRouteInfoFromNameServer,通过名称服务器获取到了有关Topic的路由信息,调用topicRouteDataIsChange方法和原来topicRouteTable保存的路由信息进行比较
topicRouteDataIsChange方法:

private boolean topicRouteDataIsChange(TopicRouteData olddata, TopicRouteData nowdata) {
if (olddata == null || nowdata == null)
return true;
TopicRouteData old = olddata.cloneTopicRouteData();
TopicRouteData now = nowdata.cloneTopicRouteData();
Collections.sort(old.getQueueDatas());
Collections.sort(old.getBrokerDatas());
Collections.sort(now.getQueueDatas());
Collections.sort(now.getBrokerDatas());
return !old.equals(now);
}

若是没有发生改变,任然要调用isNeedUpdateTopicRouteInfo方法检查是否有需要更新

isNeedUpdateTopicRouteInfo方法:

private boolean isNeedUpdateTopicRouteInfo(final String topic) {
boolean result = false;
{
Iterator<Entry<String, MQProducerInner>> it = this.producerTable.entrySet().iterator();
while (it.hasNext() && !result) {
Entry<String, MQProducerInner> entry = it.next();
MQProducerInner impl = entry.getValue();
if (impl != null) {
result = impl.isPublishTopicNeedUpdate(topic);
}
}
}

{
Iterator<Entry<String, MQConsumerInner>> it = this.consumerTable.entrySet().iterator();
while (it.hasNext() && !result) {
Entry<String, MQConsumerInner> entry = it.next();
MQConsumerInner impl = entry.getValue();
if (impl != null) {
result = impl.isSubscribeTopicNeedUpdate(topic);
}
}
}

return result;
}

分别对所有的消费者和生产者进行检查是否有需要更新有关该Topic的路由信息

当存在需要跟新的情况时,在updateTopicRouteInfoFromNameServer中
首先从topicRouteData中取出BrokerData,即Broker的路由信息,进行更新
再根据topicRouteData从中获取消费者生产者的消息路由信息,分别进行更新

③定时清除离线的Broker,以及向当前在线的Broker发送心跳包
cleanOfflineBroker清除离线的Broker:

private void cleanOfflineBroker() {
try {
if (this.lockNamesrv.tryLock(LOCK_TIMEOUT_MILLIS, TimeUnit.MILLISECONDS))
try {
ConcurrentHashMap<String, HashMap<Long, String>> updatedTable = new ConcurrentHashMap<String, HashMap<Long, String>>();

Iterator<Entry<String, HashMap<Long, String>>> itBrokerTable = this.brokerAddrTable.entrySet().iterator();
while (itBrokerTable.hasNext()) {
Entry<String, HashMap<Long, String>> entry = itBrokerTable.next();
String brokerName = entry.getKey();
HashMap<Long, String> oneTable = entry.getValue();

HashMap<Long, String> cloneAddrTable = new HashMap<Long, String>();
cloneAddrTable.putAll(oneTable);

Iterator<Entry<Long, String>> it = cloneAddrTable.entrySet().iterator();
while (it.hasNext()) {
Entry<Long, String> ee = it.next();
String addr = ee.getValue();
if (!this.isBrokerAddrExistInTopicRouteTable(addr)) {
it.remove();
log.info("the broker addr[{} {}] is offline, remove it", brokerName, addr);
}
}

if (cloneAddrTable.isEmpty()) {
itBrokerTable.remove();
log.info("the broker[{}] name's host is offline, remove it", brokerName);
} else {
updatedTable.put(brokerName, cloneAddrTable);
}
}

if (!updatedTable.isEmpty()) {
this.brokerAddrTable.putAll(updatedTable);
}
} finally {
this.lockNamesrv.unlock();
}
} catch (InterruptedException e) {
log.warn("cleanOfflineBroker Exception", e);
}
}

这里的brokerAddrTable是会通过②中的定时任务来更新,遍历其中的所有Broker信息,通过isBrokerAddrExistInTopicRouteTable方法,进行检查:

private boolean isBrokerAddrExistInTopicRouteTable(final String addr) {
Iterator<Entry<String, TopicRouteData>> it = this.topicRouteTable.entrySet().iterator();
while (it.hasNext()) {
Entry<String, TopicRouteData> entry = it.next();
TopicRouteData topicRouteData = entry.getValue();
List<BrokerData> bds = topicRouteData.getBrokerDatas();
for (BrokerData bd : bds) {
if (bd.getBrokerAddrs() != null) {
boolean exist = bd.getBrokerAddrs().containsValue(addr);
if (exist)
return true;
}
}
}

return false;
}

通过比对topicRouteTable中的所有TopicRouteData保存的BrokerAddrs来判断,若是Broker不存在,需要进行清除,进而更新brokerAddrTable

sendHeartbeatToAllBrokerWithLock定时向Broker发送心跳包:

public void sendHeartbeatToAllBrokerWithLock() {
if (this.lockHeartbeat.tryLock()) {
try {
this.sendHeartbeatToAllBroker();
this.uploadFilterClassSource();
} catch (final Exception e) {
log.error("sendHeartbeatToAllBroker exception", e);
} finally {
this.lockHeartbeat.unlock();
}
} else {
log.warn("lock heartBeat, but failed.");
}
}

这一部分就不详细介绍,主要还是通过Netty客户端完成心跳包的发送

④定时持久化消费者队列的消费进度,这个在分析消费者时再详细说明

⑤定时调整消费者端的线程池的大小,还是在分析消费者时再详细说明

startScheduledTask创建的五个定时任务结束,回到MQClientInstance的start方法
接着开启pullMessageService服务,为消费者拉取消息
然后开启rebalanceService服务,用来均衡消息队列
这两个服务在有关消费者时再介绍

接着通过:

this.defaultMQProducer.getDefaultMQProducerImpl().start(false);

开启push service服务
其中defaultMQProducer是在前面MQClientInstance构造方法中创建的

this.defaultMQProducer = new DefaultMQProducer(MixAll.CLIENT_INNER_PRODUCER_GROUP);

只不过他调用的start方法,参数为false,也就是没有调用mQClientFactory的start方法
后续会介绍其用途

到这DefaultMQProducerImpl的start方法已经基本完毕,只不过在最后,会通过mQClientFactory的sendHeartbeatToAllBrokerWithLock方法,给所有Broker发送一次心跳包

到此,Producer的启动结束

内容来自用户分享和网络整理,不保证内容的准确性,如有侵权内容,可联系管理员处理 点击这里给我发消息
标签: