分布式服务SOA架构体系的服务注册中心主要提供服务调用的解析服务，指引服务调用方（消费者）找到“远方”的服务提供者，完成网络通信，那么RocketMQ的路由中心存储的是什么数据呢？作为一款高性能的消息中间件，如何避免NameServer的单点故障，提供高可用性呢？让我们带着这些疑问，一起进入RocketMQ NameServer的精彩世界。本节主要包括两个方面：
- NameServer的整体架构设计。
- NameServer动态路由发现与剔除机制。
1. NameServer架构设计
RocketMQ的逻辑部署如图所示：
NameServer与每台Broker服务器保持长连接，并间隔10s检测Broker是否存活，如果检测到Broker宕机，则从路由注册表中将其移除，但是路由变化不会马上通知消息生产者。为什么要这样设计呢？这是为了降低NameServer实现的复杂性，因此需要在消息发送端提供容错机制来保证消息发送的高可用性，这部分会在后面的文章中详细介绍。
NameServer本身的高可用性可通过部署多台NameServer服务器来实现，但彼此之间互不通信。虽然NameServer服务器之间在某一时刻的数据并不会完全相同，但对消息发送不会造成重大影响，无非就是短暂造成消息发送不均衡，这也是RocketMQ NameServer设计的一个亮点。
消息客户端与NameServer、Broker的交互设计要点如下：
- Broker每隔30s向NameServer集群的每一台机器发送心跳包，包含自身创建的topic路由等信息。
- 消息客户端每隔30s向NameServer更新对应topic的路由信息。
- NameServer收到Broker发送的心跳包时会记录时间戳。
- NameServer每隔10s会扫描一次brokerLiveTable（存放心跳包的时间戳信息），如果在120s内没有收到心跳包，则认为Broker失效，更新topic的路由信息，将失效的Broker信息移除。
2. NameServer路由注册、故障剔除
NameServer的主要作用是为消息生产者和消息消费者提供关于topic的路由信息，那么NameServer就需要存储路由的基础信息，并且能够管理Broker节点，包括路由注册、路由删除等功能。
2.1. 路由元信息
NameServer的路由实现类是org.apache.rocketmq.namesrv.routeinfo.RouteInfoManager。在了解路由注册之前，我们先看一下NameServer到底存储了哪些信息，如下所示：
private final HashMap<String/* topic */, Map<String /* brokerName */ , 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;
- topicQueueTable: topic消息队列的路由信息，消息发送时根据路由表进行负载均衡。
- brokerAddrTable: Broker基础信息，包含brokerName、所属集群名称、主备Broker地址。
- clusterAddrTable: Broker集群信息，存储集群中所有Broker的名称。
- brokerLiveTable: Broker状态信息，NameServer每次收到心跳包时会替换该信息。
- filterServerTable: Broker上的FilterServer列表，用于类模式消息过滤。类模式过滤机制在4.4及以后版本被废弃。
RocketMQ 2主2从数据结构如图所示：
运行时数据结构如下图所示：
2.2 路由注册
RocketMQ路由注册是通过Broker与NameServer的心跳功能实现的。Broker启动时向集群中所有的NameServer发送心跳语句，每隔30s向集群中所有的NameServer发送心跳包，NameServer收到Broker心跳包时会先更新brokerLiveTable缓存中BrokerLiveInfo的lastUpdateTimestamp，然后每隔10s扫描一次brokerLiveTable，如果连续120s没有收到心跳包，NameServer将移除该Broker的路由信息，同时关闭Socket连接。
2.2.1 Broker发送心跳包核心代码
BrokerController#start，启动定时任务：
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);
BrokerOuterAPI#registerBrokerAll，遍历所有nameServer，依次向NameServer发送心跳包：
for (final String namesrvAddr : nameServerAddressList) {
brokerOuterExecutor.execute(() -> {
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();
}
});
}
接下来就是封装API请求了，这里就不详细介绍了，有兴趣的话，可以参考源码。
2.2.2 NameServer处理心跳包
RouteInfoManager#registerBroker 分为5步，已经通过注释标注了，并且省略了部分代码：
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 {
//1. 加写锁
this.lock.writeLock().lockInterruptibly();
//如果clusterName不存在，则创建一个HashSet，put进map
Set<String> brokerNames = this.clusterAddrTable.computeIfAbsent(clusterName, k -> new HashSet<>());
brokerNames.add(brokerName);
boolean registerFirst = false;
//2. 维护brokerData信息
BrokerData brokerData = this.brokerAddrTable.get(brokerName);
if (null == brokerData) {
registerFirst = true;
brokerData = new BrokerData(clusterName, brokerName, new HashMap<>());
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()) {
log.debug("remove entry {} from brokerData", item);
it.remove();
}
}
// ....
//3. 维护 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()) {
this.createAndUpdateQueueData(brokerName, entry.getValue());
}
}
}
}
//4. 更新 BrokerLiveInfo
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);
}
//5. 注册 Broker 的过滤器 Server 地址列表
if (filterServerList != null) {
if (filterServerList.isEmpty()) {
this.filterServerTable.remove(brokerAddr);
} else {
this.filterServerTable.put(brokerAddr, filterServerList);
}
}
//...
} finally {
this.lock.writeLock().unlock();
}
} catch (Exception e) {
log.error("registerBroker Exception", e);
}
return result;
}
设计亮点：NameServer与Broker保持长连接，Broker的状态信息存储在brokerLive-Table中，NameServer每收到一个心跳包，将更新brokerLiveTable中关于Broker的状态信息以及路由表(topicQueueTable、brokerAddrTable、brokerLiveTable、filterServer-Table)。更新上述路由表(HashTable)使用了锁粒度较少的读写锁，允许多个消息发送者并发读操作，保证消息发送时的高并发。同一时刻NameServer只处理一个Broker心跳包，多个心跳包请求串行执行，这也是读写锁经典的使用场景。
2.3 路由删除
NameServer会每隔10s扫描一次brokerLiveTable状态表，如果BrokerLive的lastUpdate-Timestamp时间戳距当前时间超过120s，则认为Broker失效，移除该Broker，关闭与Broker的连接，同时更新topicQueueTable、brokerAddrTable、brokerLiveTable、filterServerTable。RocketMQ有两个触发点来触发路由删除操作：
- NameServer定时扫描brokerLiveTable，检测上次心跳包与当前系统时间的时间戳，如果时间戳大于120s，则需要移除该Broker信息。
- Broker在正常关闭的情况下，会执行unregisterBroker指令。
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());
}
}
}
我们知道scanNotActiveBroker在NameServer中每10s执行一次。逻辑也很简单，先遍历brokerLiveInfo路由表(HashMap)，检测BrokerLiveInfo的LastUpdateTimestamp上次收到心跳包的时间，如果超过120s，则认为该Broker已不可用，然后将它移除并关闭连接，最后删除与该Broker相关的路由信息。
路由表维护过程需要申请写锁，其核心处理逻辑与路由注册相反（把添加操作改成删除操作），但是步骤相似，核心代码在RouteInfoManager # onChannelDestroy中，这里不再赘述了。
2.4 路由发现
RocketMQ路由发现是非实时的，当topic路由出现变化后，NameServer不主动推送给客户端，而是由客户端定时拉取主题最新的路由。
路由发现实体类TopicRouteData核心属性如下所示：
private String orderTopicConf;
private List<QueueData> queueDatas;
private List<BrokerData> brokerDatas;
private HashMap<String/* brokerAddr */, List<String>/* Filter Server */> filterServerTable;
NameServer路由发现实现类为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;
}
3. 小结
本章主要介绍了NameServer路由功能，包含路由元数据、路由注册与发现机制。路由发现机制用图所示：
NameServer路由发现与删除机制就介绍到这里了，我们会发现这种设计存在这样一种情况：NameServer需要等Broker失效至少120s才能将该Broker从路由表中移除，如果在Broker故障期间，消息生产者根据主题获取到的路由信息包含已经宕机的Broker，就会导致消息发送失败。那么这种情况怎么办，岂不是消息发送不是高可用的？让我们带着这个疑问进入RocketMQ消息发送的学习。
