【Nacos】Nacos服务注册与发现 心跳检测机制源码剖析

在前两篇文章，介绍了springboot的主动设置原理，而nacos的服务注册就依赖主动设置原理。
Nacos

Nacos核心功能点

服务注册 :Nacos Client会通过发送REST哀求的方式向Nacos Server注册本身的服务，提供自身的元数据，好比ip地址、端口等信息。Nacos Server接收到注册哀求后，就会把这些元数据信息存储在一个双层的内存Map中。
服务心跳: 在服务注册后，Nacos Client会维护一个定时心跳来持续关照Nacos Server，说明服务不停处于可用状态，防止被剔除。默认5s发送一次心跳。
服务健康查抄: Nacos Server会开启一个定时使命用来查抄注册服务实例的健康情况，对于超过15s没有收到客户端心跳的实例会将它 的healthy属性置为false(客户端服务发现时不会发现)，如果某个实例超过30秒没有收到心跳，直接剔除该实例(被剔除的实比方果恢复 发送心跳则会重新注册)
服务发现: 服务消耗者(Nacos Client)在调用服务提供者的服务时，会发送一个REST哀求给Nacos Server，获取上面注册的服务清 单，并且缓存在Nacos Client本地，同时会在Nacos Client本地开启一个定时使命定时拉取服务端最新的注册表信息更新到本地缓存
服务同步: Nacos Server集群之间会互相同步服务实例，用来保证服务信息的一致性。
客户端 服务注册&心跳发送

在客户端中，也就是开发的应用，包含引入有 nacos-discovery 而路径下包含有一个spring.factories 在主动设置的时候，会加载
如下设置类。

1. org.springframework.boot.autoconfigure.EnableAutoConfiguration=\
2.   com.alibaba.cloud.nacos.discovery.NacosDiscoveryAutoConfiguration,\
3.   com.alibaba.cloud.nacos.ribbon.RibbonNacosAutoConfiguration,\
4.   com.alibaba.cloud.nacos.endpoint.NacosDiscoveryEndpointAutoConfiguration,\
5.   com.alibaba.cloud.nacos.registry.NacosServiceRegistryAutoConfiguration,\

复制代码

其中 NacosServiceRegistryAutoConfiguration 中引入了三个类 NacosServiceRegistry、NacosRegistration、NacosAutoServiceRegistration 其中 NacosAutoServiceRegistration 继承了 AbstractAutoServiceRegistration

1.         public void onApplicationEvent(WebServerInitializedEvent event) {
2.                 bind(event);
3.         }
4.         // 启动
5.         this.start();
6.         //注册服务
7.         register();

复制代码

组装实例信息，ip 端口 服务权重 集群名字 源信息 以及是否

1.         private Instance getNacosInstanceFromRegistration(Registration registration) {
2.                 Instance instance = new Instance();
3.                 instance.setIp(registration.getHost());
4.                 instance.setPort(registration.getPort());
5.                 instance.setWeight(nacosDiscoveryProperties.getWeight());
6.                 instance.setClusterName(nacosDiscoveryProperties.getClusterName());
7.                 instance.setEnabled(nacosDiscoveryProperties.isInstanceEnabled());
8.                 instance.setMetadata(registration.getMetadata());
9.                 instance.setEphemeral(nacosDiscoveryProperties.isEphemeral());
10.                 return instance;
11.         }

复制代码

1.         // 注册实例
2.         namingService.registerInstance(serviceId, group, instance);
4.         // 服务名 组名 实例
5.    public void registerInstance(String serviceName, String groupName, Instance instance) throws NacosException {
6.        // 实例非法校验
7.        NamingUtils.checkInstanceIsLegal(instance);
8.        // 获取组服务名
9.        String groupedServiceName = NamingUtils.getGroupedName(serviceName, groupName);
10.       
11.        if (instance.isEphemeral()) {
12.                        // 构建心跳信息
13.            BeatInfo beatInfo = beatReactor.buildBeatInfo(groupedServiceName, instance);
14.            // 启动一个心跳发送线程
15.            beatReactor.addBeatInfo(groupedServiceName, beatInfo);
16.        }
17.        // 实际就是调用 instance
18.        serverProxy.registerService(groupedServiceName, groupName, instance);
19.    }
21.         // 心跳发送线程
22.         executorService.schedule(new BeatTask(beatInfo), beatInfo.getPeriod(), TimeUnit.MILLISECONDS);

复制代码

发送心跳线程

所谓的客户端心跳，其实就是启动一个线程，然后定时给一个接口发送调用。

1.         class BeatTask implements Runnable {
2.         
3.         BeatInfo beatInfo;
4.         
5.         public BeatTask(BeatInfo beatInfo) {
6.             this.beatInfo = beatInfo;
7.         }
8.         
9.         @Override
10.         public void run() {
11.                 // 如果停止 直接返回
12.             if (beatInfo.isStopped()) {
13.                 return;
14.             }
15.             // 获取下次时间
16.             long nextTime = beatInfo.getPeriod();
17.             // 实际就是调用服务端的一个心跳接口  /instance/beat
18.             JsonNode result = serverProxy.sendBeat(beatInfo, BeatReactor.this.lightBeatEnabled);
19.             // 如果结束，启动另外一个 开始下次的心跳线程发送
20.             executorService.schedule(new BeatTask(beatInfo), nextTime, TimeUnit.MILLISECONDS);
21.         }
22.     }

