Nacos注册表结构和海量服务注册与并发读写原理 源码分析
nacos作为功能强大且流行的服务注册框架，具有服务动态发现、服务配置、服务管理及流量控制等功能。今天，我们一起来了解下其服务注册原理以及如何解决高并发读写问题。
1、注册表结构
首先我们回想一下Nacos的服务模型，如下图所示。从上到下分别是：命令空间->服务分组->服务->集群->实例。各个部分对应有什么作用呢。
1、命名空间起到环境隔离的作用，比如隔离生产环境和测试环境；
2、服务分组，当服务太多可对服务进行高一层的分组，默认DEFAULT_GROUP
3、服务，比如订单服务，用户服务
4、集群，服务可以在全国各地部署几百个实例，可把杭州或上海的实例放到各自的杭州集群或上海集群中
5、实例就是真正一个部署的实例
好了，那么如果用Java数据结构来表示上面的注册关系表呢，没错，可以用嵌套的Map集合，各服务模型和Map的对应关系如下
Nacos采用数据的分级存储模型，对应到Java代码用多层Map表示。最外层ServiceMap以命名空间为key,value也是一个内层Map。内层Map以group拼接serviceName作为key,value是一个服务对象。服务对象内部也是一个ClusterMap,以集群名字作为key,value是一个集群对象。集群对象内部是一个实例集合,分为持久和临时实例集合。源码中的命名如下
/**
* ServiceManager服务管理器内部
* Map(namespace, Map(group::serviceName, Service)).
*/
Map<String, Map<String, Service>> serviceMap = new ConcurrentHashMap<>();
// service对象持有集群map
// Map<inerfaceName,Cluster>
Map<String, Cluster> clusterMap = new HashMap<>();
// 集群对象 永久实例集合
Set<Instance> persistentInstances = new HashSet<>();
// 临时实例集合
Set<Instance> ephemeralInstances = new HashSet<>();
2、数十万服务注册压力与并发写冲突的解决
核心思想：
Nacos内部接收到服务注册请求后，不是同步写数据，而是把请求包装成任务放入阻塞队列中，然后由单线程的线程池死循环从队列中取任务，异步完成实例更新，从而提高了并发写能力。
并发写冲突通过对相同的service服务加锁解决，同时由单线程的线程池处理实例更新任务也避免了并发写问题。
接下来，我们用核心源码来验证。
首先，根据命令空间id和服务名称得到对应的service对象，对service加锁，解决添加实例的时候并发写问题。接着，调用addIpAddresses方法获取最新的实例列表（包含旧和新的），再调用consistencyService的put方法（重点）真正添加实例。
public void addInstance(String namespaceId, String serviceName, boolean ephemeral, Instance... ips)
throws NacosException {
// 创建实例列表唯一Id，即服务唯一Id
String key = KeyBuilder.buildInstanceListKey(namespaceId, serviceName, ephemeral);
// 获取实例对应的服务
Service service = getService(namespaceId, serviceName);
// 添加实例到服务时，对当前服务加锁，解决添加实例并发写问题
synchronized (service) {
// 更新并返回最新的服务列表，采用复制的方式避免了并发读写问题
List<Instance> instanceList = addIpAddresses(service, ephemeral, ips);
// 写时复制技术：创建新的实例列表，再替换旧的实例列表
Instances instances = new Instances();
instances.setInstanceList(instanceList);
// 新的实例列表同步到nacos集群中
consistencyService.put(key, instances);
}
}
addIpAddresses方法调用updateIpAddresses方法，得到最新的实例列表（包含旧实例），当新实例的集群不存在时，还会顺便新建集群并添加到service服务中。如果新实例和旧实例相同，会拷贝旧的实例的id，并覆盖旧实例。
public List<Instance> updateIpAddresses(Service service, String action, boolean ephemeral, Instance... ips)
throws NacosException {
// 获取旧的实例Map
Datum datum = consistencyService
.get(KeyBuilder.buildInstanceListKey(service.getNamespaceId(), service.getName(), ephemeral));
// 当前老服务对应的实例
List<Instance> currentIPs = service.allIPs(ephemeral);
// 写时复制，创建新的Map对象
Map<String, Instance> currentInstances = new HashMap<>(currentIPs.size());
Set<String> currentInstanceIds = Sets.newHashSet();
for (Instance instance : currentIPs) { // 旧的实例列表复制新容器中
currentInstances.put(instance.toIpAddr(), instance);
currentInstanceIds.add(instance.getInstanceId());
}
Map<String, Instance> instanceMap; // 拷贝旧的实例Map
if (datum != null && null != datum.value) {
// 拷贝旧的有效实例列表
instanceMap = setValid(((Instances) datum.value).getInstanceList(), currentInstances);
} else {
instanceMap = new HashMap<>(ips.length);
}
// 把新实例和旧实例都放到服务中，如果旧实例存在以旧实例信息为准。整个过程都是新建再替换，避免了并发读写问题
for (Instance instance : ips) {
if (!service.getClusterMap().containsKey(instance.getClusterName())) {
Cluster cluster = new Cluster(instance.getClusterName(), service); // 当前集群不存在则新建,并添加到旧集群列表中
cluster.init();
service.getClusterMap().put(instance.getClusterName(), cluster);
Loggers.SRV_LOG
.warn("cluster: {} not found, ip: {}, will create new cluster with default configuration.",
instance.getClusterName(), instance.toJson());
}
if (UtilsAndCommons.UPDATE_INSTANCE_ACTION_REMOVE.equals(action)) {
instanceMap.remove(instance.getDatumKey()); // action=remove则删除实例
} else {
Instance oldInstance = instanceMap.get(instance.getDatumKey());
if (oldInstance != null) { // 旧实例存在，把旧实例的id复制给新实例
instance.setInstanceId(oldInstance.getInstanceId());
} else {
instance.setInstanceId(instance.generateInstanceId(currentInstanceIds));
}
instanceMap.put(instance.getDatumKey(), instance); // 新实例再替换老的实例
}
}
...
return new ArrayList<>(instanceMap.values());
}
consistencyService.put方法跳转到实现类DistroConsistencyServiceImpl（处理临时实例的一致性协议ditro）进行执行。
put方法分两步，本地更新和远程更新。
public void put(String key, Record value) throws NacosException {
// 利用阻塞队列本地更新
onPut(key, value);
// 同样采用异步的方式进行集群同步
distroProtocol.sync(new DistroKey(key, KeyBuilder.INSTANCE_LIST_KEY_PREFIX), DataOperation.CHANGE,
globalConfig.getTaskDispatchPeriod() / 2);
}
我们进去onPut方法，再用notifier对象调用addTask方法添加任务，那么Notifier是什么。
public void onPut(String key, Record value) {
if (KeyBuilder.matchEphemeralInstanceListKey(key)) {
Datum<Instances> datum = new Datum<>();
datum.value = (Instances) value;
datum.key = key;
datum.timestamp.incrementAndGet();
dataStore.put(key, datum); // 只是存到Map中，由其他线程异步获取
}
if (!listeners.containsKey(key)) {
return;
}
notifier.addTask(key, DataOperation.CHANGE);
}
Notifier原来是一个runnable对象，内部有一个阻塞队列tasks，调用addTask就是往阻塞队列添加任务。再看一下它的run方法，死循环从队列中获取任务并调用handle方法，异步更新实例列表。handle方法采用写时复制，把实例更新到clusterMap中，留到后面再说。那么Notifier线程什么时候开始提交执行的呢。
public class Notifier implements Runnable {
private ConcurrentHashMap<String, String> services = new ConcurrentHashMap<>(10 * 1024);
private BlockingQueue<Pair<String, DataOperation>> tasks = new ArrayBlockingQueue<>(1024 * 1024);
public void addTask(String datumKey, DataOperation action) {
if (services.containsKey(datumKey) && action == DataOperation.CHANGE) {
return;
}
if (action == DataOperation.CHANGE) {
services.put(datumKey, StringUtils.EMPTY);
}
// 往阻塞队列中添加任务
tasks.offer(Pair.with(datumKey, action));
}
@Override
public void run() {
Loggers.DISTRO.info("distro notifier started");
for (; ; ) {
try {
Pair<String, DataOperation> pair = tasks.take();
handle(pair);
} catch (Throwable e) {
Loggers.DISTRO.error("[NACOS-DISTRO] Error while handling notifying task", e);
}
}
}
...
}
在DistroConsistencyServiceImpl看到如下init方法，被PostConstruct修饰，说明在依赖注入完成后就会调用，本质是提交给一个单线程的线程池执行，即Executors.newScheduledThreadPool(1)，单线程也避免了并发写问题。
@PostConstruct
public void init() {
GlobalExecutor.submitDistroNotifyTask(notifier);
}
下面distroProtocol.sync是集群数据同步的逻辑了，找到集群所有的结点，把任务放到一个ConcurrentHashMap中，然后线程池取任务再进行同步。基于Distro模式的同步是异步进行的，并且失败时会将任务重新入队并重试，因此不保证同步结果的强一致性，属于AP模式的一致性策略。
public void sync(DistroKey distroKey, DataOperation action, long delay) {
for (Member each : memberManager.allMembersWithoutSelf()) {
DistroKey distroKeyWithTarget = new DistroKey(distroKey.getResourceKey(), distroKey.getResourceType(),
each.getAddress());
DistroDelayTask distroDelayTask = new DistroDelayTask(distroKeyWithTarget, action, delay);
distroTaskEngineHolder.getDelayTaskExecuteEngine().addTask(distroKeyWithTarget, distroDelayTask);
}
}
protected void processTasks() {
Collection<Object> keys = getAllTaskKeys();
for (Object taskKey : keys) {
AbstractDelayTask task = removeTask(taskKey);
if (null == task) {
continue;
}
NacosTaskProcessor processor = getProcessor(taskKey);
if (null == processor) {
getEngineLog().error("processor not found for task, so discarded. " + task);
continue;
}
try {
// ReAdd task if process failed 失败重试
if (!processor.process(task)) {
retryFailedTask(taskKey, task);
}
} catch (Throwable e) {
getEngineLog().error("Nacos task execute error : " + e.toString(), e);
retryFailedTask(taskKey, task);
}
}
}
3、并发读写原理
nacos在更新实例列表的时候，采用CopyOnWrite技术，先拷贝旧的实例列表，然后更新拷贝的实例列表，最后再用拷贝的列表替换旧的实例列表，就避免了并发读写问题。
我们用核心源码来验证一下。
发现更新实例列表的时候，直接把引用指向一个新new的对象，而不是在原来的实例列表进行增删改，这就解决了脏读问题。更新过程中的旧实例列表不受影响，用户依然依然可以读取。
public void updateIps(List<Instance> ips, boolean ephemeral) {
...
// 上面都是日志功能，没有实际的功能
// 将实例列表应用指向新的实例列表
toUpdateInstances = new HashSet<>(ips);
if (ephemeral) {
ephemeralInstances = toUpdateInstances;
} else {
persistentInstances = toUpdateInstances;
}
}
参考：某马