阅读说明：本文主要剖析Nacos服务注册的关键源码，为了讲清楚源码中的关键步骤与实现细节，会从三个维度进行解析：
- 时序图：快速了解服务注册的交互链路。
- 方法说明：快速了解主要方法的作用。
- 源码解析：解析服务注册实现细节（关键代码均有注释说明）。
由于本文篇幅较长，各位看官可以通过时序图 方法说明的方式快速了解Nacos服务注册的实现步骤以及设计思想，源码当作补充说明。
版本说明：服务注册源码分析基于Nacos-1.4.3版本。
服务注册客户端入口Nacos客户端依赖：
<dependency>
<groupId>com.alibaba.cloud</groupId>
<artifactId>spring-cloud-starter-alibaba-nacos-discovery</artifactId>
</dependency>
基于Springboot自动装备原理，服务启动时会自动加载Nacos客户端相关的配置类。
如图所示：
其中，与服务注册相关的配置类为：NacosServiceRegistryAutoConfiguration。
如图所示：
NacosServiceRegistryAutoConfiguration是Nacos服务注册的自动配置类，该类的主要功能是按条件注入NacosAutoServiceRegistration、NacosServiceRegistry、NacosRegistration等服务注册相关的Bean。
客户端发送服务注册请求服务注册客户端交互逻辑，如图所示：
NacosAutoServiceRegistration继承了AbstractAutoServiceRegistration，该类实现了ApplicationListener接口的onApplicationEvent()方法。
服务启动，Spring初始化完成后，发布对应的初始化完成事件（WebServerInitializedEvent），触发AbstractAutoServiceRegistration.onApplicationEvent()方法调用逻辑，开启注册流程。
源码如下：
@Override
@SuppressWarnings("deprecation")
public void onApplicationEvent(WebServerInitializedEvent event) {
bind(event);
}
@Deprecated
public void bind(WebServerInitializedEvent event) {
...
// 通过CAS方式设置服务端口
this.port.compareAndSet(0, event.getWebServer().getPort());
// 开启服务注册流程
this.start();
}
start()方法主要做以下事情：
- 发布服务注册相关事件。
- 执行服务注册。
- 发布服务注册完成事件。
- 更新服务运行状态。
源码如下：
public void start() {
...
// 服务启动时，服务运行状态为false
if (!this.running.get()) {
// 发布服务预注册事件
this.context.publishEvent(
new InstancePreRegisteredEvent(this, getRegistration()));
// 执行服务注册
register();
if (shouldRegisterManagement()) {
registerManagement();
}
// 发布服务注册完成事件
this.context.publishEvent(
new InstanceRegisteredEvent<>(this, getConfiguration()));
// 更新服务运行状态为true
this.running.compareAndSet(false, true);
}
}
register()方法主要做以下事情：
- 获取服务注册基本信息（如：服务名、服务IP、端口等）。
- 调用NacosServiceRegistry#register方法执行服务注册。
源码如下：
protected void register() {
this.serviceRegistry.register(getRegistration());
}
其中，getRegistration()主要作用是从NacosRegistration中获取Nacos服务注册基本信息。
NacosRegistrationNacosRegistration主要用于管理Nacos服务注册基本信息，该类主要封装了NacosDiscoveryProperties的配置信息（application.yml），同时，该类实现了Registration接口（继承自ServiceInstance），Registration是Springcloud提供的服务注册信息通用接口，用于管理服务注册的基本信息（如：服务ID、服务IP、端口等）。
NacosServiceRegistryNacosServiceRegistry类实现了ServiceRegistry接口，该接口是Springcloud提供的服务注册通用接口，声明了服务注册、取消注册等方法。
NacosServiceRegistry#register方法主要做以下事情：
- 创建服务实例封装服务注册信息。
- 调用NamingService#registerInstance方法发起Nacos服务注册。
源码如下：
// com.alibaba.cloud.nacos.registry.NacosServiceRegistry.register
@Override
public void register(Registration registration) {
// ...
// 获取Nacos命名服务
NamingService namingService = namingService();
// 获取服务名，默认为应用名
String serviceId = registration.getServiceId();
// 获取服务分组名，默认为DEFAULT_GROUP
String group = nacosDiscoveryProperties.getGroup();
// 创建服务实例，封装服务注册信息（如：服务IP、端口、权重、服务节点类型等）
Instance instance = getNacosInstanceFromRegistration(registration);
try {
// 调用NamingService.registerInstance()方法进行服务注册
namingService.registerInstance(serviceId, group, instance);
// ...
}catch (Exception e) {
// ...
}
}
NamingService
NamingService接口定义了服务注册、服务发现等服务注册相关的核心方法，它的默认实现为NacosNamingService。
NacosNamingService#registerInstance方法主要做以下事情：
- 构建心跳信息，并向Nacos服务端定时（5秒/次）发送心跳请求（临时实例走的是Nacos的AP模式，Nacos服务端需要检测服务的可用性）。
- 调用NamingProxy#registerService方法向Nacos服务端注册服务实例。
源码如下：
// com.alibaba.nacos.client.naming.NacosNamingService#registerInstance
public void registerInstance(String serviceName, String groupName, Instance instance) throws NacosException {
NamingUtils.checkInstanceIsLegal(instance);
String groupedServiceName = NamingUtils.getGroupedName(serviceName, groupName);
// 判断是否为临时实例
if (instance.isEphemeral()) {
// 构建服务心跳信息
BeatInfo beatInfo = beatReactor.buildBeatInfo(groupedServiceName, instance);
// 构建心跳信息，并向Nacos服务端定时（5秒/次）发送心跳请求
beatReactor.addBeatInfo(groupedServiceName, beatInfo);
}
// 向Nacos服务端注册服务实例
serverProxy.registerService(groupedServiceName, groupName, instance);
}
NamingProxy#registerService方法主要做以下事情：
- 组装请求参数。
- 调用服务注册接口（/v1/ns/instance）向Nacos服务端注册服务实例。
源码如下：
// com.alibaba.nacos.client.naming.net.NamingProxy.registerService
public void registerService(String serviceName, String groupName, Instance instance) throws NacosException {
// ...
// 组装请求参数
final Map<String, String> params = new HashMap<String, String>(16);
params.put(CommonParams.NAMESPACE_ID, namespaceId);
params.put(CommonParams.SERVICE_NAME, serviceName);
params.put(CommonParams.GROUP_NAME, groupName);
params.put(CommonParams.CLUSTER_NAME, instance.getClusterName());
params.put("ip", instance.getIp());
params.put("port", String.valueOf(instance.getPort()));
params.put("weight", String.valueOf(instance.getWeight()));
params.put("enable", String.valueOf(instance.isEnabled()));
params.put("healthy", String.valueOf(instance.isHealthy()));
params.put("ephemeral", String.valueOf(instance.isEphemeral()));
params.put("metadata", JacksonUtils.toJson(instance.getMetadata()));
// 调用Nacos服务端服务注册接口（/v1/ns/instance）注册服务实例。
reqApi(UtilAndComs.nacosUrlInstance, params, HttpMethod.POST);
}
public String reqApi(String api, Map<String, String> params, String method) throws NacosException {
return reqApi(api, params, Collections.EMPTY_MAP, method);
}
public String reqApi(String api, Map<String, String> params, Map<String, String> body, String method)
throws NacosException {
return reqApi(api, params, body, getServerList(), method);
}
public String reqApi(String api, Map<String, String> params, Map<String, String> body, List<String> servers,
String method) throws NacosException {
// ...
if (StringUtils.isNotBlank(nacosDomain)) {
for (int i = 0; i < maxRetry; i ) {
try {
// 调用服务注册接口（/v1/ns/instance）向Nacos服务端注册服务实例
return callServer(api, params, body, nacosDomain, method);
} catch (NacosException e) {
// ...
}
}
} else {
Random random = new Random(System.currentTimeMillis());
int index = random.nextInt(servers.size());
for (int i = 0; i < servers.size(); i ) {
String server = servers.get(index);
try {
// 调用服务注册接口（/v1/ns/instance）向Nacos服务端注册服务实例
return callServer(api, params, body, server, method);
} catch (NacosException e) {
// ...
}
index = (index 1) % servers.size();
}
}
// ...
}服务端服务注册入口
Nacos服务端启动类为nacos-console模块中的com.alibaba.nacos.Nacos，在nacos-console模块中引入了nacos-naming模块，该模块中的com.alibaba.nacos.naming.controllers包中提供了服务注册、服务发现、健康检查等核心接口。如图所示：
获取服务最新服务实例列表交互逻辑，如图所示：
InstanceController
Nacos服务端接收客户端发送的服务注册请求后，调用InstanceController#register方法处理服务注册请求。
InstanceController#register方法主要做以下事情：
- 根据请求参数获取namespaceId，默认为：public。
- 根据请求参数获取serviceName，格式：groupName@@serviceName。
- 根据请求参数构建服务实例。
- 调用ServiceManage#registerInstance方法注册服务实例。
源码如下：
// com.alibaba.nacos.naming.controllers.InstanceController.register
@CanDistro
@PostMapping
@Secured(parser = NamingResourceParser.class, action = ActionTypes.WRITE)
public String register(HttpServletRequest request) throws Exception {
// 根据客户端请求参数获取namespaceId，默认为：public
final String namespaceId = WebUtils
.optional(request, CommonParams.NAMESPACE_ID, Constants.DEFAULT_NAMESPACE_ID);
// 根据客户端请求参数获取serviceName，格式：groupName@@serviceName
final String serviceName = WebUtils.required(request, CommonParams.SERVICE_NAME);
NamingUtils.checkServiceNameFormat(serviceName);
// 根据客户端请求参数构建服务实例
final Instance instance = parseInstance(request);
// 调用ServiceManage#registerInstance方法注册服务实例
serviceManager.registerInstance(namespaceId, serviceName, instance);
return "ok";
}
ServiceManage
ServiceManage是Nacos服务的核心管理类，其中定义了服务注册表以及服务相关的各种核心API。如图所示：
ServiceManager#registerInstance方法主要做以下事情：
- 创建并初始化服务，将其添加到注册表中（第一次注册服务）。
- 根据namespaceId,、serviceName获取对应的服务。
- 向服务中添加要注册的服务实例。
源码如下：
public void registerInstance(String namespaceId, String serviceName, Instance instance) throws NacosException {
// 创建并初始化服务，将其添加到注册表中（第一次注册服务）
createEmptyService(namespaceId, serviceName, instance.isEphemeral());
// 根据namespaceId,、serviceName获取对应的服务
Service service = getService(namespaceId, serviceName);
// ...
// 向服务中添加要注册的服务实例
addInstance(namespaceId, serviceName, instance.isEphemeral(), instance);
}
ServiceManager#createEmptyService方法主要做以下事情：
- 判断服务在注册表中是否存在：
- 存在，则不做处理。
- 不存在，则创建服务：
- 创建服务并将服务添加到注册表中。
- 开启服务心跳检测。
- 监听服务实例变更。
- 如果不是创建本地服务，则根据服务实例类型（临时、持久）选择不同协议进行服务注册与同步。
源码如下：
public void createEmptyService(String namespaceId, String serviceName, boolean local) throws NacosException {
createServiceIfAbsent(namespaceId, serviceName, local, null);
}
// com.alibaba.nacos.naming.core.ServiceManager#createServiceIfAbsent
public void createServiceIfAbsent(String namespaceId, String serviceName, boolean local, Cluster cluster)
throws NacosException {
// 根据namespaceId、serviceName从注册表中获取对应的服务
Service service = getService(namespaceId, serviceName);
// 如果服务不存在，则创建并初始化服务
if (service == null) {
Loggers.SRV_LOG.info("creating empty service {}:{}", namespaceId, serviceName);
// 创建服务
service = new Service();
// 设置服务名
service.setName(serviceName);
// 设置namespaceId
service.setNamespaceId(namespaceId);
// 设置分组名
service.setGroupName(NamingUtils.getGroupName(serviceName));
// now validate the service. if failed, exception will be thrown
service.setLastModifiedMillis(System.currentTimeMillis());
// 根据服务实例列表生成MD5校验值（如：3c84b68316cb8449ca2842a990e2dd7d）
service.recalculateChecksum();
if (cluster != null) {
cluster.setService(service);
service.getClusterMap().put(cluster.getName(), cluster);
}
// 校验服务名和服务集群名格式
service.validate();
// 该方法主要做一下事情：
// 1.将服务添加到注册表中
// 2.初始化服务并启动心跳检测（5秒/次）
// 3.监听服务实例变更
putServiceAndInit(service);
// 不是创建本地服务
if (!local) {
// 根据服务实例类型（临时、持久）选择不同协议进行服务注册与同步：
// 临时实例：基于Distro协议将服务信息同步给集群中的其他Nacos节点（即：AP模式）
// 持久实例：基于Raft协议将服务信息同步给集群中的其他Nacos节点（即：CP模式）
addOrReplaceService(service);
}
}
}
ServiceManager#addInstance方法主要做以下事情：
- 根据namespaceId、serviceName、实例类型（临时、持久）生成服务唯一标识。
- 根据namespaceId,、serviceName获取对应的服务。
- 对服务对象添加同步锁：
- 获取服务最新的服务实例列表（旧实例列表 待注册实例列表）。
- 根据服务实例类型（临时、持久）选择不同协议进行服务注册与同步：
- 临时实例：基于Distro协议将服务信息同步给集群中的其他Nacos节点（即：AP模式）。
- 持久实例：基于Raft协议将服务信息同步给集群中的其他Nacos节点（即：CP模式）。
源码如下：
// com.alibaba.nacos.naming.core.ServiceManager#addInstance
public void addInstance(String namespaceId, String serviceName, boolean ephemeral, Instance... ips)
throws NacosException {
// 根据namespaceId、serviceName、实例类型（临时、持久）构建服务唯一标识：
// 临时实例：com.alibaba.nacos.naming.iplist.ephemeral.namespaceId##groupName@@serviceName
// 持久实例：com.alibaba.nacos.naming.iplist.namespaceId##groupName@@serviceName
String key = KeyBuilder.buildInstanceListKey(namespaceId, serviceName, ephemeral);
// 根据namespaceId,、serviceName获取对应的服务
Service service = getService(namespaceId, serviceName);
// 对服务对象添加同步锁
synchronized (service) {
// 通过对比待注册实例列表与旧实例列表，获取最新的实例列表（旧实例列表 待注册实例列表）。
// 注意：此处会将当前服务实例列表拷贝到一个新的服务实例列表中，基于CopyOnWrite思想
List<Instance> instanceList = addIpAddresses(service, ephemeral, ips);
// 将新的服务实例列表封装为Instances
Instances instances = new Instances();
instances.setInstanceList(instanceList);
// 根据服务实例类型（临时、持久）选择不同协议进行服务注册与同步：
// 临时实例：基于Distro协议将服务信息同步给集群中的其他Nacos节点（即：AP模式）
// 持久实例：基于Raft协议将服务信息同步给集群中的其他Nacos节点（即：CP模式）
consistencyService.put(key, instances);
}
}
ServiceManager#addIpAddresses->updateIpAddresses方法主要做以下事情：
- 根据服务信息尝试从数据集（DataStore）中获取旧的服务实例列表。
- 对比待处理的服务实例列表和旧的服务实例列表：
- 如果是删除操作，则将服务实例从最新的服务实例列表中移除。
- 如果是新增或更新操作，则将服务实例加入或更新到最新的服务实例列表中。
源码如下：
private List<Instance> addIpAddresses(Service service, boolean ephemeral, Instance... ips) throws NacosException {
return updateIpAddresses(service, UtilsAndCommons.UPDATE_INSTANCE_ACTION_ADD, ephemeral, ips);
}
public List<Instance> updateIpAddresses(Service service, String action, boolean ephemeral, Instance... ips)
throws NacosException {
// 根据服务唯一标识获取当前数据集（dataStore）中旧的服务实例列表，返回值是Datum（key：服务唯一标识，value：服务实例列表）
Datum datum = consistencyService
.get(KeyBuilder.buildInstanceListKey(service.getNamespaceId(), service.getName(), ephemeral));
// 根据实例类型（临时、持久）获取现有服务实例列表
List<Instance> currentIPs = service.allIPs(ephemeral);
// 创建Map来保存现有服务实例列表，key：服务实例IP Port，value：服务实例
Map<String, Instance> currentInstances = new HashMap<>(currentIPs.size());
// 创建Set来保存现有服务实例的instanceId（实例唯一标识）
Set<String> currentInstanceIds = Sets.newHashSet();
// 遍历现有服务实例列表
for (Instance instance : currentIPs) {
// 将现有服务实例保存到新创建的Map中
currentInstances.put(instance.toIpAddr(), instance);
// 将现有服务实例的instanceId保存到新创建的Set中
currentInstanceIds.add(instance.getInstanceId());
}
// 定义新的Map来保存更新后的服务实例列表
Map<String, Instance> instanceMap;
if (datum != null && null != datum.value) {
// 如果服务存在旧的实例列表，则更新旧实例的健康状态和心跳时间并将其添加到新的Map中
instanceMap = setValid(((Instances) datum.value).getInstanceList(), currentInstances);
} else {
// 如果服务不存在旧的实例列表，则创建新Map
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);
// ...
}
if (UtilsAndCommons.UPDATE_INSTANCE_ACTION_REMOVE.equals(action)) {
// 如果是删除操作，则将待处理服务实例从Map中移除
instanceMap.remove(instance.getDatumKey());
} else {
// 根据待处理服务信息组成服务唯一标识从旧的服务实例列表中查找
Instance oldInstance = instanceMap.get(instance.getDatumKey());
// 如果是旧的服务实例，则更新服务实例信息
if (oldInstance != null) {
instance.setInstanceId(oldInstance.getInstanceId());
} else {
// 如果是新的服务实例，则新增服务实例信息（即：生成新的instanceId）
instance.setInstanceId(instance.generateInstanceId(currentInstanceIds));
}
// 将新的服务实例加入Map中
instanceMap.put(instance.getDatumKey(), instance);
}
}
// ...
// 将最新服务实例列表对应的Map转为对应的List，并返回
return new CopyOnWriteArrayList<>(instanceMap.values());
}
ConsistencyService#put方法为服务注册与同步核心逻辑，该方法主要做以下事情：
根据服务唯一标识前缀判断服务实例类型（临时、持久）：
- com.alibaba.nacos.naming.iplist.ephemeral：临时实例，则基于Distro协议将服务信息同步给集群中的其他Nacos节点（即：AP模式）。
- com.alibaba.nacos.naming.iplist：持久实例，则基于Raft协议将服务信息同步给集群中的其他Nacos节点（即：CP模式）。
- 以临时实例为例，处理步骤：
- 将服务对应的实例列表信息封装成Datum加入到DataStore（作用是根据服务唯一标识获取对应的服务实例列表）中。
- 将服务唯一标识、操作类型封装成一个Task任务加入到DistroConsistencyServiceImpl.Notifier的任务队列中。
- 遍历集群中除自己以外的其他Nacos节点：
- 先将服务信息（资源）、资源类型（如：服务信息同步、元数据同步）以及目标节点封装成DistroKey。
- 再将DistroKey、操作类型（如：新增、删除实例）封装成延时任务DelayTask（延时时长：1秒）添加到Nacos延时任务执行器的任务集合中。
源码如下：
// com.alibaba.nacos.naming.consistency.DelegateConsistencyServiceImpl#put
public void put(String key, Record value) throws NacosException {
// 根据服务唯一标识前缀判断服务实例类型（临时、持久）
mapConsistencyService(key).put(key, value);
}
// com.alibaba.nacos.naming.consistency.ephemeral.distro.DistroConsistencyServiceImpl#put
@Override
public void put(String key, Record value) throws NacosException {
// 1.将服务对应的实例列表信息封装成Datum加入到DataStore（作用是根据服务唯一标识获取对应的服务实例列表）中。
// 2.将服务唯一标识、操作类型封装成一个Task任务加入到DistroConsistencyServiceImpl.Notifier的任务队列中。
onPut(key, value);
// 遍历集群中除自己以外的其他Nacos节点：
// 先将服务信息（资源）、资源类型（如：服务信息同步、元数据同步）以及目标节点封装成DistroKey。
// 再将DistroKey、操作类型（如：新增、删除实例）封装成延时任务DelayTask（延时时长：1秒）添加到Nacos延时任务执行器的任务集合中。
distroProtocol.sync(new DistroKey(key, KeyBuilder.INSTANCE_LIST_KEY_PREFIX), DataOperation.CHANGE,
globalConfig.getTaskDispatchPeriod() / 2);
}
public void onPut(String key, Record value) {
// 判断是否为临时实例
if (KeyBuilder.matchEphemeralInstanceListKey(key)) {
// 将服务信息封装成Datum（key：服务唯一标识，value：服务实例列表）
Datum<Instances> datum = new Datum<>();
datum.value = (Instances) value;
datum.key = key;
datum.timestamp.incrementAndGet();
// 将Datum添加到DataStore中
dataStore.put(key, datum);
}
if (!listeners.containsKey(key)) {
return;
}
// 向Notifier的任务队列中添加任务
notifier.addTask(key, DataOperation.CHANGE);
}
public void addTask(String datumKey, DataOperation action) {
// 如果服务集合中存在服务并且服务操作类型为更新，则直接返回
if (services.containsKey(datumKey) && action == DataOperation.CHANGE) {
return;
}
// 如果服务操作类型为更新，则将服务添加到服务集合中
if (action == DataOperation.CHANGE) {
services.put(datumKey, StringUtils.EMPTY);
}
// 向Notifier的任务队列中添加任务
tasks.offer(Pair.with(datumKey, action));
}
public void sync(DistroKey distroKey, DataOperation action, long delay) {
// 遍历集群中除自己以外的其他Nacos节点
for (Member each : memberManager.allMembersWithoutSelf()) {
// 构建DistroKey，其中：
// resourceKey为服务唯一标识
// resourceType为com.alibaba.nacos.naming.iplist.（表示同步服务信息）
// targetServer为Nacos节点
DistroKey distroKeyWithTarget = new DistroKey(distroKey.getResourceKey(), distroKey.getResourceType(),
each.getAddress());
// 根据创建的DistroKey和操作类型构建Distro延时任务（延时时长：1秒）
DistroDelayTask distroDelayTask = new DistroDelayTask(distroKeyWithTarget, action, delay);
// 向延时任务执行器（NacosDelayTaskExecuteEngine）中添加延时任务
distroTaskEngineHolder.getDelayTaskExecuteEngine().addTask(distroKeyWithTarget, distroDelayTask);
if (Loggers.DISTRO.isDebugEnabled()) {
Loggers.DISTRO.debug("[DISTRO-SCHEDULE] {} to {}", distroKey, each.getAddress());
}
}
}服务注册
服务注册交互逻辑，如图所示：
DistroConsistencyServiceImpl初始化时会开启一个线程池DISTRO_NOTIFY_EXECUTOR，其中只有一个线程（即：DistroConsistencyServiceImpl.Notifier），主要作用是通过死循环的方式从任务队列（Notifier.tasks）中取出任务进行处理（即：将服务实例列表更新到注册表中）。
源码如下：
// com.alibaba.nacos.naming.consistency.ephemeral.distro.DistroConsistencyServiceImpl#init
@PostConstruct
public void init() {
GlobalExecutor.submitDistroNotifyTask(notifier);
}
// com.alibaba.nacos.naming.consistency.ephemeral.distro.DistroConsistencyServiceImpl.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);
// ...
@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.Notifier#handle方法主要作用是遍历任务中的服务信息：
- 如果是更新操作，则更新服务对应的实例列表。
- 如果是删除操作，则从服务对应的实例列表中删除实例。
源码如下：
private void handle(Pair<String, DataOperation> pair) {
try {
// ...
// 遍历任务中服务信息（Service继承来RecordListener）
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) {
// ...
}
}
// ...
} catch (Throwable e) {
// ...
}
}
更新操作
更新操作由Service#onChange方法进行处理，该方法主要做以下事情：
- 更新服务对应的实例列表信息。
- 更新服务MD5校验码。
源码如下：
public void onChange(String key, Instances value) throws Exception {
// ...
// 更新服务对应的实例列表信息
updateIPs(value.getInstanceList(), KeyBuilder.matchEphemeralInstanceListKey(key));
// 更新服务MD5校验码
recalculateChecksum();
}
public void updateIPs(Collection<Instance> instances, boolean ephemeral) {
// 创建Map：用于保存服务对应的Cluster信息
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) {
continue;
}
if (StringUtils.isEmpty(instance.getClusterName())) {
instance.setClusterName(UtilsAndCommons.DEFAULT_CLUSTER_NAME);
}
// 如果服务实例对应的集群信息不存在，则创建并初始化集群信息
if (!clusterMap.containsKey(instance.getClusterName())) {
// ... 日志信息
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) {
// ... 日志信息
}
}
for (Map.Entry<String, List<Instance>> entry : ipMap.entrySet()) {
//make every ip mine
List<Instance> entryIPs = entry.getValue();
// 更新集群实例列表
clusterMap.get(entry.getKey()).updateIps(entryIPs, ephemeral);
}
// 设置服务最后更新时间
setLastModifiedMillis(System.currentTimeMillis());
// 发布服务信息变更事件（即：将服务变更通知订阅该服务的客户端）
getPushService().serviceChanged(this);
// ...
}
// com.alibaba.nacos.naming.core.Cluster#updateIps
public void updateIps(List<Instance> ips, boolean ephemeral) {
// 根据实例类型（临时、持久）获取旧的服务实例列表
Set<Instance> toUpdateInstances = ephemeral ? ephemeralInstances : persistentInstances;
// 将旧的服务实例列表转储到新创建Map中（key：服务唯一标识，value：服务实例）
HashMap<String, Instance> oldIpMap = new HashMap<>(toUpdateInstances.size());
for (Instance ip : toUpdateInstances) {
oldIpMap.put(ip.getDatumKey(), ip);
}
// 对比待处理服务实例列表和旧服务实例列表，获取需要更新的服务实例列表
List<Instance> updatedIPs = updatedIps(ips, oldIpMap.values());
if (updatedIPs.size() > 0) {
for (Instance ip : updatedIPs) {
Instance oldIP = oldIpMap.get(ip.getDatumKey());
if (!ip.isMarked()) {
ip.setHealthy(oldIP.isHealthy());
}
// ...
}
}
// 获取待处理服务实例列表中不存在于旧服务实例列表中的实例（即：新增实例），更新该实例的健康状态为健康
List<Instance> newIPs = subtract(ips, oldIpMap.values());
if (newIPs.size() > 0) {
// ... 日志信息
for (Instance ip : newIPs) {
// 更新服务实例健康状态为健康
HealthCheckStatus.reset(ip);
}
}
// 获取旧服务实例列表中不存在于待处理实例列表中的实例（即：移除实例），更新该实例的健康状态为死亡
List<Instance> deadIPs = subtract(oldIpMap.values(), ips);
if (deadIPs.size() > 0) {
// ... 日志信息
for (Instance ip : deadIPs) {
// 更新服务实例健康状态为死亡
HealthCheckStatus.remv(ip);
}
}
toUpdateInstances = new HashSet<>(ips);
// 用最新的服务实例列表覆盖旧的服务实例列表（CopyOnWrite思想）
if (ephemeral) {
ephemeralInstances = toUpdateInstances;
} else {
persistentInstances = toUpdateInstances;
}
}
/**
* 获取需要更新的服务实例列表
* @param newInstance 待处理服务实例列表
* @param oldInstance 旧服务实例列表
* @return
*/
private List<Instance> updatedIps(Collection<Instance> newInstance, Collection<Instance> oldInstance) {
// 获得待处理服务实例列表和旧服务实例列表的交集
List<Instance> intersects = (List<Instance>) CollectionUtils.intersection(newInstance, oldInstance);
// 将交集中的实例存储到stringIpAddressMap中，key：ip port，value：instance
Map<String, Instance> stringIpAddressMap = new ConcurrentHashMap<>(intersects.size());
for (Instance instance : intersects) {
stringIpAddressMap.put(instance.getIp() ":" instance.getPort(), instance);
}
// 定义交集Map，key：服务实例，value：标识
// 1-表示实例只存在于待处理服务实例列表或者旧服务实例列表中
// 2-表示实例同时存在于待处理服务实例列表和旧服务实例列表中
Map<String, Integer> intersectMap = new ConcurrentHashMap<>(newInstance.size() oldInstance.size());
// 定义需要更新的服务实例Map
Map<String, Instance> updatedInstancesMap = new ConcurrentHashMap<>(newInstance.size());
// 定义待处理的服务实例Map
Map<String, Instance> newInstancesMap = new ConcurrentHashMap<>(newInstance.size());
// 将旧服务实例列表中实例加入交集Map中，并设置实例的标识为1
for (Instance instance : oldInstance) {
if (stringIpAddressMap.containsKey(instance.getIp() ":" instance.getPort())) {
intersectMap.put(instance.toString(), 1);
}
}
// 将待处理服务实例列表中实例加入交集Map中，如果实例在交集Map中已经存在，则设置为2（即：替换），否则设置为1（即：新增）
for (Instance instance : newInstance) {
if (stringIpAddressMap.containsKey(instance.getIp() ":" instance.getPort())) {
if (intersectMap.containsKey(instance.toString())) {
intersectMap.put(instance.toString(), 2);
} else {
intersectMap.put(instance.toString(), 1);
}
}
// 将待处理实例加入newInstancesMap中（这一步很重要，后续用于新实例覆盖旧实例）
newInstancesMap.put(instance.toString(), instance);
}
// 遍历交集Map
for (Map.Entry<String, Integer> entry : intersectMap.entrySet()) {
String key = entry.getKey();
Integer value = entry.getValue();
// 将交集Map中所有标识为1（注意：标识为1的实例包含待处理实例和旧实例）且存在于待处理实例列表中的实例加入最新的服务实例列表中（即：需要更新的旧实例 新实例）
if (value == 1) {
if (newInstancesMap.containsKey(key)) {
updatedInstancesMap.put(key, newInstancesMap.get(key));
}
}
}
return new ArrayList<>(updatedInstancesMap.values());
}
删除操作
更新操作由Service#onDelete方法进行处理，该方法主要做以下事情：
- 更新服务对应的实例列表信息（即：从服务实例列表中移除实例）。
源码如下：
@Override
public void onDelete(String key) throws Exception {
boolean isEphemeral = KeyBuilder.matchEphemeralInstanceListKey(key);
for (Cluster each : clusterMap.values()) {
each.updateIps(Collections.emptyList(), isEphemeral);
}
}
// 其中each.updateIps方法就是更新操作源码中的updateIps方法。服务信息同步
在前面分析ConsistencyService#put方法时，会将服务信息封装成延时任务DelayTask（延时时长：1秒）添加到Nacos延时任务执行器（NacosDelayTaskExecuteEngine）的任务集合中。
NacosDelayTaskExecuteEngine在创建时会创建一个定时任务去执行NacosDelayTaskExecuteEngine#processTasks方法。
源码如下：
public NacosDelayTaskExecuteEngine(String name, int initCapacity, Logger logger, long processInterval) {
super(logger);
tasks = new ConcurrentHashMap<Object, AbstractDelayTask>(initCapacity);
// 创建一个单线程的线程池
processingExecutor = ExecutorFactory.newSingleScheduledExecutorService(new NameThreadFactory(name));
// 创建定时任务处理延时任务（延时时长：100ms，时间间隔：100ms）
processingExecutor
.scheduleWithFixedDelay(new ProcessRunnable(), processInterval, processInterval, TimeUnit.MILLISECONDS);
}
// com.alibaba.nacos.common.task.engine.NacosDelayTaskExecuteEngine.ProcessRunnable
// 延时任务处理线程
private class ProcessRunnable implements Runnable {
@Override
public void run() {
try {
// 处理任务，如：Nacos集群中节点之间的数据同步
processTasks();
} catch (Throwable e) {
getEngineLog().error(e.toString(), e);
}
}
}
NacosDelayTaskExecuteEngine#processTasks方法主要做以下事情：
- 根据DistroKey获取对应的延时任务。
- 根据taskKey获取对应的任务处理器。
- 执行任务处理器的processor()方法对任务进行处理。
- 任务处理失败，则将任务重新加入到任务集合中进行重试。
源码如下：
protected void processTasks() {
Collection<Object> keys = getAllTaskKeys();
for (Object taskKey : keys) {
// 根据taskKey（DistroKey）获取对应的延时任务
AbstractDelayTask task = removeTask(taskKey);
if (null == task) {
continue;
}
// 根据taskKey获取对应的任务处理器，默认为：TaskExecuteWorker
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);
}
}
}
private void retryFailedTask(Object key, AbstractDelayTask task) {
task.setLastProcessTime(System.currentTimeMillis());
addTask(key, task);
}
【阅读推荐】
更多精彩内容（如：Redis、数据结构与算法、Kafka等）请移步【南秋同学】个人主页进行查阅。
【作者简介】
一枚热爱技术和生活的老贝比，专注于Java领域，关注【南秋同学】带你一起学习成长～
,
