Nacos源码之注册中心的实现详解
leo的跟班 人气:0引言
在平时的工作中多多少少都会接触到注册中心，当你的应用从单机到拆分成多个服务，每个服务又有多个实例的情况时，那么对服务IP地址管理的要求就会越来越高。而注册中心就是干这个的。
最经典的注册中心实现方式是Zookeeper，在很多RPC框架中都有基于Zookeeper注册中心的实现，如Dubbo,Motan。有兴趣的可以直接去阅读相关源码。
对于注册中心的使用，其实就是在yaml文件中做一些配置，然后有对应的管理页面可以查看和操作。当然我们肯定不仅仅局限于使用，更需要了解其背后的实现和设计。因为公司最新的应用使用的是Nacos，所以近期简单阅读了一下Nacos关于注册中心的源码实现。
基于的版本是2.1.2。
1-从DEMO出发
对于源码的阅读，可以从最简单的demo入门。
Nacos作为一个服务端，想要使用其服务发现功能，可以直接使用其提供的客户端代码。
在RPC框架中，我们的服务一般分为Provider和Consumer.
Provider会向注册中心进行服务注册
Properties properties = new Properties();
properties.setProperty("serverAddr", System.getProperty("serverAddr", "localhost"));
properties.setProperty("namespace", System.getProperty("namespace", "public"));
NamingService naming = NamingFactory.createNamingService(properties);
//注册
naming.registerInstance("nacos.test.3", "11.11.11.11", 8888, "TEST1");
System.out.println("instances after register: " + naming.getAllInstances("nacos.test.3"));
Consumer则会监听对于服务注册的实例信息
Properties properties = new Properties();
properties.setProperty("serverAddr", System.getProperty("serverAddr", "localhost"));
properties.setProperty("namespace", System.getProperty("namespace", "public"));
NamingService naming = NamingFactory.createNamingService(properties);
Executor executor = new ThreadPoolExecutor(1, 1, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>(),
runnable -> {
Thread thread = new Thread(runnable);
thread.setName("test-thread");
return thread;
});
//订阅服务列表
naming.subscribe("nacos.test.3", new AbstractEventListener() {
//EventListener onEvent is sync to handle, If process too low in onEvent, maybe block other onEvent callback.
//So you can override getExecutor() to async handle event.
@Override
public Executor getExecutor() {
return executor;
}
@Override
public void onEvent(Event event) {
System.out.println("serviceName: " + ((NamingEvent) event).getServiceName());
System.out.println("instances from event: " + ((NamingEvent) event).getInstances());
}
});
上面我们就已经完成了服务的注册与发现，虽然在项目中都是基于SpringBoot整合来实现，但是其本质都是基于这些API代码来实现。
2-服务注册
通过上面的案例，我们已经知道了如何想Nacos进行服务的注册。接下来就来看看在注册的过程中都做了哪些事情吧。
void registerInstance(String serviceName, String ip, int port, String clusterName) throws NacosException;
通过注册的接口，我们大概可以知道注册一个服务需要哪些要素
服务名称，实例的ip和端口，以及实例所属的集群名称。这些要素其实就组成了Nacos服务的分级存储模型。
在Service的上层是Group和Namespace，他们共同组成了注册中心的数据模型。
我们通过官方文档的两张图可以更加详细的了解服务发现的数据模型:
有了上面的知识，那么阅读后面的源码会轻松不少。
registerInstance接口会通过http或者grpc的方式向-->Nacos发起请求，来进行服务的注册。那么
这里就需要阅读服务的代码了。
服务端会通过下面的方法实现服务的注册。
getInstanceOperator().registerInstance(namespaceId, serviceName, instance);
这个方法有两个实现类，我这里选择的是InstanceOperatorServiceImpl来查看(Nacos V1版本的实现，另一种是V2版本的实现)。
通过ServiceManager方法中的registerInstance去注册实例 ，步骤如下:
public void registerInstance(String namespaceId, String serviceName, Instance instance) throws NacosException {
NamingUtils.checkInstanceIsLegal(instance);
createEmptyService(namespaceId, serviceName, instance.isEphemeral());
Service service = getService(namespaceId, serviceName);
checkServiceIsNull(service, namespaceId, serviceName);
addInstance(namespaceId, serviceName, instance.isEphemeral(), instance);
}
1-创建Service
对应服务分级存储模型中的服务(如果存在就直接返回)
/** * Map(namespace, Map(group::serviceName, Service)). */ private final Map<String, Map<String, Service>> serviceMap = new ConcurrentHashMap<>();
上面是Service的存储结构ConcurrentHashMap。是一个双层Map，先通过namespace找，然后在通过group和serviceName找到具体的service。这里其实可以回头看我们之前的demo代码，理解一下所传递的参数。
创建Serivce后会有一个putServiceAndInit方法做一些初始化操作，需要特别注意:
private void putServiceAndInit(Service service) throws NacosException {
putService(service);
service = getService(service.getNamespaceId(), service.getName());
service.init();
consistencyService
.listen(KeyBuilder.buildInstanceListKey(service.getNamespaceId(), service.getName(), true), service);
consistencyService
.listen(KeyBuilder.buildInstanceListKey(service.getNamespaceId(), service.getName(), false), service);
Loggers.SRV_LOG.info("[NEW-SERVICE] {}", service.toJson());
}
这里的consistencyService代表了Nacos中的一致性协议。Nacos支持CP 协议以及 AP 协议。
对于注册中心的服务发现功能，常用的就是AP协议，来保障服务的可用性。这里内容较多就不展开了。我们只需要关注AP协议对应的实现类DistroConsistencyServiceImpl，Distro 协议是阿里自研的最终⼀致性协议。
Service初始化完成后往这个协议里面加了一个Listen监听。当有对应的事件发生时，就会调用Service中onChange方法。
2-添加实例到Service中
public void addInstance(String namespaceId, String serviceName, boolean ephemeral, Instance... ips)
throws NacosException {
String key = KeyBuilder.buildInstanceListKey(namespaceId, serviceName, ephemeral);
Service service = getService(namespaceId, serviceName);
//锁
synchronized (service) {
//Compare and get new instance list.将新注册的和已经存在的都返回
List<Instance> instanceList = addIpAddresses(service, ephemeral, ips);
Instances instances = new Instances();
instances.setInstanceList(instanceList);
consistencyService.put(key, instances);
}
}
consistencyService.put(key, instances); 最终是将所有的instances通过一致性协议写入到Nacos集群中。
到这里当前的registerInstance已经结束了，后面的操作就全部在Distro协议中去完成了。
3-AP 协议下 consistencyService.put方法
在Distro协议下，Nacos的每个节点都是平等的处理写请求，并且把新数据会同步到其他的节点(关于此实现的详细介绍这里不在展开)。
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);
}
if (!listeners.containsKey(key)) {
return;
}
notifier.addTask(key, DataOperation.CHANGE);
}
可以看到数据会先写入到DataStore中，可以看到也是一个ConcurrentHashMap。
private Map<String, Datum> dataMap = new ConcurrentHashMap<>(1024);
public void put(String key, Datum value) {
dataMap.put(key, value);
}
然后会将其添加到一个阻塞队列中
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));
}
BlockingQueue<Pair<String, DataOperation>> tasks = new ArrayBlockingQueue<>(1024 * 1024);
这里放入阻塞队列中，说明肯定有异步线程去单独消费。这样做的可以提升整体服务注册的性能，并且可以避免并发加锁的情况，因为是在一个线程中进行处理：
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);
}
}
}
在handel中就对应了具体的处理
for (RecordListener listener : listeners.get(datumKey)) {
count++;
try {
if (action == DataOperation.CHANGE) {
listener.onChange(datumKey, dataStore.get(datumKey).value);
continue;
}
if (action == DataOperation.DELETE) {
listener.onDelete(datumKey);
continue;
}
} catch (Throwable e) {
Loggers.DISTRO.error("[NACOS-DISTRO] error while notifying listener of key: {}", datumKey, e);
}
}
这里listener.onChange(datumKey, dataStore.get(datumKey).value);就会触发前面Service中添加的监听，代码也就走回了Service中的onChange方法。
4-Serivce.onChange方法
public void onChange(String key, Instances value) throws Exception {
Loggers.SRV_LOG.info("[NACOS-RAFT] datum is changed, key: {}, value: {}", key, value);
for (Instance instance : value.getInstanceList()) {
if (instance == null) {
// Reject this abnormal instance list:
throw new RuntimeException("got null instance " + key);
}
if (instance.getWeight() > 10000.0D) {
instance.setWeight(10000.0D);
}
if (instance.getWeight() < 0.01D && instance.getWeight() > 0.0D) {
instance.setWeight(0.01D);
}
}
//更新实例信息，并且通知订阅者服务变化的信息
updateIPs(value.getInstanceList(), KeyBuilder.matchEphemeralInstanceListKey(key));
recalculateChecksum();
}
这个方法主要做的就是将Service中的Cluster和Instance数据进行更新(就是前面图中数据模型对应的集群和实例)，并且通知订阅者。
整个更新逻辑都在updateIPs方法中，更新的方式使用了copy-on-write的思想。
public void updateIPs(Collection<Instance> instances, boolean ephemeral) {
//新的ipMap
Map<String, List<Instance>> ipMap = new HashMap<>(clusterMap.size());
for (String clusterName : clusterMap.keySet()) {
ipMap.put(clusterName, new ArrayList<>());
}
for (Instance instance : instances) {
try {
if (instance == null) {
Loggers.SRV_LOG.error("[NACOS-DOM] received malformed ip: null");
continue;
}
if (StringUtils.isEmpty(instance.getClusterName())) {
instance.setClusterName(UtilsAndCommons.DEFAULT_CLUSTER_NAME);
}
if (!clusterMap.containsKey(instance.getClusterName())) {
Loggers.SRV_LOG.warn(
"cluster: {} not found, ip: {}, will create new cluster with default configuration.",
instance.getClusterName(), instance.toJson());
Cluster cluster = new Cluster(instance.getClusterName(), this);
cluster.init();
getClusterMap().put(instance.getClusterName(), cluster);
}
List<Instance> clusterIPs = ipMap.get(instance.getClusterName());
if (clusterIPs == null) {
clusterIPs = new LinkedList<>();
ipMap.put(instance.getClusterName(), clusterIPs);
}
clusterIPs.add(instance);
} catch (Exception e) {
Loggers.SRV_LOG.error("[NACOS-DOM] failed to process ip: " + instance, e);
}
}
for (Map.Entry<String, List<Instance>> entry : ipMap.entrySet()) {
//make every ip mine
List<Instance> entryIPs = entry.getValue();
//更新到真正的clusterMap中
clusterMap.get(entry.getKey()).updateIps(entryIPs, ephemeral);
}
setLastModifiedMillis(System.currentTimeMillis());
getPushService().serviceChanged(this);
ApplicationUtils.getBean(DoubleWriteEventListener.class).doubleWriteToV2(this, ephemeral);
StringBuilder stringBuilder = new StringBuilder();
for (Instance instance : allIPs()) {
stringBuilder.append(instance.toIpAddr()).append('_').append(instance.isHealthy()).append(',');
}
Loggers.EVT_LOG.info("[IP-UPDATED] namespace: {}, service: {}, ips: {}", getNamespaceId(), getName(),
stringBuilder.toString());
}
最终是直接将Instance数据进行了替换，使用这种方式可以解决读写并发互斥的情况。
private Set<Instance> ephemeralInstances = new HashSet<>(); ephemeralInstances = toUpdateInstances;
最后通过PushService来通知订阅者，底层是基于UDP的推送。
3-走回DEMO
到此我们在回到demo中，大概就知道naming.subscribe方法中的onEvent方法是如何被调用的了。肯定是接收到了pushService的通知，然后进行回调了所有的subscribe。这块代码大家可以自行去查看。
4-总结和说明
上面的代码其实基本都是在1.x的版本中已经存在和使用的，在2.x的版本中Nacos服务端注册提供了v2版本的实现如果是InstanceControllerV2。
也提供了v2版本的实现，大家也可以去学习：
instanceServiceV2.registerInstance(namespaceId, serviceName, instance);
并且这只是Nacos的服务发现的源码的一小部门，只是入门和熟悉了一下Nacos的源码。还有很多功能没有涉及到，如健康检查、Distro协议的设计，是如何保证最终一致性的。这些在后面有时间的时候也会做一些记录和分享。
加载全部内容