NameServer在RocketMQ中主要承担的就是路由的管理、服务注册、以及服务的发现。在RocketMQ这承担着很重要的责任。
整体架构：
消息生产者在发送消息前需要考虑的问题就是，我需要发给谁？地址在哪儿？对于消费者也一样。那么NameServer就是用来解决这个问题的。
首先消息服务器(Broker)在启动的时候会向所有的NameServer注册自己的信息，然后消息生产者就会在发送消息前就会先从NameServer获取这些路由信息，然后选择一个合适的服务器去发送消息。
在这当中NameServer与每台Broker保持的是长连接，每隔10s就会去检测Broker是否存活，超过120s没有更新路由信息，就会将这个broker的路由信息给他剔除，broker是每隔30s就会向所有NameServer发送心跳包。同样的，客户端(消息消费者和生产者)也会每隔30s从NameServer更新路由信息。
NameServer存储了Broker的那些信息？
/**topic消息队列的路由信息，消息发送时候根据路由表进行负载均衡*/
private final HashMap<String/* topic */, List<QueueData>> topicQueueTable;
/**broker基础信息*/
private final HashMap<String/* brokerName */, BrokerData> brokerAddrTable;
/**broker集群信息，*/
private final HashMap<String/* clusterName */, Set<String/* brokerName */>> clusterAddrTable;
/**Broker状态信息，每次收到心跳包就会替换里面的信息*/
private final HashMap<String/* brokerAddr */, BrokerLiveInfo> brokerLiveTable;
/**消息过滤要用到的一个map*/
private final HashMap<String/* brokerAddr */, List<String>/* Filter Server */> filterServerTable;NameServer启动之后就会初始化一些基本信息，然后就是等待broker发送心跳包，然后处理。然后就是每隔10s去检测这些broker的路由信息，然后发现已经超过120s没有更新，就会将它剔除。NameServer还提供路由发现功能，就是支持客户端通过topic名称来查找它的路由信息。
Broker发送心跳包(即注册路由)
首先看一下Broker发送心跳包：
在broker启动的时候，会创建一个线程池每隔30s去注册自己的路由信息，也就是前面所说的发送心跳包。
/**
private final ScheduledExecutorService scheduledExecutorService = Executors.newSingleThreadScheduledExecutor(new ThreadFactoryImpl("BrokerControllerScheduledThread"));
*/
this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
@Override
public void run() {
try {
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);
//brokerConfig.getRegisterNameServerPeriod() 默认值为：1000*30再来看一下里面的registerBrokerAll(true, false, brokerConfig.isForceRegister())方法：
1.首先封装topic信息，是跟据自身的 topicConfigTable 来构建的，topicConfigTable是一个ConcurrentMap。
2.然后就是遍历topic信息，封装topic队列信息。
3.发送心跳包
public synchronized void registerBrokerAll(final boolean checkOrderConfig, boolean oneway, boolean forceRegister) {
//封装topic信息
TopicConfigSerializeWrapper topicConfigWrapper = this.getTopicConfigManager().buildTopicConfigSerializeWrapper();
//判断是否具有可读可写权限
if (!PermName.isWriteable(this.getBrokerConfig().getBrokerPermission())
|| !PermName.isReadable(this.getBrokerConfig().getBrokerPermission())) {
ConcurrentHashMap<String, TopicConfig> topicConfigTable = new ConcurrentHashMap<String, TopicConfig>();
//遍历topic
for (TopicConfig topicConfig : topicConfigWrapper.getTopicConfigTable().values()) {
//封装topic队列信息，这个对象的数据结构就跟NameServer的 QueueData 结构是一样的了
TopicConfig tmp =
new TopicConfig(topicConfig.getTopicName(), topicConfig.getReadQueueNums(), topicConfig.getWriteQueueNums(),
this.brokerConfig.getBrokerPermission());
topicConfigTable.put(topicConfig.getTopicName(), tmp);
}
topicConfigWrapper.setTopicConfigTable(topicConfigTable);
}
if (forceRegister || needRegister(this.brokerConfig.getBrokerClusterName(),
this.getBrokerAddr(),
this.brokerConfig.getBrokerName(),
this.brokerConfig.getBrokerId(),
this.brokerConfig.getRegisterBrokerTimeoutMills())) {
//注册
doRegisterBrokerAll(checkOrderConfig, oneway, topicConfigWrapper);
}
}然后再来看一下doRegisterBrokerAll方法：
这个方法主要就是去获取本机的ip地址，以及broker的配置信息
private void doRegisterBrokerAll(boolean checkOrderConfig, boolean oneway,
TopicConfigSerializeWrapper topicConfigWrapper) {
//重点是这个方法
List<RegisterBrokerResult> registerBrokerResultList = this.brokerOuterAPI.registerBrokerAll(
this.brokerConfig.getBrokerClusterName(),
this.getBrokerAddr(), //获取broker地址和端口号
this.brokerConfig.getBrokerName(),
this.brokerConfig.getBrokerId(),
this.getHAServerAddr(), //注意这里，这里就是去获取的本级的地址和IP
topicConfigWrapper,
this.filterServerManager.buildNewFilterServerList(),
oneway,
this.brokerConfig.getRegisterBrokerTimeoutMills(),
this.brokerConfig.isCompressedRegister());
if (registerBrokerResultList.size() > 0) {
RegisterBrokerResult registerBrokerResult = registerBrokerResultList.get(0);
if (registerBrokerResult != null) {
if (this.updateMasterHAServerAddrPeriodically && registerBrokerResult.getHaServerAddr() != null) {
this.messageStore.updateHaMasterAddress(registerBrokerResult.getHaServerAddr());
}
this.slaveSynchronize.setMasterAddr(registerBrokerResult.getMasterAddr());
if (checkOrderConfig) {
this.getTopicConfigManager().updateOrderTopicConfig(registerBrokerResult.getKvTable());
}
}
}
}然后再来看最后的构建请求，以及发送心跳包的方法：
首先获取所有的NameServer地址，然后封装请求头和请求体，最后通过netty将心跳包发送出去。
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();
//获取所有的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地址
for (final String namesrvAddr : nameServerAddressList) {
brokerOuterExecutor.execute(new Runnable() {
@Override
public void run() {
try {
//通过netty 发送！
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处理心跳包
NameServer处理请求的方法是在processRequest方法里的：
org.apache.rocketmq.namesrv.processor.DefaultRequestProcessor#processRequest
其中的switch case下面有这样一段代码：
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);
}方法registerBroker：
里面有一个方法，将请求解析之后，作为参数传入到registerBroker方法
RegisterBrokerResult result = this.namesrvController.getRouteInfoManager().registerBroker(
requestHeader.getClusterName(),
requestHeader.getBrokerAddr(),
requestHeader.getBrokerName(),
requestHeader.getBrokerId(),
requestHeader.getHaServerAddr(),
topicConfigWrapper,
null,
ctx.channel()
);registerBroker方法：
在这里面就可以看到更新心跳包的方法啦。可以看到更新nameServer的路由信息都在这个方法里面执行了。
更新clusterAddrTable,
更新brokerAddrTable,
更新topicQueueTable,
更新brokerLiveTable,
更新filterServerTable,
public RegisterBrokerResult registerBroker(
final String clusterName,
final String brokerAddr,
final String brokerName,
final long brokerId,
final String haServerAddr,
final TopicConfigSerializeWrapper topicConfigWrapper,
final List<String> filterServerList,
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);
}
//将broker放进去
brokerNames.add(brokerName);
boolean registerFirst = false;
//获取old brokerData
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()) {
//主节点，初次注册 更新topic理由元信息
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;
}NameServer路由删除
前面说过，NameServer会每隔10s扫描心跳包，如果超过120s就会将broker给他剔除。
NameServer启动时会启动一个线程池，没隔10s执行一次。
this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
@Override
public void run() {
NamesrvController.this.routeInfoManager.scanNotActiveBroker();
}
}, 5, 10, TimeUnit.SECONDS);再来看看scanNotActiveBroker()方法：
遍历brokerLiveTable，如果最后更新时间超过两分钟就给他移除了从brokerLiveTable。
/**
private final static long BROKER_CHANNEL_EXPIRED_TIME = 1000 * 60 * 2;
*/
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());
}
}
}然后是onChannelDestroy方法，这个方法就不细讲了，大致意思就是更新该broker的其他信息，将它的队列信息什么的给他删除干净。
NameServer理由发现
当topic信息发生变化后，是靠客户端自己主动拉取主题的最新信息，NameServer对的处理方法是：
org.apache.rocketmq.namesrv.processor.DefaultRequestProcessor#getRouteInfoByTopic
//根据主题获取对应的路由信息，然后判断是否为顺序消息
public RemotingCommand getRouteInfoByTopic(ChannelHandlerContext ctx,
RemotingCommand request) throws RemotingCommandException {
final RemotingCommand response = RemotingCommand.createResponseCommand(null);
final GetRouteInfoRequestHeader requestHeader =
(GetRouteInfoRequestHeader) request.decodeCommandCustomHeader(GetRouteInfoRequestHeader.class);
//获取主题路由信息
TopicRouteData topicRouteData = this.namesrvController.getRouteInfoManager().pickupTopicRouteData(requestHeader.getTopic());
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();
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;
}
![学习open62541 --- [76] 使用智能指针处理内存释放问题](https://img-blog.csdnimg.cn/9e4908a8eb5d4a3b97ab33163835bcc2.png)
![[CTF/网络安全] 攻防世界 PHP2 解题详析](https://img-blog.csdnimg.cn/5dd3c2799ff04a6db1c713e87932ef2b.png#pic_center)
