RocketMQ路由中心NameServer
阿新 • • 發佈：2020-06-24
1. 前言
NameServer 是整個Rocketmq 的大腦，是為了控制多臺訊息伺服器，解決單點宕機問題問題而設計的。
2. NameServer啟動流程
從NameServer啟動類org.apache.rocketmq.namesrv.NameSrvStartup可以看出，NameServer的啟動第一步先載入NameServerConfig配置。
final NamesrvConfig namesrvConfig = new NamesrvConfig();
final NettyServerConfig nettyServerConfig = new 第二步，建立NamesrvController例項，並開啟兩個定時任務：
- 每隔10s掃描一次Broker，移除處於不啟用的Broker；
- 每隔10s列印一次KV配置。
this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
@Override
public void run() {
NamesrvController.this.routeInfoManager.scanNotActiveBroker();
}
},5,10,TimeUnit.SECONDS);
this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
@Override
public void run() {
NamesrvController.this.kvConfigManager.printAllPeriodically();
}
},1,TimeUnit.MINUTES);
複製程式碼第三步，註冊鉤子函式，啟動伺服器並監聽Broker。
3. 路由註冊、路由刪除
RocketMQ路由註冊和刪除都是通過Broker與Nameserver的心跳功能實現的。Broker啟動後，每隔30s向NameServer傳送心跳包，NameServer收到心跳包後會更新收到此 Broker 的的“最新更新時間” lastUpdateTimestamp，然後每隔10s掃描一此，如果連續120s沒有收到心跳包，NameServer 將會移除此Broker的路由資訊，同時關閉socket連線。
3.1 路由註冊
傳送心跳包
//BrokerController::start
this.registerBrokerAll(true,false,true);
this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
@Override
public void run() {
try {
BrokerController.this.registerBrokerAll(true,brokerConfig.isForceRegister());
} catch (Throwable e) {
log.error("registerBrokerAll Exception",e);
}
}
},1000 * 10,Math.max(10000,Math.min(brokerConfig.getRegisterNameServerPeriod(),60000)),//每隔30s傳送一次心跳包
TimeUnit.MILLISECONDS);
//BrokerOuterAPI::registerBrokerAll
//遍歷NameServer列表，Broker依次向NameServer傳送心跳包
public List<RegisterBrokerResult> registerBrokerAll(
final String clusterName,final String brokerAddr,final String brokerName,final long brokerId,final String haServerAddr,final TopicConfigSerializeWrapper topicConfigWrapper,final List<String> filterServerList,final boolean oneway,final int timeoutMills,final boolean compressed) {
final List<RegisterBrokerResult> registerBrokerResultList = Lists.newArrayList();
List<String> nameServerAddressList = this.remotingClient.getNameServerAddressList();
if (nameServerAddressList != null && nameServerAddressList.size() > 0) {
//封裝請求頭
final RegisterBrokerRequestHeader requestHeader = new RegisterBrokerRequestHeader();
requestHeader.setBrokerAddr(brokerAddr);//broker地址
requestHeader.setBrokerId(brokerId);//0,1 0:master 1:slave
requestHeader.setBrokerName(brokerName);//broker名稱
requestHeader.setClusterName(clusterName);//叢集名稱
requestHeader.setHaServerAddr(haServerAddr);//master地址
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);
//此處引入同步計數器，fanally減數，如堵塞，將等待
final CountDownLatch countDownLatch = new CountDownLatch(nameServerAddressList.size());
for (final String namesrvAddr : nameServerAddressList) {
brokerOuterExecutor.execute(new Runnable() {
@Override
public void run() {
try {
//註冊
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;
}
複製程式碼NameServer 處理心跳包
處理心跳包流程：
- 註冊加鎖RouteInfoManager::registerBroker
- 維護BrokerData資訊RouteInfoManager::registerBroker
- 註冊或更新路由資訊RouteInfoManager::createAndUpdateQueueData
- 更新BrokerLiveInfo（存活Broker資訊表）RouteInfoManager::registerBroker
- 註冊Broker的過濾器Server列表
public RegisterBrokerResult registerBroker(
final String clusterName,final Channel channel) {
RegisterBrokerResult result = new RegisterBrokerResult();
try {
try {
this.lock.writeLock().lockInterruptibly();
Set<String> brokerNames = this.clusterAddrTable.get(clusterName);
if (null == brokerNames) {
brokerNames = new HashSet<String>();
this.clusterAddrTable.put(clusterName,brokerNames);
}
brokerNames.add(brokerName);
boolean registerFirst = false;
BrokerData brokerData = this.brokerAddrTable.get(brokerName);
if (null == brokerData) {
registerFirst = true;
brokerData = new BrokerData(clusterName,brokerName,new HashMap<Long,String>());
this.brokerAddrTable.put(brokerName,brokerData);
}
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);
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()) {
this.createAndUpdateQueueData(brokerName,entry.getValue());
}
}
}
}
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);
}
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);
}
}
}
} finally {
this.lock.writeLock().unlock();
}
} catch (Exception e) {
log.error("registerBroker Exception",e);
}
return result;
}
private void createAndUpdateQueueData(final String brokerName,final TopicConfig topicConfig) {
QueueData queueData = new QueueData();
queueData.setBrokerName(brokerName);
queueData.setWriteQueueNums(topicConfig.getWriteQueueNums());
queueData.setReadQueueNums(topicConfig.getReadQueueNums());
queueData.setPerm(topicConfig.getPerm());
queueData.setTopicSynFlag(topicConfig.getTopicSysFlag());
List<QueueData> queueDataList = this.topicQueueTable.get(topicConfig.getTopicName());
if (null == queueDataList) {
queueDataList = new LinkedList<QueueData>();
queueDataList.add(queueData);
this.topicQueueTable.put(topicConfig.getTopicName(),queueDataList);
log.info("new topic registered,{} {}",topicConfig.getTopicName(),queueData);
} else {
boolean addNewOne = true;
Iterator<QueueData> it = queueDataList.iterator();
while (it.hasNext()) {
QueueData qd = it.next();
if (qd.getBrokerName().equals(brokerName)) {
if (qd.equals(queueData)) {
addNewOne = false;
} else {
log.info("topic changed,{} OLD: {} NEW: {}",qd,queueData);
it.remove();
}
}
}
if (addNewOne) {
queueDataList.add(queueData);
}
}
}
複製程式碼3.2 路由刪除
上面講到，NameServer會每隔10s掃描brokerLiveTable狀態表，如果BrokerLive的lastUpdateTimeStamp的時間戳距當前時間超過120s，則認為Broker失效，移除此broker，關閉socket連線，同時更新topicQueueTable、brokerAddrTable、brokerLiveTable、filterServerTable。
//RouteInfoManager::scanNotActiveBroker掃描
public void scanNotActiveBroker() {
Iterator<Entry<String,BrokerLiveInfo>> it = this.brokerLiveTable.entrySet().iterator();
while (it.hasNext()) {
Entry<String,BrokerLiveInfo> next = it.next();
long last = next.getValue().getLastUpdateTimestamp();
if ((last + BROKER_CHANNEL_EXPIRED_TIME) < System.currentTimeMillis()) {
RemotingUtil.closeChannel(next.getValue().getChannel());
it.remove();
log.warn("The broker channel expired,{} {}ms",next.getKey(),BROKER_CHANNEL_EXPIRED_TIME);
this.onChannelDestroy(next.getKey(),next.getValue().getChannel());
}
}
}
複製程式碼觸發刪除路由，除了上面的一種方法，還有一種，Broker正常關閉情況下，會執行unregisterBroker進行刪除。
3.3 路由發現
路由發現是非實時的，當topic路由出現變化，NameServer不主動推送給客戶端，而是有客戶端定時拉去主題最新的路由，這裡是為了速度而採用的容錯機制。
小結&參考資料
小結
看原始碼真的能體會到啥叫奇妙深刻，多看原始碼，理解原理，學得真的會深刻很多。+U +U