上一篇文章分析了MS之間的同步，Client端在多Matser多Slave下的情況，這一篇文章主要從NameServer的角度來分析，在整個 架構(gòu)中，NameServer起了非常大的多用，MS，Client和Server之間都是必須依賴NameServer的協(xié)調(diào)。
另外，在上一篇文章開始的時候，說了3.5.8版本沒有主從切換的功能，那么這篇文章也會分析一下，為什么沒有這個功能，走起~
源碼分析
RouteInfoManager
RouteInfoManager 這個類在HA相關(guān)的功能中起到了很大的作用，其中保存了集群中的路由信息
主要字段
private final static long BrokerChannelExpiredTime = 1000 * 60 * 2;// broker連接廢棄的時間間隔
private final HashMap<String/* topic */, List<QueueData>> topicQueueTable; // topic下的隊列信息
private final HashMap<String/* brokerName */, BrokerData> brokerAddrTable; // brokerName下的broker信息
private final HashMap<String/* clusterName */, Set<String/* brokerName */>> clusterAddrTable;// clusterName下的broker地址信息
private final HashMap<String/* brokerAddr */, BrokerLiveInfo> brokerLiveTable;// 存活著的Broker信息
private final HashMap<String/* brokerAddr */, List<String>/* Filter Server */> filterServerTable;//
// ....
public class BrokerData implements Comparable<BrokerData> {
private String cluster;// 集群名稱
private String brokerName;
private HashMap<Long/* brokerId */, String/* broker address */> brokerAddrs;// master和slave的地址，master的id為0
}
class BrokerLiveInfo {
private long lastUpdateTimestamp;// 上次注冊的時間
private DataVersion dataVersion;// 版本號
private Channel channel;/
private String haServerAddr;// ha相關(guān)服務的地址
}
注冊Broker：registerBroker
注冊主要是registerBroker方法，Broker會定時請求
// 省略異常捕獲代碼等代碼
public RegisterBrokerResult registerBroker(//
final String clusterName,// 1
final String brokerAddr,// 2
final String brokerName,// 3
final long brokerId,// 4
final String haServerAddr,// 5
final TopicConfigSerializeWrapper topicConfigWrapper,// 6
final List<String> filterServerList, // 7
final Channel channel// 8
) {
RegisterBrokerResult result = new RegisterBrokerResult();
// ....
// 1. 更新clusterAddrTable：獲取當前集群下的broker地址，如果沒有則創(chuàng)建，放入當前broker的信息
Set<String> brokerNames = this.clusterAddrTable.get(clusterName);
if (null == brokerNames) {
brokerNames = new HashSet<String>();
this.clusterAddrTable.put(clusterName, brokerNames);
}
brokerNames.add(brokerName);
// 2.brokerAddrTable：獲取當前brokerName下的broker信息，如果沒有則創(chuàng)建，放入當前broker的信息
boolean registerFirst = false;
BrokerData brokerData = this.brokerAddrTable.get(brokerName);
if (null == brokerData) {// 為空表示可能第一次注冊
registerFirst = true;
brokerData = new BrokerData();
brokerData.setBrokerName(brokerName);
HashMap<Long, String> brokerAddrs = new HashMap<Long, String>();
brokerData.setBrokerAddrs(brokerAddrs);
this.brokerAddrTable.put(brokerName, brokerData);
}
// 第一次注冊分為兩種情況，1. brokerAddrTable下無該brokerName的信息 2. brokerAddrTable下無該BrokerId的地址
String oldAddr = brokerData.getBrokerAddrs().put(brokerId, brokerAddr);
registerFirst = registerFirst || (null == oldAddr);
// 3.更新topicQueueTable：topicConfigWrapper 為Broker上保存的topic信息，還有一個版本號信息
if (null != topicConfigWrapper //
&& MixAll.MASTER_ID == brokerId) {
// brokerLiveTable對應
if (this.isBrokerTopicConfigChanged(brokerAddr, topicConfigWrapper.getDataVersion())//
|| registerFirst) {
ConcurrentHashMap<String, TopicConfig> tcTable =
topicConfigWrapper.getTopicConfigTable();
if (tcTable != null) {
for(Map.Entry<String,TopicConfig> entry: tcTable.entrySet()){// 更新或者創(chuàng)建 topicQueueTable中對應的topic信息
this.createAndUpdateQueueData(brokerName, entry.getValue());
}
}
}
}
// 4. 更新brokerLiveTable：
BrokerLiveInfo prevBrokerLiveInfo = this.brokerLiveTable.put(brokerAddr, //
new BrokerLiveInfo(//
System.currentTimeMillis(), //
topicConfigWrapper.getDataVersion(),//
channel, //
haServerAddr));
if (null == prevBrokerLiveInfo) {
log.info("new broker registerd, {} HAServer: {}", brokerAddr, haServerAddr);
}
// 5. 更新filterServerTable
if (filterServerList != null) {
if (filterServerList.isEmpty()) {
this.filterServerTable.remove(brokerAddr);
} else {
this.filterServerTable.put(brokerAddr, filterServerList);
}
}
// 5.返回Matser信息：如果為slave，獲取master的相關(guān)信息并返回給slave
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);
}
}
}
// ....
return result;
}
broker注冊很簡單，就是將RouteInfoManager中幾個路由信息更新一下(無則創(chuàng)建)
處理斷開的連接：onChannelDestroy
當Broker，Consumer，或者Porducer與NameServer斷開連接的時候，會觸發(fā)該方法，然后更新路由信息
- 通過channel找到broker地址
- 更新brokerLiveTable和filterServerTable：這兩個信息key為broker的地址，所以直接remove
- 更新brokerAddrTable對應地址的信息
- 更新topicQueueTable對應地址的信息
如果Master掛了，那么觸發(fā)該方法，Master相關(guān)信息會被移除，但是除此之外并沒有其他操作，Slave無法感知，所以這時候主從沒有切換，Slave還是Slave。
其實這種事情，自己想到的一種實現(xiàn)方式就是Master掛的時候找到Slave的Channel，通知其升為Master，當然這只是自己大概想到的一個實現(xiàn)方式，不一樣合理
Consumer更新路由信息
在MQClientInstance.updateTopicRouteInfoFromNameServer方法，Consumer會定時調(diào)用，其中會向NameServer獲取topic下的路由信息(保存在RouteInfoManager中)，并保存到本地收數(shù)據(jù)結(jié)構(gòu)中，例如上一篇文章講的brokerAddrTable，topicRouteTable等
Producer獲取路由信息
Producer在發(fā)送的時候會獲取topic下的隊列信息，這個和Consumer類似，通過topic，從RouteInfoManager中獲取，調(diào)用的也是updateTopicRouteInfoFromNameServer方法
NameServer非常輕量，只是用來保存整個集群的理由信息