复制代码

nacos 服务注册

对于服务端来说，就是一个API接口

1.         public String register(HttpServletRequest request) throws Exception {
2.         // 准备服务实例
3.         final Instance instance = parseInstance(request);
5.         serviceManager.registerInstance(namespaceId, serviceName, instance);
6.         return "ok";
7.     }

复制代码

注册实例其实就是三步、创建服务、获取服务、添加实例

1.     public void registerInstance(String namespaceId, String serviceName, Instance instance) throws NacosException {
2.         createEmptyService(namespaceId, serviceName, instance.isEphemeral());
3.         Service service = getService(namespaceId, serviceName);
4.         addInstance(namespaceId, serviceName, instance.isEphemeral(), instance);
5.     }

复制代码

创建服务

1.     public void createServiceIfAbsent(String namespaceId, String serviceName, boolean local, Cluster cluster)
2.             throws NacosException {
3.         // 命名空间 服务名称
4.         Service service = getService(namespaceId, serviceName);
5.         if (service == null) {
6.             service = new Service();
7.             service.setName(serviceName);
8.             service.setNamespaceId(namespaceId);
9.             service.setGroupName(NamingUtils.getGroupName(serviceName));
10.             // now validate the service. if failed, exception will be thrown
11.             service.setLastModifiedMillis(System.currentTimeMillis());
12.             service.recalculateChecksum();
13.             if (cluster != null) {
14.                 cluster.setService(service);
15.                 service.getClusterMap().put(cluster.getName(), cluster);
16.             }
17.             service.validate();
18.                         // 这里是重点
19.             putServiceAndInit(service);
20.             if (!local) {
21.                 addOrReplaceService(service);
22.             }
23.         }
24.     }
26.     private void putServiceAndInit(Service service) throws NacosException {
27.              // 添加服务
28.         putService(service);
29.         // 下面说心跳检测机制
30.         service.init();
31.     }

复制代码

这里其实就是底层的注册表的数据结构了，这里使用双查抄锁。分别是nameSpace、group、service、实例。
这里简单思索下，为什么要筹划这么复杂。一个服务可能对应多个实例。没有问题。一个分组group 可能是在同一个公司有差别的组，好比订单、支付，每个组都有本身的服务。namespace则可以分为dev\test\prod 三个差别的组。

1.         // 注册表  如何提升
2.     private final Map<String, Map<String, Service>> serviceMap = new ConcurrentHashMap<>();
4.         // 这里的机制其实就是 判断当前是否存在该nameSpaceId , 如果不存在的话，则创建一个CSLM
5.     public void putService(Service service) {
6.         if (!serviceMap.containsKey(service.getNamespaceId())) {
7.             // 添加锁 lock
8.             synchronized (putServiceLock) {
9.                 if (!serviceMap.containsKey(service.getNamespaceId())) {
10.                     serviceMap.put(service.getNamespaceId(), new ConcurrentSkipListMap<>());
11.                 }
12.             }
13.         }
14.         // 将添加服务添加到nameSpaceId 添加服务
15.         serviceMap.get(service.getNamespaceId()).put(service.getName(), service);
16.     }

复制代码

1.     // 添加实例
2.     public void addInstance(String namespaceId, String serviceName, boolean ephemeral, Instance... ips)
3.             throws NacosException {
5.         // 构建一个key
6.         String key = KeyBuilder.buildInstanceListKey(namespaceId, serviceName, ephemeral);
8.         Service service = getService(namespaceId, serviceName);
10.         synchronized (service) {
11.             List<Instance> instanceList = addIpAddresses(service, ephemeral, ips);
13.             Instances instances = new Instances();
14.             instances.setInstanceList(instanceList);
16.             // 持久化服务
17.             consistencyService.put(key, instances);
18.         }
19.     }
20.         // 这里根据判断是否ephemeral 走保存内存还是磁盘持久化
21.     private ConsistencyService mapConsistencyService(String key) {
22.         // AP CP
23.         return KeyBuilder.matchEphemeralKey(key) ? ephemeralConsistencyService : persistentConsistencyService;
24.     }

复制代码

这里其实选择的是 DistroConsistencyServiceImpl 另一个就是 RaftConsistencyServiceImpl 使用raft实现的数据持久化，这里先不介绍。
DistroConsistencyServiceImpl

