前言
再讲Nacos之前，先来讲一下服务注册和发现。我们知道，现在微服务架构是目前开发的一个趋势。服务消费者要去调用多个服务提供者组成的集群。这里需要做到以下几点：
-
1、服务消费者需要在本地配置文件中维护服务提供者集群的每个节点的请求地址。
-
2、服务提供者集群中如果某个节点宕机，服务消费者的本地配置中需要同步删除这个节点的请求地址，防止请求发送到已经宕机的节点上造成请求失败。
因此需要引入服务注册中心，它具有以下几个功能：
- 1、服务地址的管理。
- 2、服务注册。
- 3、服务动态感知。
一、Nacos介绍
Nacos致力于解决微服务中的统一配置，服务注册和发现等问题。Nacos集成了注册中心和配置中心。其相关特性包括：
- 1、服务发现和服务健康监测。
Nacos支持基于DNS和RPC的服务发现，即服务消费者可以使用DNS或者HTTP的方式来查找和发现服务。 Nacos提供对服务的实时的健康检查，阻止向不健康的主机或者服务实例发送请求。Nacos支持传输层（Ping/TCP）、应用层（HTTP、Mysql）的健康检查。
- 2、动态配置服务。
动态配置服务可以以中心化、外部化和动态化的方式管理所有环境的应用配置和服务配置。
- 3、动态DNS服务。
支持权重路由，让开发者更容易的实现中间层的负载均衡、更灵活的路由策略、流量控制以及DNS解析服务。
- 4、服务和元数据管理。
Nacos允许开发者从微服务平台建设的视角来管理数据中心的所有服务和元数据。如：服务的生命周期、静态依赖分析、服务的健康状态、服务的流量管理、路由和安全策略等。
二、Nacos注册中心实现原理分析
2.1 Nacos架构图
以下是Nacos的架构图：
其中分为这么几个模块：
- Provider APP：服务提供者。
- Consumer APP：服务消费者。
- Name Server：通过Virtual IP或者DNS的方式实现Nacos高可用集群的服务路由。
- Nacos Server：Nacos服务提供者。
- Nacos Console：Nacos控制台。
Nacos Server其中包含：
- OpenAPI：功能访问入口。
- Config Service、Naming Service：Nacos提供的配置服务、名字服务模块。
- Consistency Protocol：一致性协议，用来实现Nacos集群节点的数据同步，使用Raft算法实现。
小总结：
-
服务提供者通过VIP（Virtual IP）访问Nacos Server高可用集群，基于OpenAPI完成服务的注册和服务的查询。
-
Nacos Server的底层则通过数据一致性算法（Raft）来完成节点的数据同步。
2.2 注册中心的原理
这里对其原理做一个大致的介绍，在后文则从源码角度进行分析。
首先，服务注册的功能体现在：
- 服务实例启动时注册到服务注册表、关闭时则注销（服务注册）。
- 服务消费者可以通过查询服务注册表来获得可用的实例（服务发现）。
- 服务注册中心需要调用服务实例的健康检查API来验证其是否可以正确的处理请求（健康检查）。
大致流程：每个服务都会有一个nacos client，它用来和nacos server打交道，用来具体的服务注册、查询等操作，服务提供者在启动的时候会向nacos server注册自己，服务消费者在启动的时候订阅nacos server上的服务提供者。
Nacos服务注册和发现的实现原理的图如下：
三、服务注册
首先需要引入spring-cloud-starter-alibaba-nacos-discovery包
<parent>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-parent</artifactId>
<version>2.3.12.RELEASE</version>
<relativePath/> <!-- lookup parent from repository -->
</parent>
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<dependency>
<groupId>com.alibaba.cloud</groupId>
<artifactId>spring-cloud-starter-alibaba-nacos-discovery</artifactId>
</dependency>
</dependencies>
<dependencyManagement>
<dependencies>
<dependency>
<groupId>com.alibaba.cloud</groupId>
<artifactId>spring-cloud-alibaba-dependencies</artifactId>
<version>2.2.6.RELEASE</version>
<type>pom</type>
<scope>import</scope>
</dependency>
</dependencies>
</dependencyManagement>
-
根据spring.factories配置来完成相关类的自动注册。
-
重点来看这几个类，看名称可猜到是用来服务注册的，NacosServiceRegistryAutoConfiguration用来注册管理这几个bean。
@Configuration(proxyBeanMethods = false)
@EnableConfigurationProperties
@ConditionalOnNacosDiscoveryEnabled
@ConditionalOnProperty(value = "spring.cloud.service-registry.auto-registration.enabled",
matchIfMissing = true)
@AutoConfigureAfter({ AutoServiceRegistrationConfiguration.class,
AutoServiceRegistrationAutoConfiguration.class,
NacosDiscoveryAutoConfiguration.class })
public class NacosServiceRegistryAutoConfiguration {
@Bean
public NacosServiceRegistry nacosServiceRegistry(
NacosDiscoveryProperties nacosDiscoveryProperties) {
return new NacosServiceRegistry(nacosDiscoveryProperties);
}
@Bean
@ConditionalOnBean(AutoServiceRegistrationProperties.class)
public NacosRegistration nacosRegistration(
ObjectProvider<List<NacosRegistrationCustomizer>> registrationCustomizers,
NacosDiscoveryProperties nacosDiscoveryProperties,
ApplicationContext context) {
return new NacosRegistration(registrationCustomizers.getIfAvailable(),
nacosDiscoveryProperties, context);
}
@Bean
@ConditionalOnBean(AutoServiceRegistrationProperties.class)
public NacosAutoServiceRegistration nacosAutoServiceRegistration(
NacosServiceRegistry registry,
AutoServiceRegistrationProperties autoServiceRegistrationProperties,
NacosRegistration registration) {
return new NacosAutoServiceRegistration(registry,
autoServiceRegistrationProperties, registration);
}
}
-
NacosServiceRegistry：完成服务注册，实现ServiceRegistry。
-
NacosRegistration：用来注册时存储nacos服务端的相关信息。
-
NacosAutoServiceRegistration 继承spring中的AbstractAutoServiceRegistration，AbstractAutoServiceRegistration实现ApplicationListener<WebServerInitializedEvent>，通过事件监听来发起服务注册，到时候会调用NacosServiceRegistry.register(registration)
来看具体如何注册
/*************************************************NacosServiceRegistry**************************************************/
public class NacosServiceRegistry implements ServiceRegistry<Registration> {
@Override
public void register(Registration registration) {
if (StringUtils.isEmpty(registration.getServiceId())) {
log.warn("No service to register for nacos client...");
return;
}
NamingService namingService = namingService();
String serviceId = registration.getServiceId();
String group = nacosDiscoveryProperties.getGroup();
Instance instance = getNacosInstanceFromRegistration(registration);
try {
namingService.registerInstance(serviceId, group, instance);
log.info("nacos registry, {} {} {}:{} register finished", group, serviceId,
instance.getIp(), instance.getPort());
}
catch (Exception e) {
if (nacosDiscoveryProperties.isFailFast()) {
log.error("nacos registry, {} register failed...{},", serviceId,
registration.toString(), e);
rethrowRuntimeException(e);
}
else {
log.warn("Failfast is false. {} register failed...{},", serviceId,
registration.toString(), e);
}
}
}
}
/**************************************************NacosNamingService************************************************/
public class NacosNamingService implements NamingService {
@Override
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);
// 添加心跳检测
beatReactor.addBeatInfo(groupedServiceName, beatInfo);
}
// 完成服务注册
serverProxy.registerService(groupedServiceName, groupName, instance);
}
}
/***************************************************BeatReactor***************************************************/
public class BeatReactor implements Closeable {
private final ScheduledExecutorService executorService;
public final Map<String, BeatInfo> dom2Beat = new ConcurrentHashMap<String, BeatInfo>();
public void addBeatInfo(String serviceName, BeatInfo beatInfo) {
NAMING_LOGGER.info("[BEAT] adding beat: {} to beat map.", beatInfo);
String key = buildKey(serviceName, beatInfo.getIp(), beatInfo.getPort());
BeatInfo existBeat = null;
//fix #1733
if ((existBeat = dom2Beat.remove(key)) != null) {
existBeat.setStopped(true);
}
dom2Beat.put(key, beatInfo);
// 发起一个心跳检测任务
executorService.schedule(new BeatTask(beatInfo), beatInfo.getPeriod(), TimeUnit.MILLISECONDS);
MetricsMonitor.getDom2BeatSizeMonitor().set(dom2Beat.size());
}
/******************************************************BeatTask******************************************************/
class BeatTask implements Runnable {
@Override
public void run() {
if (beatInfo.isStopped()) {
return;
}
long nextTime = beatInfo.getPeriod();
try {
// 向nacos服务发起心跳检测
JsonNode result = serverProxy.sendBeat(beatInfo, BeatReactor.this.lightBeatEnabled);
long interval = result.get("clientBeatInterval").asLong();
boolean lightBeatEnabled = false;
if (result.has(CommonParams.LIGHT_BEAT_ENABLED)) {
lightBeatEnabled = result.get(CommonParams.LIGHT_BEAT_ENABLED).asBoolean();
}
BeatReactor.this.lightBeatEnabled = lightBeatEnabled;
if (interval > 0) {
nextTime = interval;
}
int code = NamingResponseCode.OK;
if (result.has(CommonParams.CODE)) {
code = result.get(CommonParams.CODE).asInt();
}
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) {
}
}
} catch (NacosException ex) {
NAMING_LOGGER.error("[CLIENT-BEAT] failed to send beat: {}, code: {}, msg: {}",
JacksonUtils.toJson(beatInfo), ex.getErrCode(), ex.getErrMsg());
} catch (Exception unknownEx) {
NAMING_LOGGER.error("[CLIENT-BEAT] failed to send beat: {}, unknown exception msg: {}",
JacksonUtils.toJson(beatInfo), unknownEx.getMessage(), unknownEx);
} finally {
// 发起下一次心跳检测
executorService.schedule(new BeatTask(beatInfo), nextTime, TimeUnit.MILLISECONDS);
}
}
}
}
服务提供者向nacos server发起服务注册前，先向nacos server建立起心跳检测机制，nacos server那边也有一个心跳检测，服务提供者不停的向nacos server发起心跳检测，告知自己的健康状态，nacos server发现该服务心跳检测时间超时会发布超时事件来告知服务消费者。
服务发现
服务发现由NacosWatch完成，它实现了Spring的Lifecycle接口，容器启动和销毁时会调用对应的start()和stop()方法。
来看对应源码
public class NacosWatch implements ApplicationEventPublisherAware, SmartLifecycle {
@Override
public void start() {
// cas设置运行状态为true
if (this.running.compareAndSet(false, true)) {
EventListener eventListener = listenerMap.computeIfAbsent(buildKey(),
event -> new EventListener() {
@Override
public void onEvent(Event event) {
if (event instanceof NamingEvent) {
List<Instance> instances = ((NamingEvent) event)
.getInstances();
Optional<Instance> instanceOptional = selectCurrentInstance(
instances);
instanceOptional.ifPresent(currentInstance -> {
resetIfNeeded(currentInstance);
});
}
}
});
// 获取nacos server上最新的服务提供者们
NamingService namingService = nacosServiceManager
.getNamingService(properties.getNacosProperties());
try {
// 订阅服务 并对每个服务都添加一个心跳检测监听
namingService.subscribe(properties.getService(), properties.getGroup(),
Arrays.asList(properties.getClusterName()), eventListener);
}
catch (Exception e) {
log.error("namingService subscribe failed, properties:{}", properties, e);
}
// 延时执行一个服务发现任务
this.watchFuture = this.taskScheduler.scheduleWithFixedDelay(
this::nacosServicesWatch, this.properties.getWatchDelay());
}
}
@Override
public void stop(Runnable callback) {
this.stop();
callback.run();
}
@Override
public void stop() {
// 设置运行状态为false 然后取消正在执行的任务
if (this.running.compareAndSet(true, false)) {
if (this.watchFuture != null) {
// shutdown current user-thread,
// then the other daemon-threads will terminate automatic.
this.taskScheduler.shutdown();
this.watchFuture.cancel(true);
}
EventListener eventListener = listenerMap.get(buildKey());
try {
NamingService namingService = nacosServiceManager
.getNamingService(properties.getNacosProperties());
// 取消已经下线的服务订阅，发起取消订阅操作并删除订阅监听
namingService.unsubscribe(properties.getService(), properties.getGroup(),
Arrays.asList(properties.getClusterName()), eventListener);
}
catch (Exception e) {
log.error("namingService unsubscribe failed, properties:{}", properties,
e);
}
}
}
public void nacosServicesWatch() {
// nacos doesn't support watch now , publish an event every 30 seconds.
// nacos不支持立即通知，每30秒发布一个事件
this.publisher.publishEvent(
new HeartbeatEvent(this, nacosWatchIndex.getAndIncrement()));
}
}
大致流程：nacos client这边在spring容器启动后执行一个服务订阅操作的延时任务，这个任务执行时先拉取nacos server那边最新的服务列表，然后与本地缓存的服务列表进行比较，取消订阅下线的服务，然后每隔30秒向nacos server发起订阅操作，订阅所有服务。
服务消费者如何实时感知服务提供者的状态信息呢？
- 1、服务消费者订阅后会执行一个轮询任务（每10s执行一次）用来拉取最新的服务提供者信息并实时更新,实现在HostReactor中的UpdateTask完成，下面来看代码
public class HostReactor implements Closeable {
public class UpdateTask implements Runnable {
long lastRefTime = Long.MAX_VALUE;
private final String clusters;
private final String serviceName;
/**
* the fail situation. 1:can't connect to server 2:serviceInfo's hosts is empty
*/
private int failCount = 0;
public UpdateTask(String serviceName, String clusters) {
this.serviceName = serviceName;
this.clusters = clusters;
}
private void incFailCount() {
int limit = 6;
if (failCount == limit) {
return;
}
failCount++;
}
private void resetFailCount() {
failCount = 0;
}
@Override
public void run() {
long delayTime = DEFAULT_DELAY;
try {
// 拿到当前的服务信息
ServiceInfo serviceObj = serviceInfoMap.get(ServiceInfo.getKey(serviceName, clusters));
//如果为null,说明本地没有，需要从服务端获取
if (serviceObj == null) {
//拉取最新的服务列表随后更新
updateService(serviceName, clusters);
return;
}
// 当前服务未及时更新 进行更新操作
//判断服务是否已过期，当前服务的最后一次更新时间 <= 全局的最后一次更新
if (serviceObj.getLastRefTime() <= lastRefTime) {
//调用updateService从服务端获取地址列表，更新服务列表
updateService(serviceName, clusters);
serviceObj = serviceInfoMap.get(ServiceInfo.getKey(serviceName, clusters));
} else {
// if serviceName already updated by push, we should not override it
// since the push data may be different from pull through force push
//如果服务已经被基于push机制的情况下做了更新，那么我们不需要覆盖本地服务。
//因为push过来的数据和pull数据不同，所以这里只是调用请求去刷新服务
refreshOnly(serviceName, clusters);
}
// 设置服务最新的更新时间
lastRefTime = serviceObj.getLastRefTime();
// 订阅被取消，如果没有实现订阅或者futureMap中不包含指定服务信息，则中断更新请求
if (!notifier.isSubscribed(serviceName, clusters) && !futureMap
.containsKey(ServiceInfo.getKey(serviceName, clusters))) {
// abort the update task
NAMING_LOGGER.info("update task is stopped, service:" + serviceName + ", clusters:" + clusters);
return;
}
if (CollectionUtils.isEmpty(serviceObj.getHosts())) {
incFailCount();
return;
}
delayTime = serviceObj.getCacheMillis();
resetFailCount();
} catch (Throwable e) {
incFailCount();
NAMING_LOGGER.warn("[NA] failed to update serviceName: " + serviceName, e);
} finally {
// 继续下一次轮询 延后10s执行，实现重复轮询
executor.schedule(this, Math.min(delayTime << failCount, DEFAULT_DELAY * 60), TimeUnit.MILLISECONDS);
}
}
}
}
- 2、上面服务注册时说过，服务提供者注册时nacos服务端也有一个相应的心跳检测，当心跳检测超时也就是未及时收到服务提供者的心跳包，nacos server判定该服务状态异常，随后通过UDP推送服务信息用来告知对应服务消费者，服务消费者通过PushReceiver来处理udp协议，HostReactor.processServiceJson(String json)来更新本地服务列表。
public class PushReceiver implements Runnable, Closeable {
private static final Charset UTF_8 = Charset.forName("UTF-8");
private static final int UDP_MSS = 64 * 1024;
private ScheduledExecutorService executorService;
private DatagramSocket udpSocket;
private HostReactor hostReactor;
private volatile boolean closed = false;
public PushReceiver(HostReactor hostReactor) {
try {
this.hostReactor = hostReactor;
String udpPort = getPushReceiverUdpPort();
if (StringUtils.isEmpty(udpPort)) {
this.udpSocket = new DatagramSocket();
} else {
this.udpSocket = new DatagramSocket(new InetSocketAddress(Integer.parseInt(udpPort)));
}
//开启一个线程池，不断接受服务端传递过来的数据
this.executorService = new ScheduledThreadPoolExecutor(1, new ThreadFactory() {
@Override
public Thread newThread(Runnable r) {
Thread thread = new Thread(r);
thread.setDaemon(true);
thread.setName("com.alibaba.nacos.naming.push.receiver");
return thread;
}
});
//调用run方法
this.executorService.execute(this);
} catch (Exception e) {
NAMING_LOGGER.error("[NA] init udp socket failed", e);
}
}
@Override
public void run() {
//通过while循环不断监听客户端Nacos Server传递过来的数据，实现一个Push的机制
while (!closed) {
try {
// byte[] is initialized with 0 full filled by default
byte[] buffer = new byte[UDP_MSS];
DatagramPacket packet = new DatagramPacket(buffer, buffer.length);
//初始化一个监听，不断接受客户端Nacos Server传递过来的数据
udpSocket.receive(packet);
String json = new String(IoUtils.tryDecompress(packet.getData()), UTF_8).trim();
NAMING_LOGGER.info("received push data: " + json + " from " + packet.getAddress().toString());
PushPacket pushPacket = JacksonUtils.toObj(json, PushPacket.class);
String ack;
if ("dom".equals(pushPacket.type) || "service".equals(pushPacket.type)) {
// 处理变更信息
hostReactor.processServiceJson(pushPacket.data);
// send ack to server
ack = "{\"type\": \"push-ack\"" + ", \"lastRefTime\":\"" + pushPacket.lastRefTime + "\", \"data\":"
+ "\"\"}";
} else if ("dump".equals(pushPacket.type)) {
// dump data to server
ack = "{\"type\": \"dump-ack\"" + ", \"lastRefTime\": \"" + pushPacket.lastRefTime + "\", \"data\":"
+ "\"" + StringUtils.escapeJavaScript(JacksonUtils.toJson(hostReactor.getServiceInfoMap()))
+ "\"}";
} else {
// do nothing send ack only
ack = "{\"type\": \"unknown-ack\"" + ", \"lastRefTime\":\"" + pushPacket.lastRefTime
+ "\", \"data\":" + "\"\"}";
}
udpSocket.send(new DatagramPacket(ack.getBytes(UTF_8), ack.getBytes(UTF_8).length,
packet.getSocketAddress()));
} catch (Exception e) {
if (closed) {
return;
}
NAMING_LOGGER.error("[NA] error while receiving push data", e);
}
}
}
}
参考： https://www.cnblogs.com/zzz-blogs/p/14243912.html
https://blog.csdn.net/xingxinggua9620/article/details/113403062
https://blog.csdn.net/Zong_0915/article/details/113001226
标签：服务,Spring,nacos,Nacos,beatInfo,instance,Alibaba,public From： https://blog.51cto.com/u_14014612/6017404