上一篇文章分析了MS之间的同步，Client端在多Matser多Slave下的情况，这一篇文章主要从NameServer的角度来分析，在整个 架构中，NameServer起了非常大的多用，MS，Client和Server之间都是必须依赖NameServer的协调。
另外，在上一篇文章开始的时候，说了3.5.8版本没有主从切换的功能，那么这篇文章也会分析一下，为什么没有这个功能，走起~
源码分析
RouteInfoManager
RouteInfoManager 这个类在HA相关的功能中起到了很大的作用，其中保存了集群中的路由信息
主要字段
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相关服务的地址
}
注册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地址，如果没有则创建，放入当前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信息，如果没有则创建，放入当前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()){// 更新或者创建 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的相关信息并返回给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中几个路由信息更新一下(无则创建)
处理断开的连接：onChannelDestroy
当Broker，Consumer，或者Porducer与NameServer断开连接的时候，会触发该方法，然后更新路由信息
- 通过channel找到broker地址
- 更新brokerLiveTable和filterServerTable：这两个信息key为broker的地址，所以直接remove
- 更新brokerAddrTable对应地址的信息
- 更新topicQueueTable对应地址的信息
如果Master挂了，那么触发该方法，Master相关信息会被移除，但是除此之外并没有其他操作，Slave无法感知，所以这时候主从没有切换，Slave还是Slave。
其实这种事情，自己想到的一种实现方式就是Master挂的时候找到Slave的Channel，通知其升为Master，当然这只是自己大概想到的一个实现方式，不一样合理
Consumer更新路由信息
在MQClientInstance.updateTopicRouteInfoFromNameServer方法，Consumer会定时调用，其中会向NameServer获取topic下的路由信息(保存在RouteInfoManager中)，并保存到本地收数据结构中，例如上一篇文章讲的brokerAddrTable，topicRouteTable等
Producer获取路由信息
Producer在发送的时候会获取topic下的队列信息，这个和Consumer类似，通过topic，从RouteInfoManager中获取，调用的也是updateTopicRouteInfoFromNameServer方法
NameServer非常轻量，只是用来保存整个集群的理由信息
