今天我们来深入剖析 Nacos 作为注册中心的底层实现原理，从源码层面解析服务注册到服务发现的完整流程。
1. Nacos 简介
在深入探讨 Nacos 之前，我们先来理解服务注册与发现的基本概念。在当前的微服务架构趋势下，服务消费者需要调用由多个服务提供者组成的集群。这通常面临两个挑战：
- 服务消费者需要在本地配置中维护服务提供者集群每个节点的请求地址。
- 当服务提供者集群中的某个节点宕机时，服务消费者的本地配置需要同步删除该节点的地址，以防止请求发送至故障节点导致失败。
因此，引入服务注册中心成为必然，其主要功能包括：
Nacos 致力于解决微服务中的统一配置、服务注册与发现等问题，它集成了注册中心与配置中心。其核心特性包括：
1. 服务发现与健康监测
Nacos 支持基于 DNS 和 RPC 的服务发现，服务消费者可以使用 DNS 或 HTTP 方式查找服务。它提供对服务的实时健康检查，阻止请求被发送到不健康的实例。Nacos 支持传输层（Ping/TCP）和应用层（HTTP、MySQL）的健康检查。
2. 动态配置服务
动态配置服务允许以中心化、外部化和动态化的方式管理所有环境的应用配置和服务配置。
3. 动态 DNS 服务
支持权重路由，让开发者更容易实现负载均衡、灵活的路由策略、流量控制以及 DNS 解析服务。
4. 服务与元数据管理
Nacos 允许从微服务平台建设的视角管理数据中心的所有服务与元数据，例如服务的生命周期、静态依赖分析、健康状态、流量管理、路由和安全策略等。
2. Nacos 注册中心实现原理分析
2.1 Nacos 架构概览
下图清晰地展示了 Nacos 的核心架构：
架构主要包含以下几个模块：
- Provider APP：服务提供者。
- Consumer APP：服务消费者。
- Name Server：通过 Virtual IP 或 DNS 的方式实现 Nacos 高可用集群的服务路由。
- Nacos Server：Nacos 服务提供者。
- OpenAPI：功能访问入口。
- Config Service, Naming Service：Nacos 提供的配置服务、名字服务模块。
- Consistency Protocol：一致性协议，用于实现 Nacos 集群节点的数据同步，采用 Raft 算法。
- Nacos Console：Nacos 控制台。
简要总结：
- 服务提供者通过 VIP 访问 Nacos Server 高可用集群，基于 OpenAPI 完成服务注册与查询。
- Nacos Server 底层通过数据一致性算法（Raft）来完成集群节点的数据同步。
2.2 注册中心基本原理
注册中心的核心功能体现在：
- 服务注册：服务实例启动时注册到服务注册表，关闭时注销。
- 服务发现：服务消费者通过查询服务注册表获取可用实例。
- 健康检查：注册中心需要调用服务实例的健康检查 API 来验证其能否正确处理请求。
Nacos 服务注册与发现的原理如下图所示：
3. Nacos 源码深度解析
接下来，我们将从服务注册和服务发现两个角度，深入源码分析 Nacos 的实现机制。
3.1 Nacos 服务注册流程
首先，我们关注 spring-cloud-commons 这个包。其中的 ServiceRegistry 接口是 Spring Cloud 定义的服务注册标准，任何希望集成到 Spring Cloud 生态中实现服务注册的组件，都需要实现这个接口。
接口定义如下：
public interface ServiceRegistry<R extends Registration> {
void register(R registration);
void deregister(R registration);
void close();
void setStatus(R registration, String status);
<T> T getStatus(R registration);
}
对于 Nacos 而言，该接口的具体实现类是 NacosServiceRegistry，它位于 spring-cloud-alibaba-nacos-discovery 包中。
回到 spring-cloud-commons 包，其中的 spring.factories 文件包含了自动装配的配置信息。
该文件配置了自动装配类，如图所示：
项目启动时，会根据此文件导入相应的自动配置类，并对相关属性进行自动装配。这里导入了 AutoServiceRegistrationAutoConfiguration 类，顾名思义，它是服务注册相关的自动配置类。
该配置类的完整代码如下：
@Configuration(
proxyBeanMethods = false
)
@Import({AutoServiceRegistrationConfiguration.class})
@ConditionalOnProperty(
value = {"spring.cloud.service-registry.auto-registration.enabled"},
matchIfMissing = true
)
public class AutoServiceRegistrationAutoConfiguration {
@Autowired(
required = false
)
private AutoServiceRegistration autoServiceRegistration;
@Autowired
private AutoServiceRegistrationProperties properties;
public AutoServiceRegistrationAutoConfiguration() {
}
@PostConstruct
protected void init() {
if (this.autoServiceRegistration == null && this.properties.isFailFast()) {
throw new IllegalStateException("Auto Service Registration has been requested, but there is no AutoServiceRegistration bean");
}
}
}
AutoServiceRegistrationAutoConfiguration 中注入了 AutoServiceRegistration 实例，其类关系图如下：
我们重点看其抽象父类 AbstractAutoServiceRegistration：
public abstract class AbstractAutoServiceRegistration<R extends Registration> implements AutoServiceRegistration,
ApplicationContextAware,
ApplicationListener<WebServerInitializedEvent> {
public void onApplicationEvent(WebServerInitializedEvent event) {
this.bind(event);
}
}
该类实现了 ApplicationListener 接口，并监听 WebServerInitializedEvent 事件。这意味着：
NacosAutoServiceRegistration 会监听 WebServerInitializedEvent 事件。- 当 Web 服务器初始化完成后，会触发事件，调用
onApplicationEvent() 方法，该方法最终会调用 NacosServiceRegistry 的 register() 方法（NacosServiceRegistry 实现了 Spring 的服务注册标准接口 ServiceRegistry）。
register() 方法的核心是调用 Nacos Client SDK 中 NamingService 的 registerInstance() 方法来完成服务注册。
public void register(Registration registration) {
if (StringUtils.isEmpty(registration.getServiceId())) {
log.warn("No service to register for nacos client...");
} else {
String serviceId = registration.getServiceId();
String group = this.nacosDiscoveryProperties.getGroup();
Instance instance = this.getNacosInstanceFromRegistration(registration);
try {
this.namingService.registerInstance(serviceId, group, instance);
log.info("nacos registry, {} {} {}:{} register finished", new Object[]{group, serviceId, instance.getIp(), instance.getPort()});
} catch (Exception var6) {
log.error("nacos registry, {} register failed...{},", new Object[]{serviceId, registration.toString(), var6});
ReflectionUtils.rethrowRuntimeException(var6);
}
}
}
public void registerInstance(String serviceName, String groupName, Instance instance) throws NacosException {
if (instance.isEphemeral()) {
BeatInfo beatInfo = new BeatInfo();
beatInfo.setServiceName(NamingUtils.getGroupedName(serviceName, groupName));
beatInfo.setIp(instance.getIp());
beatInfo.setPort(instance.getPort());
beatInfo.setCluster(instance.getClusterName());
beatInfo.setWeight(instance.getWeight());
beatInfo.setMetadata(instance.getMetadata());
beatInfo.setScheduled(false);
long instanceInterval = instance.getInstanceHeartBeatInterval();
beatInfo.setPeriod(instanceInterval == 0L ? DEFAULT_HEART_BEAT_INTERVAL : instanceInterval);
// 1.addBeatInfo() 负责创建心跳信息实现健康监测。Nacos Server 必须确保注册的服务实例是健康的，心跳监测是服务健康监测的一种手段。
this.beatReactor.addBeatInfo(NamingUtils.getGroupedName(serviceName, groupName), beatInfo);
}
// 2.registerService() 实现服务的注册
this.serverProxy.registerService(NamingUtils.getGroupedName(serviceName, groupName), groupName, instance);
}
接着看心跳监测方法 addBeatInfo()：
public void addBeatInfo(String serviceName, BeatInfo beatInfo) {
LogUtils.NAMING_LOGGER.info("[BEAT] adding beat: {} to beat map.", beatInfo);
String key = this.buildKey(serviceName, beatInfo.getIp(), beatInfo.getPort());
BeatInfo existBeat = null;
if ((existBeat = (BeatInfo)this.dom2Beat.remove(key)) != null) {
existBeat.setStopped(true);
}
this.dom2Beat.put(key, beatInfo);
// 通过 schedule() 方法，定时向服务端发送数据包，并启动线程持续检测服务端回应。
// 如果在指定时间内未收到服务端回应，则认为服务器出现故障。
// 参数1：可以理解为这个实例的相关信息（BeatTask）。
// 参数2：一个 long 类型的时间，代表从现在开始推迟执行的时间，默认是 5000。
// 参数3：时间单位，默认是毫秒，结合 5000 即代表每 5 秒发送一次心跳数据包。
this.executorService.schedule(new BeatReactor.BeatTask(beatInfo), beatInfo.getPeriod(), TimeUnit.MILLISECONDS);
MetricsMonitor.getDom2BeatSizeMonitor().set((double)this.dom2Beat.size());
}
当心跳检查正常，代表待注册的服务是健康的，随后执行注册方法 registerService()：
public void registerService(String serviceName, String groupName, Instance instance) throws NacosException {
LogUtils.NAMING_LOGGER.info("[REGISTER-SERVICE] {} registering service {} with instance: {}", new Object[]{this.namespaceId, serviceName, instance});
Map<String, String> params = new HashMap(9);
params.put("namespaceId", this.namespaceId);
params.put("serviceName", serviceName);
params.put("groupName", groupName);
params.put("clusterName", 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", JSON.toJSONString(instance.getMetadata()));
// 可以看出，这里将服务实例的必要参数存入 Map，然后通过 OpenAPI 的方式发送 POST 注册请求。
this.reqAPI(UtilAndComs.NACOS_URL_INSTANCE, params, (String)"POST");
}
服务注册流程总结
我们可以通过几个问答来梳理整个框架：
问题1：Nacos 的服务注册为什么与 spring-cloud-commons 包相关？
回答：
- Nacos 服务注册依赖
spring-cloud-starter-alibaba-nacos-discovery 包。
- 该包依赖了
spring-cloud-commons 包。
spring-cloud-commons 中定义了服务注册标准接口 ServiceRegistry，任何集成到 Spring Cloud 的服务注册组件都需要实现此接口。- 因此，Nacos 的具体实现类为
NacosServiceRegistry。
问题2：为什么项目添加了依赖，服务启动时却没有注册到 Nacos？
回答：
- 本文提到，Nacos 服务注册涉及对
WebServerInitializedEvent 事件的监听。
- 因此，该项目必须是一个 Web 项目。
- 请检查 pom 文件中是否依赖了
spring-boot-starter-web。
问题3：spring-cloud-commons 包还有什么作用？
回答：
- 该包下的
spring.factories 文件配置了服务注册的自动配置类，支持自动装配。
- 这个配置类是
AutoServiceRegistrationAutoConfiguration，它注入了 AutoServiceRegistration，而 NacosAutoServiceRegistration 是其具体实现。
- 当 Web 服务器初始化时，通过绑定的事件监听器，会触发并执行服务注册逻辑。
本质上，spring-cloud-commons 做了两件事：
- 引入 Spring 事件监听机制，在容器初始化后触发 Nacos 服务注册。
- 定义了
ServiceRegistry 接口，Nacos 的服务注册实现需要遵循此规范。
Nacos 服务注册完整流程梳理：
- 服务（项目）启动时，根据
spring-cloud-commons 中 spring.factories 的配置，自动装配 AutoServiceRegistrationAutoConfiguration 类。
AutoServiceRegistrationAutoConfiguration 类中注入 AutoServiceRegistration，其最终实现子类 (NacosAutoServiceRegistration) 实现了 Spring 的 ApplicationListener 监听器。- Web 服务初始化完成后，发布
WebServerInitializedEvent 事件，监听器触发，调用服务注册方法，即 NacosServiceRegistry 的 register() 方法。
- 该方法主要调用 Nacos Client SDK 中
NamingService 的 registerInstance() 方法来完成服务注册。
registerInstance() 方法主要做两件事：服务实例的健康监测和实例的注册。- 通过
schedule() 方法定时发送心跳数据包，检测实例健康状态。
- 若健康监测通过，调用
registerService() 方法，通过 OpenAPI 方式执行服务注册，将实例 Instance 的相关信息存入 Map 并发起 HTTP 请求。
3.2 Nacos 服务发现流程
我们需要明确 Nacos 服务发现发生的时机。通常，在进行微服务远程接口调用时会发生服务发现。例如，使用 OpenFeign 进行远程调用时，需要指定微服务名称，这个名称就是用于服务发现的关键。
Nacos 在进行服务发现时，会调用 NacosServerList 类下的 getServers() 方法：
public class NacosServerList extends AbstractServerList<NacosServer> {
private List<NacosServer> getServers() {
try {
String group = this.discoveryProperties.getGroup();
// 1. 通过唯一的 serviceId（通常是服务名称）和组名来获取对应的所有实例。
List<Instance> instances = this.discoveryProperties.namingServiceInstance().selectInstances(this.serviceId, group, true);
// 2. 将 List<Instance> 转换成 List<NacosServer> 数据，然后返回。
return this.instancesToServerList(instances);
} catch (Exception var3) {
throw new IllegalStateException("Can not get service instances from nacos, serviceId=" + this.serviceId, var3);
}
}
}
接下来看 NacosNamingService.selectInstances() 方法：
public List<Instance> selectInstances(String serviceName, String groupName, boolean healthy) throws NacosException {
return this.selectInstances(serviceName, groupName, healthy, true);
}
该方法最终会调用其重载方法：
public List<Instance> selectInstances(String serviceName, String groupName, List<String> clusters,
boolean healthy, boolean subscribe) throws NacosException {
// 保存服务实例信息的对象
ServiceInfo serviceInfo;
// 如果该消费者订阅了这个服务，那么会在本地维护一个服务列表，服务从本地获取
if (subscribe) {
serviceInfo = this.hostReactor.getServiceInfo(NamingUtils.getGroupedName(serviceName, groupName), StringUtils.join(clusters, ","));
} else {
// 否则实例会从 Nacos 服务中心进行获取。
serviceInfo = this.hostReactor.getServiceInfoDirectlyFromServer(NamingUtils.getGroupedName(serviceName, groupName), StringUtils.join(clusters, ","));
}
return this.selectInstances(serviceInfo, healthy);
}
这里应该重点关注 this.hostReactor 对象，它内部有几个重要的 Map 存储结构：
public class HostReactor {
private static final long DEFAULT_DELAY = 1000L;
private static final long UPDATE_HOLD_INTERVAL = 5000L;
// 存放异步更新任务的回调结果（Future）
private final Map<String, ScheduledFuture<?>> futureMap;
// 本地已存在的服务列表缓存，key是服务名称，value是ServiceInfo对象
private Map<String, ServiceInfo> serviceInfoMap;
// 待更新的实例列表
private Map<String, Object> updatingMap;
// 定时任务执行器（负责服务列表的实时更新）
private ScheduledExecutorService executor;
....
}
再看它的 getServiceInfo() 方法：
public ServiceInfo getServiceInfo(String serviceName, String clusters) {
LogUtils.NAMING_LOGGER.debug("failover-mode: " + this.failoverReactor.isFailoverSwitch());
String key = ServiceInfo.getKey(serviceName, clusters);
if (this.failoverReactor.isFailoverSwitch()) {
return this.failoverReactor.getService(key);
} else {
// 1.先通过serviceName即服务名获得一个serviceInfo
ServiceInfo serviceObj = this.getServiceInfo0(serviceName, clusters);
// 如果没有serviceInfo，则根据参数创建一个新的serviceInfo对象，并同时维护到本地缓存Map和更新状态Map（serviceInfoMap和updatingMap）
if (null == serviceObj) {
serviceObj = new ServiceInfo(serviceName, clusters);
this.serviceInfoMap.put(serviceObj.getKey(), serviceObj);
this.updatingMap.put(serviceName, new Object());
// 2. updateServiceNow()，立刻去Nacos服务端拉取数据。
this.updateServiceNow(serviceName, clusters);
this.updatingMap.remove(serviceName);
} else if (this.updatingMap.containsKey(serviceName)) {
synchronized(serviceObj) {
try {
serviceObj.wait(5000L);
} catch (InterruptedException var8) {
LogUtils.NAMING_LOGGER.error("[getServiceInfo] serviceName:" + serviceName + ", clusters:" + clusters, var8);
}
}
}
// 3. 定时更新实例信息
this.scheduleUpdateIfAbsent(serviceName, clusters);
// 最后返回服务实例数据（前面已经进行了更新或等待更新完成）
return (ServiceInfo)this.serviceInfoMap.get(serviceObj.getKey());
}
}
来看下 scheduleUpdateIfAbsent() 方法：
// 通过心跳的方式，异步定时（默认每10秒）去更新一次数据，并不是只有在调用服务时才更新。
public void scheduleUpdateIfAbsent(String serviceName, String clusters) {
if (this.futureMap.get(ServiceInfo.getKey(serviceName, clusters)) == null) {
synchronized(this.futureMap) {
if (this.futureMap.get(ServiceInfo.getKey(serviceName, clusters)) == null) {
// 创建一个UpdateTask更新线程任务，定期（如每10秒）去异步更新缓存数据
ScheduledFuture<?> future = this.addTask(new HostReactor.UpdateTask(serviceName, clusters));
this.futureMap.put(ServiceInfo.getKey(serviceName, clusters), future);
}
}
}
}
服务发现流程总结
常有人说 Nacos 的一个优点是，当某个服务实例宕机后，短时间内不会对消费者造成太大影响，因为客户端维护了一个本地服务列表（缓存）。其原理如下：
- Nacos 的服务发现，通常通过订阅的形式来获取服务数据。
- 通过订阅的方式，数据主要从本地的服务注册列表缓存（serviceInfoMap）中获取。如果不订阅，则每次都会从 Nacos 服务端直接获取，此时要求对应服务必须是健康的（否则无法获取）。
- 在代码设计上，通过多个 Map 结构来存放和协调实例数据，例如用
serviceInfoMap 缓存数据，用 updatingMap 控制更新并发，用 futureMap 管理定时更新任务。
Nacos 服务发现完整流程梳理：
- 以调用远程接口（如 OpenFeign）为例，当执行远程调用时，会触发服务发现过程。
- 服务发现先执行
NacosServerList 类中的 getServers() 方法，将远程调用接口上 @FeignClient 注解中的服务名作为 serviceId，传入 NacosNamingService.selectInstances() 方法中。
- 根据
subscribe 参数的值来决定是从本地缓存列表获取服务，还是直接从 Nacos 服务端获取（通常为 true，走本地缓存）。
- 以从本地缓存获取为例，会调用
HostReactor.getServiceInfo() 来获取服务信息 (ServiceInfo)。Nacos 客户端通过 3 个 Map 共同维护本地缓存：
- 本地缓存 Map ->
serviceInfoMap
- 更新状态 Map ->
updatingMap
- 异步任务 Map ->
futureMap
最终结果从 serviceInfoMap 中获取。
HostReactor.getServiceInfo() 方法通过 this.scheduleUpdateIfAbsent() 和 updateServiceNow() 方法实现服务的定时异步更新和立即更新。scheduleUpdateIfAbsent() 方法通过线程池提交异步更新任务 (UpdateTask)，将任务结果 (ScheduledFuture) 保存到 futureMap 中。该任务定期执行，负责更新本地缓存 (serviceInfoMap) 中的数据，保持与注册中心同步。
结语
本文从架构和源码层面详细解析了 Nacos 作为服务注册中心的核心机制。理解服务注册的自动装配、事件监听流程，以及服务发现的本地缓存、定时更新策略，对于设计和排查微服务架构中的问题至关重要。掌握这些原理，能帮助开发者更好地运用 Nacos，构建高可用的分布式系统。如果你想深入了解更多关于微服务架构和 Spring Boot 的实践，欢迎到 云栈社区 的 后端 & 架构 板块与其他开发者交流探讨。
发表于 3 天前
|
查看: 14|
回复: 0
