RocketMQ NameServer保障数据一致性实现方法讲解
小王曾是少年 人气:0路由注册角度
对于ZooKeeper这样的强一致性组件，使用主从分离的架构，数据只写到主节点，主从之间的数据同步通过内部机制来进行数据复制。
对于RocketMQ来说，NameServer节点之间是互相不进行通信的，这样也就无法进行数据复制。RocketMQ采用的机制是：在Broker节点启动的时候，轮询所有的NameServer节点，并与每个NameServer节点建立长连接，发送注册请求。
相应的，NameServer节点内部也会维护一个Broker列表，用来动态存储Broker的信息，做服务发现。
与此同时，Broker使用心跳机制来向所有NameServer节点证明自己是存活的，即定期发送心跳包；收到心跳包之后，NameServer节点会更新这个Broker的最新存活时间。
注意： NameServer节点在处理心跳包时，存在多个请求同时处理同一张表的情况，为了保证并发安全性，RocketMQ引入了读写锁（ReadWriteLock），保证了多个Producer并发读取路由信息不受影响，但同一时刻只能处理一个Broker发来的心跳包，这也符合读多写少的经典场景。
路由剔除
正常情况下：
如果Broker下线，则会与NameServer断开长连接，底层基于Netty的通道关闭监听器会监听到连接断开事件，然后将这个Broker信息剔除。
异常情况下：
NameServer有一个周期为10s的定时任务，定期扫描Broker表，如果超过120s没有收到某个Broker的心跳包，则会判定其失效并移除。
对于日常运维的需求，RocketMQ提供了优雅剔除路由信息的方式，即可以先禁止Broker的写权限，这样发送到这个Broker的请求都会收到一个NO_PERMISSION的响应，客户端自动重试其他的Broker。
路由发现
生产者视角：
一般是在发送第一条消息时，才会根据Topic从NameServer获取路由信息
消费者视角：
订阅的Topic一般是固定的，所以在启动时就会拉取
针对路由信息可能变化的场景，RocketMQ提供了定时拉取Topic最新路由信息的机制，以应对Broker集群发生变化的场景。
DefaultMQProducer和DefaultMQConsumer有一个pollNameServerInterval的配置项，用于指定从NameServer获取路由信息的周期，其底层依赖MQClientInstance类，MQClientInstance类中的updateTopicRouteInfoFromNameServer方法，可以根据指定的时间间隔，周期性地从NameServer里拉取路由信息。在拉取时，会将当前启动的Producer和Consumer需要用到的Topic列表放到一个集合里，逐个进行更新，源码如下：
/**
* 更新单个Topic路由信息
*/
public boolean updateTopicRouteInfoFromNameServer(final String topic) {
return updateTopicRouteInfoFromNameServer(topic, false, null);
}
/**
* 更新单个Topic路由信息
*/
public boolean updateTopicRouteInfoFromNameServer(final String topic, boolean isDefault,
DefaultMQProducer defaultMQProducer) {
try {
if (this.lockNamesrv.tryLock(LOCK_TIMEOUT_MILLIS, TimeUnit.MILLISECONDS)) {
try {
TopicRouteData topicRouteData;
if (isDefault && defaultMQProducer != null) {
// 使用默认TopicKey获取TopicRouteData
topicRouteData = this.mQClientAPIImpl.getDefaultTopicRouteInfoFromNameServer(defaultMQProducer.getCreateTopicKey(), 1000 * 3);
if (topicRouteData != null) {
for (QueueData data : topicRouteData.getQueueDatas()) {
int queueNums = Math.min(defaultMQProducer.getDefaultTopicQueueNums(), data.getReadQueueNums());
data.setReadQueueNums(queueNums);
data.setWriteQueueNums(queueNums);
}
}
} else {
topicRouteData = this.mQClientAPIImpl.getTopicRouteInfoFromNameServer(topic, 1000 * 3);
}
if (topicRouteData != null) {
TopicRouteData old = this.topicRouteTable.get(topic);
boolean changed = topicRouteDataIsChange(old, topicRouteData);
if (!changed) {
changed = this.isNeedUpdateTopicRouteInfo(topic);
} else {
log.info("the topic[{}] route info changed, old[{}] ,new[{}]", topic, old, topicRouteData);
}
if (changed) {
// 克隆出一个实例cloneTopicRouteData ： topicRouteData会被设置到下面的publishInfo/subscribeInfo
TopicRouteData cloneTopicRouteData = topicRouteData.cloneTopicRouteData();
// 更新Broker地址相关信息，当某个Broker心跳超时后，会被从brokerAddrTable中移除
for (BrokerData bd : topicRouteData.getBrokerDatas()) {
this.brokerAddrTable.put(bd.getBrokerName(), bd.getBrokerAddrs());
}
// Update Pub info
{
TopicPublishInfo publishInfo = topicRouteData2TopicPublishInfo(topic, topicRouteData);
publishInfo.setHaveTopicRouterInfo(true);
Iterator<Entry<String, MQProducerInner>> it = this.producerTable.entrySet().iterator();
while (it.hasNext()) {
Entry<String, MQProducerInner> entry = it.next();
MQProducerInner impl = entry.getValue();
if (impl != null) {
impl.updateTopicPublishInfo(topic, publishInfo);
}
}
}
// Update sub info
{
Set<MessageQueue> subscribeInfo = topicRouteData2TopicSubscribeInfo(topic, topicRouteData);
Iterator<Entry<String, MQConsumerInner>> it = this.consumerTable.entrySet().iterator();
while (it.hasNext()) {
Entry<String, MQConsumerInner> entry = it.next();
MQConsumerInner impl = entry.getValue();
if (impl != null) {
impl.updateTopicSubscribeInfo(topic, subscribeInfo);
}
}
}
log.info("topicRouteTable.put. Topic = {}, TopicRouteData[{}]", topic, cloneTopicRouteData);
this.topicRouteTable.put(topic, cloneTopicRouteData);
return true;
}
} else {
log.warn("updateTopicRouteInfoFromNameServer, getTopicRouteInfoFromNameServer return null, Topic: {}. [{}]", topic, this.clientId);
}
} catch (MQClientException e) {
if (!topic.startsWith(MixAll.RETRY_GROUP_TOPIC_PREFIX)) {
log.warn("updateTopicRouteInfoFromNameServer Exception", e);
}
} catch (RemotingException e) {
log.error("updateTopicRouteInfoFromNameServer Exception", e);
throw new IllegalStateException(e);
} finally {
this.lockNamesrv.unlock();
}
} else {
log.warn("updateTopicRouteInfoFromNameServer tryLock timeout {}ms. [{}]", LOCK_TIMEOUT_MILLIS, this.clientId);
}
} catch (InterruptedException e) {
log.warn("updateTopicRouteInfoFromNameServer Exception", e);
}
return false;
}
当Broker宕机时，还可以通过客户端的重试机制来解决，避免因为定时更新路由信息不及时导致的服务宕机~~
加载全部内容