使用nacos注册都经历了什么?
客户端进行注册服务
通常我们使用这样的api就可以进行服务注册了
namingService.registerInstance(serviceName,"ddddddddd","localhost",
8092);
默认注册的都是临时节点:
instance.isEphemeral()
客户端做了什么
1-客户端向nacos server 服务注册的同时会启动一个心跳的定时任务。
beatReactor.addBeatInfo(NamingUtils.getGroupedName(serviceName, groupName), beatInfo);
executorService.schedule(new BeatTask(beatInfo), beatInfo.getPeriod(), TimeUnit.MILLISECONDS);
定期检查服务端是否还存在注册的此服务信息。
如果不存在则会重新发起注册请求,BeatTask:
if (code == NamingResponseCode.RESOURCE_NOT_FOUND) {
Instance instance = new Instance();
instance.setPort(beatInfo.getPort());
instance.setIp(beatInfo.getIp());
instance.setWeight(beatInfo.getWeight());
instance.setmetadata(beatInfo.getmetadata());
instance.setClusterName(beatInfo.getCluster());
instance.setServiceName(beatInfo.getServiceName());
instance.setInstanceId(instance.getInstanceId());
instance.setEphemeral(true);
try {
//这里会发起注册的请求
serverProxy.registerService(beatInfo.getServiceName(),
NamingUtils.getGroupName(beatInfo.getServiceName()), instance);
} catch (Exception ignore) {
}
}
2-想nacos服务端发起注册的http请求进行服务注册
serverProxy.registerService(NamingUtils.getGroupedName(serviceName, groupName), groupName, instance);
通过http请求向服务端发起post请求进行服务注册：
reqAPI(UtilAndComs.NACOS_URL_INSTANCE, params, HttpMethod.POST);
服务端做了什么
服务端通过InstanceController中的register方法对客户端的http请求进行处理:
serviceManager.registerInstance(namespaceId, serviceName, instance);
1-serviceManager和consistencyService
serviceManager是一个管理服务注册的类,并且被spring管理,
服务端启动后会进行执行init() 方法:
consistencyService.listen(KeyBuilder.SERVICE_meta_KEY_PREFIX, this);
将自己加入到到consistencyService的监听集合中。
consistencyService是一个一致性的接口,这里基于我们关注基于AP原则实现的注册
中心,所以其实现类是DistroConsistencyServiceImpl,并且也被spring管理,初始
化的时候会执行init()方法:
@PostConstruct
public void init() {
//启动两个任务,一个进行从其他服务加载数据,一个进行监听配置的变更
GlobalExecutor.submit(loadDataTask);
GlobalExecutor.submitDistroNotifyTask(notifier);
}
loadDataTask:当服务端重启或者新增一个nacos服务实例的时候会从其他服务器
同步数据到当前服务器,然后将服务信息存入内存中:
//1-将配置存入内存中
dataStore.put(entry.getKey(), entry.getValue());
并且会通知加入到DistroConsistencyServiceImpl监听集合
的监听器(为什么从其他服务器同步,因为nacos在AP原则下将数据保存到了内存中)：
因此loadDataTask最终会触发ServiceManager的onChange方法。
2-Service
一个Service代表一个服务,每个服务中包含一个Cluster,每个Cluster包含了这个
服务所有注册的服务实例信息。
继续看registerInstance方法:
createEmptyService(namespaceId, serviceName, instance.isEphemeral());
每个服务都会创建一个Service对象存入map中
通过putServiceAndInit方法
将其服务对象Service自身加入到DistroConsistencyServiceImpl的监听集合中,
consistencyService.listen(KeyBuilder.buildInstanceListKey(service.getNamespaceId(), service.getName(), true), service);
consistencyService.listen(KeyBuilder.buildInstanceListKey(service.getNamespaceId(), service.getName(), false), service);
这样当服务对应的信息发生改变,同样会触发Service中的回调,来更改实例信息。
3-addInstance 注册实例+持久化+通知客户端的订阅者
这里的instance实例是客户端请求中传递过来进行注册服务的信息,
代表一个服务一个具体的实例信息:ip 端口 权重等。
addInstance(namespaceId, serviceName, instance.isEphemeral(), instance);
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) {
//这里返回的是当前服务在集群中所有的服务列表信息
List instanceList = addIpAddresses(service, ephemeral, ips);
Instances instances = new Instances();
instances.setInstanceList(instanceList);
//这里会将最新的服务列表保存到内存,并且通知客户端的订阅者服务列表更新了
consistencyService.put(key, instances);
}
}
这里首先获取服务对应的Service对象。
然后调用addIpAddresses对其添加实例信息
在addIpAddresses方法中可以看到如下代码:
//首先从服务对象中的Cluster中获取现有内存中保存的实例列表
List currentIPs = service.allIPs(ephemeral);
......
if (!service.getClusterMap().containsKey(instance.getClusterName())) {
//这里可以看到每个服务的Service对象都根据ClusterName管理着一个Cluster,Cluster保存了当前服务所有的实例列表。也是就instance.
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());
}
Cluster中通过set来保存实例信息列表:
private Set ephemeralInstances = new HashSet<>();
实例列表信息是什么时候追加到Cluster中的呢?(后面会说明)
addIpAddresses方法最终会将当前注册的实例信息和已经注册的实例信息
都加入到Set中返回出去。
获取到返回的服务列表信息后,肯定就需要进行持久化和通知客户端的订阅者了。
addInstance方法中的consistencyService.put(key, instances);
实现了这一切。
来到DistroConsistencyServiceImpl的put方法:
这里的value就是实例的列表:
public void put(String key, Record value) throws NacosException {
onPut(key, value);
taskDispatcher.addTask(key);
}
public void onPut(String key, Record value) {
if (KeyBuilder.matchEphemeralInstanceListKey(key)) {
Datum 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, ApplyAction.CHANGE);
}
可以看到在基于AP原则的情况下服务端存储的注册信息是依赖于
dataStore,其实就是一个map
private Map dataMap = new ConcurrentHashMap<>(1024);
存入内存后,如果是更新的服务,则之前的notifier任务就体现其作用了:
因为之前的putServiceAndInit方法已经将每个服务Service本身作为
监听器加入到了DistroConsistencyServiceImpl的监听集合中,
因此Notifier在处理这个服务的时候会去回调最终走到ApplyAction.CHANGE的分支
for (RecordListener listener : listeners.get(datumKey)) {
count++;
try {
if (action == ApplyAction.CHANGE) {
listener.onChange(datumKey, dataStore.get(datumKey).value);
continue;
}
if (action == ApplyAction.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中的onChange方法,然后将实例信息更新到Service中的Cluster中。(这里更新了Cluster的实例列表),并且发送udp请求给订阅服务列表信息的客户端
//这里通知发送请求到客户端
updateIPs(value.getInstanceList(), KeyBuilder.matchEphemeralInstanceListKey(key));
getPushService().serviceChanged(this);
总结
这里对使用nacos注册中心的注册情况进行了总结,核心的几个类为:
DistroConsistencyServiceImpl：实现基于AP的注册中心,将服务注册信息保存到内存map中。并在服务启动的时候去其他服务器同步
数据。当服务信息变化的时候对服务本身Service进行通知。
Service:服务对应的对象,保存了服务所有的实例列表在其Cluster中。
nacos中不管是注册中心还是配置中心都大量使用了观察者的设计模式,所以有必要对常见的设计模式进行学习和使用。
后续会对其进行学习和分享。