1.     public void put(String key, Record value) throws NacosException {
2.         onPut(key, value);
3.         distroProtocol.sync(new DistroKey(key, KeyBuilder.INSTANCE_LIST_KEY_PREFIX), DataOperation.CHANGE,
4.                 globalConfig.getTaskDispatchPeriod() / 2);
5.     }
7.     // 这里调用put
8.         public void onPut(String key, Record value) {
9.         if (KeyBuilder.matchEphemeralInstanceListKey(key)) {
10.             Datum<Instances> datum = new Datum<>();
11.             datum.value = (Instances) value;
12.             datum.key = key;
13.             datum.timestamp.incrementAndGet();
14.             //将数据保存到内存中
15.             dataStore.put(key, datum);
16.         }
17.         if (!listeners.containsKey(key)) {
18.             return;
19.         }
20.         // 但是这里调用了一个任务
21.         notifier.addTask(key, DataOperation.CHANGE);
22.     }
23.         // 其实就是一个map
24.     private Map<String, Datum> dataMap = new ConcurrentHashMap<>(1024);
25.    
26.     public void put(String key, Datum value) {
27.         dataMap.put(key, value);
28.     }
30.     @PostConstruct
31.     public void init() {
32.         // 初始化 构造方法执行的时候 进行处理
33.         GlobalExecutor.submitDistroNotifyTask(notifier);
34.     }

复制代码

从源码中可以看到，在类初始化的时候，创建一个使命异步进行执行。 其实就是将当前服务进行异步使命注册，可以提拔性能。添加和获取使命。
源码英华：很多开源框架为了提拔操作性能会大量使用这种异步使命及内存队列操作，这些操作本省不需要写入立即返回乐成，用这种方式可以提拔操作性能很大资助

1.    public class Notifier implements Runnable {
2.                 // 保存服务
3.         private ConcurrentHashMap<String, String> services = new ConcurrentHashMap<>(10 * 1024);
4.                 // 阻塞队列
5.         private BlockingQueue<Pair<String, DataOperation>> tasks = new ArrayBlockingQueue<>(1024 * 1024);
6.                 // 初始化的时候 添加一个任务到阻塞队列中
7.         public void addTask(String datumKey, DataOperation action) {
8.             if (services.containsKey(datumKey) && action == DataOperation.CHANGE) {
9.                 return;
10.             }
11.             if (action == DataOperation.CHANGE) {
12.                 services.put(datumKey, StringUtils.EMPTY);
13.             }
14.             tasks.offer(Pair.with(datumKey, action));
15.         }
17.         @Override
18.         public void run() {
19.             // 为什么异步设计 : 提升性能
20.             // 阻塞队列 ,会线程等待 wait
21.             // 并发、反射、网络、IO
22.             for (; ; ) {
23.                       // 异步处理
24.                Pair<String, DataOperation> pair = tasks.take();
25.                handle(pair);
26.             }
27.         }
28.     }

复制代码

心跳检测机制

其实就是服务注册的时候 启动一个线程，然后查抄所有实例的心跳检测，对于超过15s没有收到客户端心跳的实例会将它 的healthy属性置为false(客户端服务发现时不会发现)，如果某个实例超过30秒没有收到心跳，直接剔除该实例(被剔除的实比方果恢复 发送心跳则会重新注册)

1.     /**
2.      * Init service.
3.      */
4.     public void init() {
5.         // 心跳检查线程
6.         HealthCheckReactor.scheduleCheck(clientBeatCheckTask);
7.         for (Map.Entry<String, Cluster> entry : clusterMap.entrySet()) {
8.             entry.getValue().setService(this);
9.             entry.getValue().init();
10.         }
11.     }
13.         // 初始化5S后执行，每5S执行一次
14.     public static void scheduleCheck(ClientBeatCheckTask task) {
15.         futureMap.putIfAbsent(task.taskKey(), GlobalExecutor.scheduleNamingHealth(task, 5000, 5000, TimeUnit.MILLISECONDS));
16.     }
18.     //
19.         public void run() {
20.         List<Instance> instances = service.allIPs(true);
22.             for (Instance instance : instances) {
23.                 // 当前时间 减去 心跳超时时间
24.                 if (System.currentTimeMillis() - instance.getLastBeat() > instance.getInstanceHeartBeatTimeOut()) {
25.                         if (instance.isHealthy()) {
26.                             // 健康状态
27.                             instance.setHealthy(false);
28.                             getPushService().serviceChanged(service);
29.                             ApplicationUtils.publishEvent(new InstanceHeartbeatTimeoutEvent(this, instance));
30.                         }
31.                     }
32.                 }
33.             }
35.             // then remove obsolete instances:
36.             for (Instance instance : instances) {
37.                 if (System.currentTimeMillis() - instance.getLastBeat() > instance.getIpDeleteTimeout()) {
38.                     // delete instance
39.                     Loggers.SRV_LOG.info("[AUTO-DELETE-IP] service: {}, ip: {}", service.getName(),
40.                             JacksonUtils.toJson(instance));
41.                     deleteIp(instance);
42.                 }
43.             }
44.         }
45.     }

复制代码

至此，我们就根本上过了一遍，服务的注册 以及心跳检测机制，本篇主要是针对nacos1.4.1 的源码学习，关于后续的服务发现，以及2.X版本的源码 讲解，后续在继续。

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。