源碼版本 4.6.0
先看一個(gè)簡(jiǎn)單消息發(fā)送的例子：
public static void main(String[] args) throws MQClientException, InterruptedException {
DefaultMQProducer producer = new DefaultMQProducer("ProducerGroupName");
producer.start();
for (int i = 0; i < 128; i++)
try {
{
Message msg = new Message("TopicTest",
"TagA",
"OrderID188",
"Hello world".getBytes(RemotingHelper.DEFAULT_CHARSET));
SendResult sendResult = producer.send(msg);
System.out.printf("%s%n", sendResult);
}
} catch (Exception e) {
e.printStackTrace();
}
producer.shutdown();
}
在進(jìn)行消息發(fā)送火本，即producer.send(msg)之前抓狭，需要啟動(dòng)Producer伦泥，可以猜想下在啟動(dòng)Prodducer中完成了消息發(fā)送端初始化操作胸竞，本文就是對(duì)初始化進(jìn)行分析斥季。
org.apache.rocketmq.client.producer.DefaultMQProducer#start
public void start() throws MQClientException {
// 設(shè)置生產(chǎn)者組
this.setProducerGroup(withNamespace(this.producerGroup));
// 核心初始化方法
this.defaultMQProducerImpl.start();
if (null != traceDispatcher) {
try {
// 消息軌跡相關(guān)
traceDispatcher.start(this.getNamesrvAddr(), this.getAccessChannel());
} catch (MQClientException e) {
log.warn("trace dispatcher start failed ", e);
}
}
}
org.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl#start(boolean)
public void start(final boolean startFactory) throws MQClientException {
switch (this.serviceState) {
case CREATE_JUST:
this.serviceState = ServiceState.START_FAILED;
// 檢查生產(chǎn)者組名是否符合規(guī)范训桶，不為空且不為默認(rèn)組名 DEFAULT_PRODUCER
this.checkConfig();
// 更換生產(chǎn)者實(shí)例名稱(chēng)，這個(gè)待①說(shuō)明
if (!this.defaultMQProducer.getProducerGroup().equals(MixAll.CLIENT_INNER_PRODUCER_GROUP)) {
this.defaultMQProducer.changeInstanceNameToPID();
}
// 獲取MQ客戶端工廠酣倾，注意這個(gè)MQClientManager是單例模式的舵揭，②補(bǔ)充
this.mQClientFactory = MQClientManager.getInstance().getOrCreateMQClientInstance(this.defaultMQProducer, rpcHook);
// 注冊(cè)該生產(chǎn)者，③處說(shuō)明
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);
}
// 如果注冊(cè)成功躁锡，則加入自動(dòng)創(chuàng)建主題的內(nèi)置Topic
this.topicPublishInfoTable.put(this.defaultMQProducer.getCreateTopicKey(), new TopicPublishInfo());
// 啟動(dòng)客戶端 ④處補(bǔ)充
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;
}
// 開(kāi)始發(fā)送心跳 ⑤
this.mQClientFactory.sendHeartbeatToAllBrokerWithLock();
// 掃描過(guò)期請(qǐng)求 ⑥
this.timer.scheduleAtFixedRate(new TimerTask() {
@Override
public void run() {
try {
RequestFutureTable.scanExpiredRequest();
} catch (Throwable e) {
log.error("scan RequestFutureTable exception", e);
}
}
}, 1000 * 3, 1000);
}
①處說(shuō)明：
public void changeInstanceNameToPID() {
if (this.instanceName.equals("DEFAULT")) {
this.instanceName = String.valueOf(UtilAll.getPid());
}
}
在未設(shè)置系統(tǒng)參數(shù)rocketmq.client.name的時(shí)候午绳，默認(rèn)instanceName為DEFAULT，如果未進(jìn)行設(shè)置映之，則設(shè)置為進(jìn)程ID拦焚，即啟動(dòng)JVM進(jìn)程的ID
為啥這么干蜡坊？我的思考點(diǎn)主要是以下兩點(diǎn)：
- 保證同一個(gè)JVM中，獲取的mQClientFactory只有一份赎败，獲取mQClientFactory的參數(shù)是以Instance拼接的字符串秕衙，如果Instance保持一致，就可保證在同一個(gè)JVM中僵刮，只會(huì)創(chuàng)建一個(gè)客戶端工廠据忘。這個(gè)有什么好處，首先mQClientFactory中包含了網(wǎng)絡(luò)組件搞糕，定時(shí)任務(wù)組件勇吊，消息拉取組件等，如果都是依據(jù)創(chuàng)建一個(gè)實(shí)例就獲取一個(gè)新的實(shí)例工廠窍仰，那么在JVM中可能存在多套相同的功能組件萧福，這樣即造成了資源浪費(fèi)，也可能使得一些內(nèi)部任務(wù)執(zhí)行錯(cuò)亂辈赋。
- 不同JVM中的生產(chǎn)者實(shí)例能區(qū)別開(kāi)
②處說(shuō)明：
首先MQClientManager是單例的鲫忍，也就是一個(gè)JVM中只會(huì)存在一個(gè)實(shí)例，接著看getOrCreateMQClientInstance方法钥屈，首先構(gòu)建實(shí)例ID：
String clientId = clientConfig.buildMQClientId();
|
|
v
public String buildMQClientId() {
StringBuilder sb = new StringBuilder();
// 獲取客戶端IP
sb.append(this.getClientIP());
sb.append("@");
// 拼裝InstanceName,一般情況下就是進(jìn)程ID
sb.append(this.getInstanceName());
// 設(shè)置unitName 一般為空悟民，可在Producer上設(shè)置
if (!UtilAll.isBlank(this.unitName)) {
sb.append("@");
sb.append(this.unitName);
}
return sb.toString();
}
總的來(lái)說(shuō)clientId = IP + @ + instanceName + unitName,接著就拿這個(gè)clientId去緩存中尋找，如果沒(méi)有篷就，就進(jìn)行創(chuàng)建射亏。
主要實(shí)例化的組件包含這幾個(gè)：
- mQClientAPIImpl （Netty通訊組件）
- pullMessageService （消息拉取組件）
- rebalanceService （重平衡組件）
- consumerStatsManager （消費(fèi)信息統(tǒng)計(jì)組件）
順帶說(shuō)一句，DefaultMQProducer和DefaultMQProducerImpl的關(guān)系竭业，可以這么理解智润，兩者之間互相包含，DefaultMQProducer繼承了ClientConfig未辆，更相當(dāng)于一個(gè)實(shí)例自定義配置類(lèi)的角色窟绷，DefaultMQProducerImpl實(shí)現(xiàn)MQProducerInner，消息發(fā)送主要邏輯是在這里面完成的咐柜。
③處說(shuō)明：
注冊(cè)該主題及對(duì)應(yīng)的生產(chǎn)者實(shí)例兼蜈，也就是在Map中放入該數(shù)據(jù)，即：
MQProducerInner prev = this.producerTable.putIfAbsent(group, producer);
if (prev != null) {
log.warn("the producer group[{}] exist already.", group);
return false;
}
注意這里用的是putIfAbsent拙友，如果生產(chǎn)者重復(fù)啟動(dòng)为狸，或者組名相同的生產(chǎn)者啟動(dòng)，都會(huì)注冊(cè)失敗遗契，觸發(fā)警告辐棒，并啟動(dòng)失敗。
拋出的異常：
new MQClientException("The producer group[" + this.defaultMQProducer.getProducerGroup()
+ "] has been created before, specify another name please." + FAQUrl.suggestTodo(FAQUrl.GROUP_NAME_DUPLICATE_URL),
null);
④處說(shuō)明：
獲取的客戶端實(shí)例啟動(dòng)，這個(gè)是真正的啟動(dòng)工作線程
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
// 如果Producer未設(shè)置nameServer地址漾根，則進(jìn)行遠(yuǎn)端拉取
if (null == this.clientConfig.getNamesrvAddr()) {
this.mQClientAPIImpl.fetchNameServerAddr();
}
// Start request-response channel
// 通訊組件啟動(dòng)
this.mQClientAPIImpl.start();
// Start various schedule tasks
// 定時(shí)任務(wù)啟動(dòng)
this.startScheduledTask();
// Start pull service
// 拉取線程啟動(dòng)
this.pullMessageService.start();
// Start rebalance service
// 重平衡啟動(dòng)
this.rebalanceService.start();
// Start push service
// 啟動(dòng)生產(chǎn)客戶端
this.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;
}
}
}
4.1 注意當(dāng)Producer為配置NameServer地址的時(shí)候泰涂，則進(jìn)行遠(yuǎn)端拉取，這個(gè)作用相當(dāng)大立叛，這個(gè)就讓線上環(huán)境動(dòng)態(tài)對(duì)NameServer擴(kuò)容负敏，遷移成為可能
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;
}
根據(jù)設(shè)置的NameServer路由拉取地址進(jìn)行拉取贡茅，地址拼接如下：
public static String getWSAddr() {
String wsDomainName = System.getProperty("rocketmq.namesrv.domain", DEFAULT_NAMESRV_ADDR_LOOKUP);
String wsDomainSubgroup = System.getProperty("rocketmq.namesrv.domain.subgroup", "nsaddr");
String wsAddr = "http://" + wsDomainName + ":8080/rocketmq/" + wsDomainSubgroup;
if (wsDomainName.indexOf(":") > 0) {
wsAddr = "http://" + wsDomainName + "/rocketmq/" + wsDomainSubgroup;
}
return wsAddr;
}
4.2 定時(shí)任務(wù)啟動(dòng)
private void startScheduledTask() {
if (null == this.clientConfig.getNamesrvAddr()) {
this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
@Override
public void run() {
try {
// 拉取NameSever地址 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 {
// 更新Topic路由地址
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 {
// 清除下線Broker秘蛇，發(fā)送心跳
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 {
// 定時(shí)持久化消費(fèi)進(jìn)度，對(duì)于廣播模式很重要
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 {
// 調(diào)整線程池 空實(shí)現(xiàn)顶考，沒(méi)啥用
MQClientInstance.this.adjustThreadPool();
} catch (Exception e) {
log.error("ScheduledTask adjustThreadPool exception", e);
}
}
}, 1, 1, TimeUnit.MINUTES);
}
總結(jié)下：
- 拉取NameServer地址赁还，延時(shí)10S,頻率2min
- 更新主題路由信息，延時(shí)10ms,頻率30S
- 向Broker發(fā)送心跳驹沿，延時(shí)1S,頻率30S
- 消費(fèi)進(jìn)度持久化艘策，延時(shí)1S,頻率5S
- 動(dòng)態(tài)調(diào)整線程池，不起作用
值得注意的是渊季，定時(shí)任務(wù)線程池是單線程無(wú)界隊(duì)列類(lèi)型的朋蔫，且用的FixedRate模式，實(shí)際的執(zhí)行頻率可能不是準(zhǔn)確的却汉，有興趣可以看下ScheduledExecutorService源碼
4.3 其余的組件啟動(dòng)和消費(fèi)相關(guān)驯妄，這里先不深入了
⑤處說(shuō)明：
this.mQClientFactory.sendHeartbeatToAllBrokerWithLock();
啟動(dòng)成功后開(kāi)始發(fā)送心跳，心跳發(fā)送的過(guò)程是持有鎖的合砂，個(gè)人感覺(jué)主要是避免心跳混亂青扔，特殊用途暫時(shí)沒(méi)聯(lián)想到。
心跳發(fā)送主代碼：
private void sendHeartbeatToAllBroker() {
// 準(zhǔn)備心跳發(fā)送包翩伪，主要是消費(fèi)訂閱配置和生產(chǎn)者配置等信息微猖，這個(gè)后續(xù)再詳細(xì)討論
final HeartbeatData heartbeatData = this.prepareHeartbeatData();
final boolean producerEmpty = heartbeatData.getProducerDataSet().isEmpty();
final boolean consumerEmpty = heartbeatData.getConsumerDataSet().isEmpty();
if (producerEmpty && consumerEmpty) {
log.warn("sending heartbeat, but no consumer and no producer");
return;
}
// 獲取所有的Broker地址
if (!this.brokerAddrTable.isEmpty()) {
long times = this.sendHeartbeatTimesTotal.getAndIncrement();
Iterator<Entry<String, HashMap<Long, String>>> it = this.brokerAddrTable.entrySet().iterator();
while (it.hasNext()) {
Entry<String, HashMap<Long, String>> entry = it.next();
String brokerName = entry.getKey();
HashMap<Long, String> oneTable = entry.getValue();
if (oneTable != null) {
for (Map.Entry<Long, String> entry1 : oneTable.entrySet()) {
Long id = entry1.getKey();
String addr = entry1.getValue();
if (addr != null) {
if (consumerEmpty) {
if (id != MixAll.MASTER_ID)
continue;
}
try {
// 發(fā)送心跳，超時(shí)3S
int version = this.mQClientAPIImpl.sendHearbeat(addr, heartbeatData, 3000);
if (!this.brokerVersionTable.containsKey(brokerName)) {
this.brokerVersionTable.put(brokerName, new HashMap<String, Integer>(4));
}
// 更新版本號(hào) this.brokerVersionTable.get(brokerName).put(addr, version);
if (times % 20 == 0) {
log.info("send heart beat to broker[{} {} {}] success", brokerName, id, addr);
log.info(heartbeatData.toString());
}
} catch (Exception e) {
if (this.isBrokerInNameServer(addr)) {
log.info("send heart beat to broker[{} {} {}] failed", brokerName, id, addr, e);
} else {
log.info("send heart beat to broker[{} {} {}] exception, because the broker not up, forget it", brokerName,
id, addr, e);
}
}
}
}
}
}
}
}
這塊值得注意的是：
- 如果當(dāng)前JVM中只有生產(chǎn)者實(shí)例缘屹，那么只向主節(jié)點(diǎn)發(fā)送心跳凛剥。
- 如果當(dāng)前JVM即存在生產(chǎn)者又存在消費(fèi)者，那么就向所有節(jié)點(diǎn)發(fā)送心跳轻姿。這個(gè)和消息發(fā)送邏輯当悔，消息消費(fèi)邏輯有關(guān)，后期再談踢代。
從這也可以學(xué)到盲憎，只要涉及到網(wǎng)絡(luò)請(qǐng)求，請(qǐng)加上超時(shí)胳挎，為了你的服務(wù)穩(wěn)定饼疙！
⑤處說(shuō)明：
移除過(guò)期請(qǐng)求，這個(gè)requestFutureTable的填充涉及的API：
org.apache.rocketmq.client.producer.DefaultMQProducer#request(org.apache.rocketmq.common.message.Message, long)
從方法說(shuō)明上看是發(fā)送消息，在等到該消息消費(fèi)后再返回窑眯，提供異步和同步模式的API屏积，改API在生產(chǎn)上沒(méi)實(shí)際用過(guò)，關(guān)于這個(gè)就不過(guò)多講解了磅甩。
最后提供一個(gè)總圖：
啟動(dòng)流程-Producer.png
