java開發(fā)RocketMQ之NameServer路由管理源碼分析
更新時間：2021年11月15日 11:01:10 作者：又蠢又笨的懶羊羊程序猿
這篇文章主要為大家介紹了java開發(fā)中RocketMQ之NameServer路由管理源碼分析詳解，有需要的朋友可以借鑒參考下，希望能夠有所幫助，祝大家多多進步早日升職加薪
1.前言
NameServer主要作用是為消息消費者和消息生產(chǎn)者提供關于主題Topic的路由信息，那么NameServer需要存儲路由的基本信息，還要管理Broker節(jié)點，包括路由注冊、路由刪除等。
2.路由元信息
路由元信息主要由RouteInfoManager來進行管理，這個類在上一篇文章中已經(jīng)介紹過，這里再做一下簡單的介紹。
private final static long BROKER_CHANNEL_EXPIRED_TIME = 1000 * 60 * 2; private final ReadWriteLock lock = new ReentrantReadWriteLock(); private final HashMap<String/* topic */, List<QueueData>> topicQueueTable; private final HashMap<String/* brokerName */, BrokerData> brokerAddrTable; private final HashMap<String/* clusterName */, Set<String/* brokerName */>> clusterAddrTable; private final HashMap<String/* brokerAddr */, BrokerLiveInfo> brokerLiveTable; private final HashMap<String/* brokerAddr */, List<String>/* Filter Server */> filterServerTable;
BROKER_CHANNEL_EXPIRED_TIME：NameServer與Broker空閑連接時長，在2minNameServer之內(nèi)沒有收到Broker的心跳包，則NameServer會關閉與該Broker的連接并刪除Broker的路由信息。
lock：讀寫鎖，用來保護以下用于存儲關鍵信息的非線程安全容器HashMap。
topicQueueTable：Topic消息隊列路由信息，消息發(fā)送時根據(jù)路由表進行負載均衡
brokerAddrTable：Broker基礎信息，包括brokerName、所屬集群名稱、主備Broker地址
clusterAddrTable：Broker集群信息，存儲集群中所有Broker名稱
brokerLiveTable：Broker狀態(tài)信息，NameServer每次收到心跳包會對該表進行信息維護。
filterServerTable：Broker上的FilterServer列表，用于類模式消息過濾。
RocketMQ基于定于發(fā)布機制，一個Topic擁有多個消息隊列，一個Broker為每一個主題創(chuàng)建4個讀隊列和4個寫隊列。多個Broker組成一個集群，集群由相同的多臺Broker組成Master-Slave架構，brokerId為0代表Master，大于0為Slave。BrokerLiveInfo中的lastUpdateTimestamp存儲上次收到Broker心跳包的時間。
3.路由注冊
以下是路由注冊的時序圖：
RocketMQ的路由注冊功能是通過Broker與NameServer的心跳機制實現(xiàn)的。Broker啟動時向NameServer集群發(fā)送心跳信息，每隔30s向集群所有NameServer發(fā)送心跳包，NameServer接收到之后會對brokerLiveTable中的BrokerLiveInfo的lastUpdateTimeStamp屬性值進行更新。此外NameServer會每隔10s掃描brokerLiveTable，如果檢測到有Broker連續(xù)120s沒有發(fā)送心跳包，則NameServer將移除Broker的路由信息同時關閉Socket連接。
源碼
3.1Broker路由注冊
BrokerController#start
//...進行檢查
if (!messageStoreConfig.isEnableDLegerCommitLog()) {
startProcessorByHa(messageStoreConfig.getBrokerRole());
handleSlaveSynchronize(messageStoreConfig.getBrokerRole());
//注冊Broker信息
this.registerBrokerAll(true, false, true);
}
//創(chuàng)建定時任務線程池
this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
@Override
public void run() {
try {
//每隔三十秒注冊Broker信息
BrokerController.this.registerBrokerAll(true, false, brokerConfig.isForceRegister());
} catch (Throwable e) {
log.error("registerBrokerAll Exception", e);
}
}
}, 1000 * 10, Math.max(10000, Math.min(brokerConfig.getRegisterNameServerPeriod(), 60000)), TimeUnit.MILLISECONDS);
if (this.brokerStatsManager != null) {
this.brokerStatsManager.start();
}
if (this.brokerFastFailure != null) {
this.brokerFastFailure.start();
}
}
BrokerController#registerBrokerAll
if (forceRegister || needRegister(this.brokerConfig.getBrokerClusterName(),
this.getBrokerAddr(),
this.brokerConfig.getBrokerName(),
this.brokerConfig.getBrokerId(),
this.brokerConfig.getRegisterBrokerTimeoutMills())) {
//開始注冊
doRegisterBrokerAll(checkOrderConfig, oneway, topicConfigWrapper);
}
BrokerController#doRegisterBrokerAll
//進入brokerOuterAPI#registerBrokerAll
List<RegisterBrokerResult> registerBrokerResultList = this.brokerOuterAPI.registerBrokerAll(
this.brokerConfig.getBrokerClusterName(),
this.getBrokerAddr(),
this.brokerConfig.getBrokerName(),
this.brokerConfig.getBrokerId(),
this.getHAServerAddr(),
topicConfigWrapper,
this.filterServerManager.buildNewFilterServerList(),
oneway,
this.brokerConfig.getRegisterBrokerTimeoutMills(),
this.brokerConfig.isCompressedRegister());
BrokerOuterAPI#registerBrokerAll
final List<RegisterBrokerResult> registerBrokerResultList = new CopyOnWriteArrayList<>();
//獲取nameServer地址信息
List<String> nameServerAddressList = this.remotingClient.getNameServerAddressList();
if (nameServerAddressList != null && nameServerAddressList.size() > 0) {
//進行請求頭的封裝
final RegisterBrokerRequestHeader requestHeader = new RegisterBrokerRequestHeader();
requestHeader.setBrokerAddr(brokerAddr);
requestHeader.setBrokerId(brokerId);
requestHeader.setBrokerName(brokerName);
requestHeader.setClusterName(clusterName);
requestHeader.setHaServerAddr(haServerAddr);
requestHeader.setCompressed(compressed);
//封裝注冊請求體
RegisterBrokerBody requestBody = new RegisterBrokerBody();
requestBody.setTopicConfigSerializeWrapper(topicConfigWrapper);
requestBody.setFilterServerList(filterServerList);
final byte[] body = requestBody.encode(compressed);
final int bodyCrc32 = UtilAll.crc32(body);
requestHeader.setBodyCrc32(bodyCrc32);
final CountDownLatch countDownLatch = new CountDownLatch(nameServerAddressList.size());
//遍歷nameServer地址 ---> 分別向各個nameServer注冊Broker
for (final String namesrvAddr : nameServerAddressList) {
brokerOuterExecutor.execute(new Runnable() {
@Override
public void run() {
try {
//向nameServer注冊Broker并返回注冊結果
RegisterBrokerResult result = registerBroker(namesrvAddr, oneway, timeoutMills, requestHeader, body);
if (result != null) {
registerBrokerResultList.add(result);
}
log.info("register broker[{}]to name server {} OK", brokerId, namesrvAddr);
} catch (Exception e) {
log.warn("registerBroker Exception, {}", namesrvAddr, e);
} finally {
countDownLatch.countDown();
}
}
});
}
try {
countDownLatch.await(timeoutMills, TimeUnit.MILLISECONDS);
} catch (InterruptedException e) {
}
}
return registerBrokerResultList;
BrokerOuterAPI#registerBroker
//如果是oneway發(fā)送 無需關注返回結果 直接返回null
if (oneway) {
try {
this.remotingClient.invokeOneway(namesrvAddr, request, timeoutMills);
} catch (RemotingTooMuchRequestException e) {
// Ignore
log.error("RemotingTooMuchRequestException!!!");
}
return null;
}
//返回遠程調(diào)用的響應結果
RemotingCommand response = this.remotingClient.invokeSync(namesrvAddr, request, timeoutMills);
3.2NameServer處理路由注冊
以下是NameServer處理路由注冊的時序圖：
請求發(fā)送至NameServer，會由默認的網(wǎng)絡處理解析類DefaultRequestProcessor進行處理解析，如果請求類型是REGISTER_BROKER則會將請求轉發(fā)至RouteInfoManager進行路由的注冊。
源碼
DefaultRequestProcessor#processRequest
switch (request.getCode()) {
case RequestCode.REGISTER_BROKER:
Version brokerVersion = MQVersion.value2Version(request.getVersion());
//高版本進入這個分支
if (brokerVersion.ordinal() >= MQVersion.Version.V3_0_11.ordinal()) {
return this.registerBrokerWithFilterServer(ctx, request);
} else {
return this.registerBroker(ctx, request);
}
}
DefaultRequestProcessor#registerBrokerWithFilterServer
//轉發(fā)到NameServerController中的RouteInfoManager進行路由注冊
RegisterBrokerResult result = this.namesrvController.getRouteInfoManager().registerBroker(
requestHeader.getClusterName(),
requestHeader.getBrokerAddr(),
requestHeader.getBrokerName(),
requestHeader.getBrokerId(),
requestHeader.getHaServerAddr(),
registerBrokerBody.getTopicConfigSerializeWrapper(),
registerBrokerBody.getFilterServerList(),
ctx.channel());
RouteInfoManager#registerBroker
//加入寫鎖-保護HashTable線程安全
this.lock.writeLock().lockInterruptibly();
//維護clusterAddrTable
Set<String> brokerNames = this.clusterAddrTable.get(clusterName);
if (null == brokerNames) {
brokerNames = new HashSet<String>();
this.clusterAddrTable.put(clusterName, brokerNames);
}
brokerNames.add(brokerName);
//維護brokerAddrTable
//是否是第一次注冊 默認為false
boolean registerFirst = false;
BrokerData brokerData = this.brokerAddrTable.get(brokerName);
//如果在brokerAddrTable沒有該Broker信息
if (null == brokerData) {
//設置為第一次注冊
registerFirst = true;
//創(chuàng)建對應的BrokerData信息
brokerData = new BrokerData(clusterName, brokerName, new HashMap<Long, String>());
this.brokerAddrTable.put(brokerName, brokerData);
}
//不是第一次注冊-對Broker信息進行更新
Map<Long, String> brokerAddrsMap = brokerData.getBrokerAddrs();
//Switch slave to master: first remove <1, IP:PORT> in namesrv, then add <0, IP:PORT>
//The same IP:PORT must only have one record in brokerAddrTable
Iterator<Entry<Long, String>> it = brokerAddrsMap.entrySet().iterator();
while (it.hasNext()) {
Entry<Long, String> item = it.next();
if (null != brokerAddr && brokerAddr.equals(item.getValue()) && brokerId != item.getKey()) {
it.remove();
}
}
String oldAddr = brokerData.getBrokerAddrs().put(brokerId, brokerAddr);
registerFirst = registerFirst || (null == oldAddr);
//維護topicQueueTable
if (null != topicConfigWrapper
&& MixAll.MASTER_ID == brokerId) {
if (this.isBrokerTopicConfigChanged(brokerAddr, topicConfigWrapper.getDataVersion())
|| registerFirst) {
ConcurrentMap<String, TopicConfig> tcTable =
topicConfigWrapper.getTopicConfigTable();
if (tcTable != null) {
for (Map.Entry<String, TopicConfig> entry : tcTable.entrySet()) {
//具體細節(jié)在這個方法中
this.createAndUpdateQueueData(brokerName, entry.getValue());
}
}
}
}
//維護brokerLiveTable
BrokerLiveInfo prevBrokerLiveInfo = this.brokerLiveTable.put(brokerAddr,
new BrokerLiveInfo(
System.currentTimeMillis(),
topicConfigWrapper.getDataVersion(),
channel,
haServerAddr));
if (null == prevBrokerLiveInfo) {
log.info("new broker registered, {} HAServer: {}", brokerAddr, haServerAddr);
}
//維護filterServerTable
if (filterServerList != null) {
if (filterServerList.isEmpty()) {
this.filterServerTable.remove(brokerAddr);
} else {
this.filterServerTable.put(brokerAddr, filterServerList);
}
}
if (MixAll.MASTER_ID != brokerId) {
String masterAddr = brokerData.getBrokerAddrs().get(MixAll.MASTER_ID);
if (masterAddr != null) {
BrokerLiveInfo brokerLiveInfo = this.brokerLiveTable.get(masterAddr);
if (brokerLiveInfo != null) {
result.setHaServerAddr(brokerLiveInfo.getHaServerAddr());
result.setMasterAddr(masterAddr);
}
}
}
RouteInfoManager#createAndUpdateQueueData
//創(chuàng)建并封裝新的隊列信息
QueueData queueData = new QueueData();
//Broker名稱
queueData.setBrokerName(brokerName);
//讀寫隊列個數(shù)
queueData.setWriteQueueNums(topicConfig.getWriteQueueNums());
queueData.setReadQueueNums(topicConfig.getReadQueueNums());
//操作權限
queueData.setPerm(topicConfig.getPerm());
//同步復制還是異步復制的標識
queueData.setTopicSysFlag(topicConfig.getTopicSysFlag());
//根據(jù)Topic從topicQueueTable中獲取隊列集合
List<QueueData> queueDataList = this.topicQueueTable.get(topicConfig.getTopicName());
//如果沒有獲取到
if (null == queueDataList) {
//創(chuàng)建并將新的queueData添加到集合中
queueDataList = new LinkedList<QueueData>();
queueDataList.add(queueData);
//放入topicQueueTable
this.topicQueueTable.put(topicConfig.getTopicName(), queueDataList);
log.info("new topic registered, {} {}", topicConfig.getTopicName(), queueData);
}
//queueDataList不為空
else {
boolean addNewOne = true;
Iterator<QueueData> it = queueDataList.iterator();
//對該主題已有隊列信息進行遍歷
while (it.hasNext()) {
QueueData qd = it.next();
//如果找到相同主題
if (qd.getBrokerName().equals(brokerName)) {
//已經(jīng)存在
if (qd.equals(queueData)) {
addNewOne = false;
} else {
//進行更改-先刪除后新增
log.info("topic changed, {} OLD: {} NEW: {}", topicConfig.getTopicName(), qd,
queueData);
it.remove();
}
}
}
//添加一個新的隊列信息
if (addNewOne) {
queueDataList.add(queueData);
}
}
3.3路由刪除
NameServer會每隔10s掃描brokerLiveTable狀態(tài)表，如果BrokerLive的lastUpdateTimestamp的時間戳距當前時間超過120s，則認為Broker失效，移除該Broker，關閉與Broker連接，同時更新topicQueueTable、brokerAddrTable、brokerLiveTable、filterServerTable。
刪除路由信息的兩個觸發(fā)條件：
- NameServer定期掃描brokerLiveTable檢測上次心跳包發(fā)送時間與當前系統(tǒng)的時間差，如果時間超過
120s，則需要移除該Broker路由信息 - Broker正常關閉，執(zhí)行
unregisterBroker
3.3.1Broker異常關閉
NamesrvController#initialize
//定時任務線程池--->每隔十秒掃描活躍狀態(tài)異常的Broker信息
this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
/**
* 對Not Active Broker 進行掃描
*/
@Override
public void run() {
//在routeInfoManager中進行路由刪除
NamesrvController.this.routeInfoManager.scanNotActiveBroker();
}
}, 5, 10, TimeUnit.SECONDS);
RouteInfoManager#scanNotActiveBroker
//遍歷brokerLiveTable 查看是否有broker狀態(tài)長時間沒有更新
Iterator<Entry<String, BrokerLiveInfo>> it = this.brokerLiveTable.entrySet().iterator();
//遍歷brokerLiveTable
while (it.hasNext()) {
Entry<String, BrokerLiveInfo> next = it.next();
long last = next.getValue().getLastUpdateTimestamp();
//如果收到心跳包的時間間距超過120s
if ((last + BROKER_CHANNEL_EXPIRED_TIME) < System.currentTimeMillis()) {
//關閉連接
RemotingUtil.closeChannel(next.getValue().getChannel());
//移除broker
it.remove();
log.warn("The broker channel expired, {} {}ms", next.getKey(), BROKER_CHANNEL_EXPIRED_TIME);
//維護路由表
this.onChannelDestroy(next.getKey(), next.getValue().getChannel());
}
}
RouteInfoManager#onChannelDestroy
//待刪除的BrokerAddress
String brokerAddrFound = null;
if (channel != null) {
try {
try {
//加入讀鎖
this.lock.readLock().lockInterruptibly();
Iterator<Entry<String, BrokerLiveInfo>> itBrokerLiveTable =
this.brokerLiveTable.entrySet().iterator();
//遍歷查找待刪除的Broker
while (itBrokerLiveTable.hasNext()) {
Entry<String, BrokerLiveInfo> entry = itBrokerLiveTable.next();
if (entry.getValue().getChannel() == channel) {
brokerAddrFound = entry.getKey();
break;
}
}
} finally {
//釋放鎖
this.lock.readLock().unlock();
}
} catch (Exception e) {
log.error("onChannelDestroy Exception", e);
}
}
//加入寫鎖 //根據(jù)brokerAddrFound從brokerLiveTable和filterServerTable移除對應的Broker信息 this.lock.writeLock().lockInterruptibly(); this.brokerLiveTable.remove(brokerAddrFound); this.filterServerTable.remove(brokerAddrFound);
//維護brokerAddrTable
String brokerNameFound = null;
boolean removeBrokerName = false;
Iterator<Entry<String, BrokerData>> itBrokerAddrTable =
this.brokerAddrTable.entrySet().iterator();
//遍歷brokerAddrTable
while (itBrokerAddrTable.hasNext() && (null == brokerNameFound)) {
BrokerData brokerData = itBrokerAddrTable.next().getValue();
//遍歷brokerAddress
Iterator<Entry<Long, String>> it = brokerData.getBrokerAddrs().entrySet().iterator();
while (it.hasNext()) {
Entry<Long, String> entry = it.next();
Long brokerId = entry.getKey();
String brokerAddr = entry.getValue();
//遍歷到匹配的broker
if (brokerAddr.equals(brokerAddrFound)) {
brokerNameFound = brokerData.getBrokerName();
//根據(jù)broker地址移除broker
it.remove();
log.info("remove brokerAddr[{}, {}] from brokerAddrTable, because channel destroyed",
brokerId, brokerAddr);
break;
}
}
//如果當前BrokerData中的Broker列表為空
if (brokerData.getBrokerAddrs().isEmpty()) {
removeBrokerName = true;
//從BrokerAddrTable移除該BrokerData
itBrokerAddrTable.remove();
log.info("remove brokerName[{}] from brokerAddrTable, because channel destroyed",
brokerData.getBrokerName());
}
}
//維護clusterAddrTable
if (brokerNameFound != null && removeBrokerName) {
Iterator<Entry<String, Set<String>>> it = this.clusterAddrTable.entrySet().iterator();
//遍歷clusterAddrTable
while (it.hasNext()) {
Entry<String, Set<String>> entry = it.next();
String clusterName = entry.getKey();
Set<String> brokerNames = entry.getValue();
//根據(jù)brokerName刪除cluster集群中的broker并返回是否刪除成功
boolean removed = brokerNames.remove(brokerNameFound);
if (removed) {
log.info("remove brokerName[{}], clusterName[{}] from clusterAddrTable, because channel destroyed",
brokerNameFound, clusterName);
//如果cluster已經(jīng)不包含任何broker則移除該集群
if (brokerNames.isEmpty()) {
log.info("remove the clusterName[{}] from clusterAddrTable, because channel destroyed and no broker in this cluster",
clusterName);
it.remove();
}
break;
}
}
}
//維護TopicQueueTable
if (removeBrokerName) {
Iterator<Entry<String, List<QueueData>>> itTopicQueueTable =
this.topicQueueTable.entrySet().iterator();
//遍歷TopicQueueTable
while (itTopicQueueTable.hasNext()) {
Entry<String, List<QueueData>> entry = itTopicQueueTable.next();
//主題
String topic = entry.getKey();
//主題對應的Broker信息
List<QueueData> queueDataList = entry.getValue();
Iterator<QueueData> itQueueData = queueDataList.iterator();
//遍歷Broker
while (itQueueData.hasNext()) {
QueueData queueData = itQueueData.next();
//通過brokerNameFound在queueData中查找對應的Broker信息
if (queueData.getBrokerName().equals(brokerNameFound)) {
//移除該Broker
itQueueData.remove();
log.info("remove topic[{} {}], from topicQueueTable, because channel destroyed",
topic, queueData);
}
}
//如果當前主題已經(jīng)沒有任何Broker，則移除該主題Topic
if (queueDataList.isEmpty()) {
itTopicQueueTable.remove();
log.info("remove topic[{}] all queue, from topicQueueTable, because channel destroyed",
topic);
}
}
}
//釋放寫鎖
finally {
this.lock.writeLock().unlock();
}
3.3.2Broker正常關閉
關閉入口-BrokerController
//BrokerController#shutdown
this.unregisterBrokerAll();
//BrokerController#unregisterBrokerAll
this.brokerOuterAPI.unregisterBrokerAll(
this.brokerConfig.getBrokerClusterName(),
this.getBrokerAddr(),
this.brokerConfig.getBrokerName(),
this.brokerConfig.getBrokerId());
BrokerOuterAPI#unregisterBrokerAll
//獲取NameServerAddress列表
List<String> nameServerAddressList = this.remotingClient.getNameServerAddressList();
if (nameServerAddressList != null) {
//遍歷NameServer
for (String namesrvAddr : nameServerAddressList) {
try {
//注銷指定Broker
this.unregisterBroker(namesrvAddr, clusterName, brokerAddr, brokerName, brokerId);
log.info("unregisterBroker OK, NamesrvAddr: {}", namesrvAddr);
} catch (Exception e) {
log.warn("unregisterBroker Exception, {}", namesrvAddr, e);
}
}
}
BrokerOuterAPI#unregisterBroker
//封裝注銷Broker請求頭
UnRegisterBrokerRequestHeader requestHeader = new UnRegisterBrokerRequestHeader();
requestHeader.setBrokerAddr(brokerAddr);
requestHeader.setBrokerId(brokerId);
requestHeader.setBrokerName(brokerName);
requestHeader.setClusterName(clusterName);
//封裝注銷請求
RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.UNREGISTER_BROKER, requestHeader);
//遠程調(diào)用并發(fā)送請求-返回響應結果
RemotingCommand response = this.remotingClient.invokeSync(namesrvAddr, request, 3000);
assert response != null;
switch (response.getCode()) {
case ResponseCode.SUCCESS: {
return;
}
default:
break;
}
throw new MQBrokerException(response.getCode(), response.getRemark(), brokerAddr);
忽略Netty遠程調(diào)用過程
DefaultRequestProcess#processRequest
//根據(jù)請求狀態(tài)碼-進行不同的處理
switch (request.getCode()) {
case RequestCode.UNREGISTER_BROKER:
return this.unregisterBroker(ctx, request);
}
DefaultRequestProcess#unregisterBroker
final RemotingCommand response = RemotingCommand.createResponseCommand(null);
//解碼注銷Broker的請求頭
final UnRegisterBrokerRequestHeader requestHeader =
(UnRegisterBrokerRequestHeader) request.decodeCommandCustomHeader(UnRegisterBrokerRequestHeader.class);
//轉發(fā)至RouteInfoManager注銷Broker
this.namesrvController.getRouteInfoManager().unregisterBroker(
requestHeader.getClusterName(),
requestHeader.getBrokerAddr(),
requestHeader.getBrokerName(),
requestHeader.getBrokerId());
//設置響應狀態(tài)碼
response.setCode(ResponseCode.SUCCESS);
response.setRemark(null);
//返回響應結果
return response;
RouteInfoManager#unregisterBroker
try {
try {
//加入寫鎖-保護HashMap線程安全
this.lock.writeLock().lockInterruptibly();
//由于是從BrokerController遠程調(diào)用來的請求-故不對BrokerAddr進行檢查-直接進行移除
//移除brokerLiveTable中的Broker信息
BrokerLiveInfo brokerLiveInfo = this.brokerLiveTable.remove(brokerAddr);
log.info("unregisterBroker, remove from brokerLiveTable {}, {}",
brokerLiveInfo != null ? "OK" : "Failed",
brokerAddr
);
//維護filterServerTable
this.filterServerTable.remove(brokerAddr);
boolean removeBrokerName = false;
//根據(jù)brokername獲取對應的BrokerData信息
BrokerData brokerData = this.brokerAddrTable.get(brokerName);
if (null != brokerData) {
//刪除對應Broker信息
String addr = brokerData.getBrokerAddrs().remove(brokerId);
log.info("unregisterBroker, remove addr from brokerAddrTable {}, {}",
addr != null ? "OK" : "Failed",
brokerAddr
);
//如果當前BrokerData中已經(jīng)沒有其他Broker
if (brokerData.getBrokerAddrs().isEmpty()) {
//從BrokerAddrTable移除相應的Broker
this.brokerAddrTable.remove(brokerName);
log.info("unregisterBroker, remove name from brokerAddrTable OK, {}",
brokerName
);
removeBrokerName = true;
}
}
if (removeBrokerName) {
//在clusterAddrTable中通過集群名稱獲取對應的Broker集合
Set<String> nameSet = this.clusterAddrTable.get(clusterName);
if (nameSet != null) {
//移除該集合中的Broker
boolean removed = nameSet.remove(brokerName);
log.info("unregisterBroker, remove name from clusterAddrTable {}, {}",
removed ? "OK" : "Failed",
brokerName);
//如果該集群中的Broker集合為空-移除該集群
if (nameSet.isEmpty()) {
this.clusterAddrTable.remove(clusterName);
log.info("unregisterBroker, remove cluster from clusterAddrTable {}",
clusterName
);
}
}
//移除該Broker下的所有主題Topic
this.removeTopicByBrokerName(brokerName);
}
} finally {
//釋放寫鎖
this.lock.writeLock().unlock();
}
} catch (Exception e) {
log.error("unregisterBroker Exception", e);
}
3.4路由發(fā)現(xiàn)
RocketMQ路由發(fā)現(xiàn)是非實時的，當Topic路由出現(xiàn)變化后，NameServer不會主動推送給客戶端，而是由客戶端定時拉取主題最新的路由。
DefaultRequestProcessor#processRequest
switch (request.getCode()) {
case RequestCode.GET_ROUTEINFO_BY_TOPIC:
return this.getRouteInfoByTopic(ctx, request);
}
DefaultRequestProcessor#getRouteInfoByTopic
//遠程調(diào)用返回的響應結果
final RemotingCommand response = RemotingCommand.createResponseCommand(null);
//解析獲取路由信息的請求頭
final GetRouteInfoRequestHeader requestHeader =
(GetRouteInfoRequestHeader) request.decodeCommandCustomHeader(GetRouteInfoRequestHeader.class);
//調(diào)用RouteInfoManager中的方法,從路由表topicQueueTable、brokerAddrTable、filterServerTable中
//分別填充TopicRouteData的List<QueueData>、List<BrokerData>、filterServer
TopicRouteData topicRouteData = this.namesrvController.getRouteInfoManager().pickupTopicRouteData(requestHeader.getTopic());
//如果找到主題對應你的路由信息并且該主題為順序消息
//則從NameServer KVConfig中獲取關于順序消息相關的配置填充路由信息
if (topicRouteData != null) {
if (this.namesrvController.getNamesrvConfig().isOrderMessageEnable()) {
String orderTopicConf =
this.namesrvController.getKvConfigManager().getKVConfig(NamesrvUtil.NAMESPACE_ORDER_TOPIC_CONFIG,
requestHeader.getTopic());
topicRouteData.setOrderTopicConf(orderTopicConf);
}
byte[] content = topicRouteData.encode();
//填充響應體和狀態(tài)碼
response.setBody(content);
response.setCode(ResponseCode.SUCCESS);
response.setRemark(null);
return response;
}
//設置響應碼-主題不存在
response.setCode(ResponseCode.TOPIC_NOT_EXIST);
response.setRemark("No topic route info in name server for the topic: " + requestHeader.getTopic()
+ FAQUrl.suggestTodo(FAQUrl.APPLY_TOPIC_URL));
return response;
RouteInfoManager#pickupTopicRouteData
//創(chuàng)建主題路由信息對象
TopicRouteData topicRouteData = new TopicRouteData();
//是否找到對應隊列的標識
boolean foundQueueData = false;
//是否找到對應Broker的標識
boolean foundBrokerData = false;
Set<String> brokerNameSet = new HashSet<String>();
List<BrokerData> brokerDataList = new LinkedList<BrokerData>();
topicRouteData.setBrokerDatas(brokerDataList);
HashMap<String, List<String>> filterServerMap = new HashMap<String, List<String>>();
topicRouteData.setFilterServerTable(filterServerMap);
try {
try {
//加入讀鎖
this.lock.readLock().lockInterruptibly();
//從topicQueueTable中通過主題topic查找Broker列表
List<QueueData> queueDataList = this.topicQueueTable.get(topic);
if (queueDataList != null) {
//填充Broker信息到topicRouteData.QueueDatas
topicRouteData.setQueueDatas(queueDataList);
foundQueueData = true;
Iterator<QueueData> it = queueDataList.iterator();
while (it.hasNext()) {
QueueData qd = it.next();
brokerNameSet.add(qd.getBrokerName());
}
//遍歷Broker集合
for (String brokerName : brokerNameSet) {
//從brokerAddrTable中通過BrokerName進行查找
BrokerData brokerData = this.brokerAddrTable.get(brokerName);
if (null != brokerData) {
BrokerData brokerDataClone = new BrokerData(brokerData.getCluster(), brokerData.getBrokerName(), (HashMap<Long, String>) brokerData.getBrokerAddrs().clone());
//填充BrokerData信息到brokerDatas
brokerDataList.add(brokerDataClone);
foundBrokerData = true;
for (final String brokerAddr : brokerDataClone.getBrokerAddrs().values()) {
List<String> filterServerList = this.filterServerTable.get(brokerAddr);
filterServerMap.put(brokerAddr, filterServerList);
}
}
}
}
} finally {
//釋放讀鎖
this.lock.readLock().unlock();
}
} catch (Exception e) {
log.error("pickupTopicRouteData Exception", e);
}
log.debug("pickupTopicRouteData {} {}", topic, topicRouteData);
if (foundBrokerData && foundQueueData) {
return topicRouteData;
}
return null;
3.5總結
路由注冊：Broker啟動之后會將自己的信息注冊到NameServer，NameServer收到Broker心跳包之后會更新BrokerLiveTable中對應的Broker的相關信息，特別記錄lastUpdateTimestamp。
心跳機制：Broker會每隔30s向NameServer發(fā)送心跳包，告訴NameServer自己的運行狀態(tài)良好。
路由發(fā)現(xiàn)：NameServer中的主題Topic信息更新之后，NameServer并不會實時將數(shù)據(jù)推送給訂閱方Producer，而需要Producer自行通過Topic來查詢路由信息。
路由刪除：
- Broker異常宕機：NameServer每隔10s掃描BrokerLiveTable，檢查表中各個Broker最新的發(fā)送心跳包時間，如果這個時間與當前系統(tǒng)時間差超過120s，則該Broker會被視作已掉線，NameServer會將這個Broker從表中刪除，并且更新各個表中與該Broker有關的所有信息。
- Broker正常下線：NameServer接受到BrokerController發(fā)送來的
UNREGISTER_BROKER請求之后，會執(zhí)行上述步驟，對路由信息進行更新。
本文僅作為個人學習使用，如有不足或者錯誤請指正！
以上就是java開發(fā)RocketMQ之NameServer路由管理源碼分析的詳細內(nèi)容，更多關于RocketMQ之NameServer路由管理的資料請關注腳本之家其它相關文章！
相關文章
這篇文章主要給大家介紹了關于Java HashMap自動排序的簡單剖析，文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值，需要的朋友們下面隨著小編來一起學習學習吧2020-09-09
這篇文章主要給大家介紹了在Java中如何避免寫嵌套if樣式的相關資料，文中通過示例代碼介紹的非常詳細，對大家具有一定的參考學習價值，需要的朋友們下面跟著小編一起來學習學習吧。2017-07-07
SpringBoot+WebSocket+Netty實現(xiàn)消息推送的示例代碼
這篇文章主要介紹了SpringBoot+WebSocket+Netty實現(xiàn)消息推送的示例代碼，文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值，需要的朋友們下面隨著小編來一起學習學習吧2020-04-04
Java ArrayList如何實現(xiàn)生成不重復隨機數(shù)
這篇文章主要介紹了Java ArrayList如何實現(xiàn)生成不重復隨機數(shù),文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值,需要的朋友可以參考下2020-09-09
javaweb實現(xiàn)百度GPS定位接口（經(jīng)緯度）
這篇文章主要介紹了javaweb實現(xiàn)百度GPS定位接口（經(jīng)緯度），文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值，需要的朋友們下面隨著小編來一起學習學習吧2020-02-02
這篇文章主要介紹了Java使用excel工具類導出對象功能,結合實例形式分析了java創(chuàng)建及導出Excel數(shù)據(jù)的具體步驟與相關操作技巧,需要的朋友可以參考下2017-10-10
這篇文章主要介紹了springmvc九大組件中的HandlerMapping解析,HandlerMapping表示的是一個URL與一個Handler(可以簡單的理解為Controller中有@RequestMapping注解的方法)之間的映射關系,需要的朋友可以參考下2023-09-09
IntelliJ IDEA 是一款優(yōu)秀的 Java 集成開發(fā)環(huán)境,它提供了許多強大的功能和快捷鍵,可以幫助開發(fā)者提高編碼效率和質(zhì)量,本文就在為你介紹博主常用的五個IntelliJ IDEA使用技巧,希望能夠給你帶來一些工作效率上的提升2023-10-10
這篇文章主要介紹了Java無法輸出中文問題及解決方案，具有很好的參考價值，希望對大家有所幫助。如有錯誤或未考慮完全的地方，望不吝賜教2022-12-12
SpringBoot集成Druid連接池進行SQL監(jiān)控的問題解析
這篇文章主要介紹了SpringBoot集成Druid連接池進行SQL監(jiān)控的問題解析,在SpringBoot工程中引入Druid連接池非常簡單，本文通過實例代碼給大家介紹的非常詳細，需要的朋友可以參考下2021-07-07