经验首页 前端设计 程序设计 Java相关 移动开发 数据库/运维 软件/图像 大数据/云计算 其他经验
当前位置：技术经验 » 程序设计 » 编程经验 » 查看文章
Nacos源码阅读心得
来源：cnblogs　　作者：迷茫的bug　　时间：2023/10/25 17:35:26　　对本文有异议

Nacos注册中心（1.4.1）源码解读心得

一丶Nacos介绍

　　Nacos是阿里巴巴推出的一款新开源项目，是一个更易于构建云原生应用的动态服务发现、配置管理和服务管理平台。它致力于帮助您发现、配置和管理微服务，提供了一组简单易用的特性集，帮助您快速实现动态服务发现、服务配置、服务元数据及流量管理。Nacos是构建以“服务”为中心的现代应用架构（例如微服务范式、云原生范式）的服务基础设施。

Nacos支持多种核心特性，包括：

  1. 服务发现：支持DNS与RPC服务发现，也提供原生SDK、OpenAPI等多种服务注册方式和DNS、HTTP与API等多种服务发现方式。
  2. 服务健康监测：提供对服务的实时的健康检查，阻止向不健康的主机或服务实例发送请求。
  3. 动态配置服务：提供配置统一管理功能，能够帮助我们将配置以中心化、外部化和动态化的方式管理所有环境的应用配置和服务配置。
  4. 动态DNS服务：支持动态DNS服务权重路由，能够很容易地实现中间层负载均衡、更灵活的路由策略、流量控制以及数据中心内网的简单DNS解析服务。
  5. 服务及其元数据管理：支持从微服务平台建设的视角管理数据中心的所有服务及元数据，包括管理服务的描述、生命周期、服务的静态依赖分析、服务的健康状态、服务的流量管理、路由及安全策略、服务的SLA以及最首要的metrics统计数据。

　　Nacos可以与Spring、Spring Boot、Spring Cloud集成，并能代替Spring Cloud Eureka和Spring Cloud Config。通过Nacos Server和spring-cloud-starter-alibaba-nacos-config实现配置的动态变更。它提供了一个简洁易用的UI（控制台样例Demo）帮助您管理所有的服务和应用的配置。同时，它也提供了一些简单的DNS APIs TODO帮助您管理服务的关联域名和可用的IP:PORT列表。

二丶客户端注册流程

  • 在SpringCloudAlibaba这一套微服务组件中，Nacos作为注册中心向其他的微服务提供信息，业务服务通过从Nacos中拉取所需要的服务信息，再通过Ribbon在本地做负载均衡之后通过Feign组件发起接口调用。例如在电商系统中的下单服务，库存服务，支付服务等。完成一个下单过程中，下单服务需要调用库存服务减库存，调用支付服务完成支付等，而库存服务，支付服务的信息都会存储在注册中心即Nacos服务中。服务之间的调用只需要通过注册中心获取，不再需要每个服务都去存储需要调用的服务信息了，完成了解耦。
  • 客户端想要注册到注册中心去就要先引入Nacos的客户端依赖spring-cloud-starter-alibaba-nacos-discovery，并在配置文件中配上Nacos的服务地址和命名空间等信息。想要知道Nacos客户端的注册流程就得从引入的依赖入手，从maven依赖库中找到nacos的jar包，下面有一个META-INF文件夹，里面的spring.factory文件就是springboot自动装配过程中会装配的类:

     可以看到，红框标注的类就是跟服务发现自动装配相关性比较大的类文件了，直接在项目中搜索这个类可以看到以下文件：

    1. @Configuration
    2. @EnableConfigurationProperties
    3. @ConditionalOnNacosDiscoveryEnabled
    4. @ConditionalOnProperty(value = "spring.cloud.service-registry.auto-registration.enabled", matchIfMissing = true)
    5. @AutoConfigureAfter({ AutoServiceRegistrationConfiguration.class,
    6.         AutoServiceRegistrationAutoConfiguration.class })
    7. public class NacosDiscoveryAutoConfiguration {
    9.     @Bean
    10.     public NacosServiceRegistry nacosServiceRegistry(
    11.             NacosDiscoveryProperties nacosDiscoveryProperties) {
    12.         return new NacosServiceRegistry(nacosDiscoveryProperties);
    13.     }
    15.     @Bean
    16.     @ConditionalOnBean(AutoServiceRegistrationProperties.class)
    17.     public NacosRegistration nacosRegistration(
    18.             NacosDiscoveryProperties nacosDiscoveryProperties,
    19.             ApplicationContext context) {
    20.         return new NacosRegistration(nacosDiscoveryProperties, context);
    21.     }
    23.     @Bean
    24.     @ConditionalOnBean(AutoServiceRegistrationProperties.class)
    25.     public NacosAutoServiceRegistration nacosAutoServiceRegistration(
    26.             NacosServiceRegistry registry,
    27.             AutoServiceRegistrationProperties autoServiceRegistrationProperties,
    28.             NacosRegistration registration) {
    29.         return new NacosAutoServiceRegistration(registry,
    30.                 autoServiceRegistrationProperties, registration);
    31.     }
    32. }

    在这个类中有三个@Bean注解，仔细观察前两个bean都在第三个bean的参数中，所以第三个bean就是比较重要的Bean了。

  • 直接搜索NacosAutoServiceRegistry这个类可以看到其中有一个方法名为register(),通过类名和方法名大概能猜到这就是注册的主逻辑了。 
    1. @Override
    2.     protected void register() {
    3.         if (!this.registration.getNacosDiscoveryProperties().isRegisterEnabled()) {
    4.             log.debug("Registration disabled.");
    5.             return;
    6.         }
    7.         if (this.registration.getPort() < 0) {
    8.             this.registration.setPort(getPort().get());
    9.         }
    10.         super.register();
    11.     }
  • 直接点到注册方法中去是NacosAutoServiceRegistry的抽象父类，抽象父类中有一个对象serviceRegistry，注册方法就是这个对象的register()方法，继续跟到这个register()方法中去： 
    1. @Override
    2.     public void register(Registration registration) {
    4.         if (StringUtils.isEmpty(registration.getServiceId())) {
    5.             log.warn("No service to register for nacos client...");
    6.             return;
    7.         }
    9.         String serviceId = registration.getServiceId();
    11.         Instance instance = getNacosInstanceFromRegistration(registration);
    13.         try {
    14.             namingService.registerInstance(serviceId, instance);
    15.             log.info("nacos registry, {} {}:{} register finished", serviceId,
    16.                     instance.getIp(), instance.getPort());
    17.         }
    18.         catch (Exception e) {
    19.             log.error("nacos registry, {} register failed...{},", serviceId,
    20.                     registration.toString(), e);
    21.         }
    22.     }

    可以看到通过getNacosInstanceFromRegistration（）方法将Registration 转换成为了Instance 对象，后续通过namingService.registerInstance(serviceId, instance);进行了注册的动作，而这个Instance 对象其实就是我们Nacos服务端所存储的微服务相关的一些信息。

  • 继续跟namingService.registerInstance(serviceId, instance);方法： 
    1.  @Override
    2.     public void registerInstance(String serviceName, String groupName, Instance instance) throws NacosException {
    4.         if (instance.isEphemeral()) {
    5.             BeatInfo beatInfo = new BeatInfo();
    6.             beatInfo.setServiceName(NamingUtils.getGroupedName(serviceName, groupName));
    7.             beatInfo.setIp(instance.getIp());
    8.             beatInfo.setPort(instance.getPort());
    9.             beatInfo.setCluster(instance.getClusterName());
    10.             beatInfo.setWeight(instance.getWeight());
    11.             beatInfo.setMetadata(instance.getMetadata());
    12.             beatInfo.setScheduled(false);
    13.             long instanceInterval = instance.getInstanceHeartBeatInterval();
    14.             beatInfo.setPeriod(instanceInterval == 0 ? DEFAULT_HEART_BEAT_INTERVAL : instanceInterval);
    16.             beatReactor.addBeatInfo(NamingUtils.getGroupedName(serviceName, groupName), beatInfo);
    17.         }
    19.         serverProxy.registerService(NamingUtils.getGroupedName(serviceName, groupName), groupName, instance);
    20.     }

    这是NamingService的实现类NacosNamingService中的实现，可以看到进行了一个if判断，这个其实是判断是否是一个临时的实例，如果是临时实例做一些处理，最后的注册请求是在serverProxy.registerService(NamingUtils.getGroupedName(serviceName, groupName), groupName, instance);方法中的：

    1. public void registerService(String serviceName, String groupName, Instance instance) throws NacosException {
    3.         NAMING_LOGGER.info("[REGISTER-SERVICE] {} registering service {} with instance: {}",
    4.             namespaceId, serviceName, instance);
    6.         final Map<String, String> params = new HashMap<String, String>(9);
    7.         params.put(CommonParams.NAMESPACE_ID, namespaceId);
    8.         params.put(CommonParams.SERVICE_NAME, serviceName);
    9.         params.put(CommonParams.GROUP_NAME, groupName);
    10.         params.put(CommonParams.CLUSTER_NAME, instance.getClusterName());
    11.         params.put("ip", instance.getIp());
    12.         params.put("port", String.valueOf(instance.getPort()));
    13.         params.put("weight", String.valueOf(instance.getWeight()));
    14.         params.put("enable", String.valueOf(instance.isEnabled()));
    15.         params.put("healthy", String.valueOf(instance.isHealthy()));
    16.         params.put("ephemeral", String.valueOf(instance.isEphemeral()));
    17.         params.put("metadata", JSON.toJSONString(instance.getMetadata()));
    19.         reqAPI(UtilAndComs.NACOS_URL_INSTANCE, params, HttpMethod.POST);
    21.     }

    组装了一些参数最后发起了POST请求.其中UtilAndComs.NACOS_URL_INSTANCE这个常量最后拼接出来是/nacos/v1/ns/instance

  • 得出结论：客户端通过将自己服务的信息包括ip端口，命名空间，服务名等信息组装好，向Nacos服务发起POST请求注册到注册中心去。

三丶服务端存储客户端注册的服务信息

  • 　在上面客户端注册流程最后我们得到了一个URI：/nacos/v1/ns/instance  通过这个URI我们可以去服务端的源码中搜索这个接口： 
    1.  @CanDistro
    2.     @PostMapping
    3.     @Secured(parser = NamingResourceParser.class, action = ActionTypes.WRITE)
    4.     public String register(HttpServletRequest request) throws Exception {
    5.         
    6.         final String namespaceId = WebUtils
    7.                 .optional(request, CommonParams.NAMESPACE_ID, Constants.DEFAULT_NAMESPACE_ID);
    8.         final String serviceName = WebUtils.required(request, CommonParams.SERVICE_NAME);
    9.         NamingUtils.checkServiceNameFormat(serviceName);
    10.         
    11.         final Instance instance = parseInstance(request);
    12.         
    13.         serviceManager.registerInstance(namespaceId, serviceName, instance);
    14.         return "ok";
    15.     }

    以上是服务端的注册实例的接口，主要完成了三个动作：检查服务名信息，将请求中的参数转换为Instance实例对象，注册实例，继续跟 serviceManager.registerInstance(namespaceId, serviceName, instance)注册的方法：

    1. public void registerInstance(String namespaceId, String serviceName, Instance instance) throws NacosException {
    2.         
    3.         createEmptyService(namespaceId, serviceName, instance.isEphemeral());
    4.         
    5.         Service service = getService(namespaceId, serviceName);
    6.         
    7.         if (service == null) {
    8.             throw new NacosException(NacosException.INVALID_PARAM,
    9.                     "service not found, namespace: " + namespaceId + ", service: " + serviceName);
    10.         }
    11.         
    12.         addInstance(namespaceId, serviceName, instance.isEphemeral(), instance);
    13.     }

     

    这个方法中主要动作有：创建空的Service对象，获取Service对象，注册实例等动作。

     

    1. createEmptyService(namespaceId, serviceName, instance.isEphemeral())方法：
    1.  public void createServiceIfAbsent(String namespaceId, String serviceName, boolean local, Cluster cluster)
    2.             throws NacosException {
    3.         Service service = getService(namespaceId, serviceName);
    4.         if (service == null) {
    5.             
    6.             Loggers.SRV_LOG.info("creating empty service {}:{}", namespaceId, serviceName);
    7.             service = new Service();
    8.             service.setName(serviceName);
    9.             service.setNamespaceId(namespaceId);
    10.             service.setGroupName(NamingUtils.getGroupName(serviceName));
    11.             // now validate the service. if failed, exception will be thrown
    12.             service.setLastModifiedMillis(System.currentTimeMillis());
    13.             service.recalculateChecksum();
    14.             if (cluster != null) {
    15.                 cluster.setService(service);
    16.                 service.getClusterMap().put(cluster.getName(), cluster);
    17.             }
    18.             service.validate();
    19.             
    20.             putServiceAndInit(service);
    21.             if (!local) {
    22.                 addOrReplaceService(service);
    23.             }
    24.         }
    1.  

    先调用了getService方法：

    1. public Service getService(String namespaceId, String serviceName) {
    2.         if (serviceMap.get(namespaceId) == null) {
    3.             return null;
    4.         }
    5.         return chooseServiceMap(namespaceId).get(serviceName);
    6.     }

    其中的serviceMap结构是

    1. private final Map<String, Map<String, Service>> serviceMap = new ConcurrentHashMap<>();

    这个map其实就是注册中心保存所有实例的map，最外层的key一般为namespace,里层的key一般为定义的group名，根据业务需要定义。

    1. public class Service extends com.alibaba.nacos.api.naming.pojo.Service implements Record, RecordListener<Instances> {
    2.  ...private Map<String, Cluster> clusterMap = new HashMap<>();
    3.  ...
    4. }

    map中的Service对象其中又有一个clusterMap，而Cluster对象的结构如下

    1. public class Cluster extends com.alibaba.nacos.api.naming.pojo.Cluster implements Cloneable {
    2.     ...
    3.     
    4.     @JsonIgnore
    5.     private Set<Instance> persistentInstances = new HashSet<>();
    6.     
    7.     @JsonIgnore
    8.     private Set<Instance> ephemeralInstances = new HashSet<>();
    9.     
    10.     @JsonIgnore
    11.     private Service service;
    12. ...
    13. }

    看到这里的两个HashSet中的Instance对象是否有些眼熟？他就是客户端注册到注册中心的服务实例信息

  • 所以整体来看，服务端存储各个微服务的结构如图所示：

  •  那么再回到上述的getService方法中，通过命名空间就可以得到同一组下面的服务，而chooseServiceMap(namespaceId).get(serviceName)又通过服务名来获取Service，其实得到的结构就是一个个的service其中还有一层Cluster；

  • 那么此时我的微服务还并未完成注册获取到的Service肯定是Null,继续往下走就会新构建一个Service,经过前面的赋值校验方法，会走到putServiceAndInit(service)方法中去： 
    1. private void putServiceAndInit(Service service) throws NacosException {
    2.         putService(service);
    3.         service.init();
    4.         consistencyService
    5.                 .listen(KeyBuilder.buildInstanceListKey(service.getNamespaceId(), service.getName(), true), service);
    6.         consistencyService
    7.                 .listen(KeyBuilder.buildInstanceListKey(service.getNamespaceId(), service.getName(), false), service);
    8.         Loggers.SRV_LOG.info("[NEW-SERVICE] {}", service.toJson());
    9.     }

    这个方法中做了两步很重要的操作，其中putService()方法：

    1. public void putService(Service service) {
    2.         if (!serviceMap.containsKey(service.getNamespaceId())) {
    3.             synchronized (putServiceLock) {
    4.                 if (!serviceMap.containsKey(service.getNamespaceId())) {
    5.                     serviceMap.put(service.getNamespaceId(), new ConcurrentSkipListMap<>());
    6.                 }
    7.             }
    8.         }
    9.         serviceMap.get(service.getNamespaceId()).put(service.getName(), service);
    10.     }

    就是将我们的service放到serviceMap中，源码中就用到了双检锁。

  • 而在service.init()方法中： 
    1.  public void init() {
    2.         HealthCheckReactor.scheduleCheck(clientBeatCheckTask);
    3.         for (Map.Entry<String, Cluster> entry : clusterMap.entrySet()) {
    4.             entry.getValue().setService(this);
    5.             entry.getValue().init();
    6.         }
    7.     }

    第一行的 HealthCheckReactor.scheduleCheck(clientBeatCheckTask);看类名是健康检查相关的，那么可以想到注册实例肯定要把自己的健康信息更新到注册中心去，再看参数：clientBeatCheckTask 服务心跳检查任务，点到这个类中：

    1. public class ClientBeatCheckTask implements Runnable {
    2.     
    3.     private Service service;
    4.     
    5.     @Override
    6.     public void run() {
    7.         try {
    8.             if (!getDistroMapper().responsible(service.getName())) {
    9.                 return;
    10.             }
    11.             
    12.             if (!getSwitchDomain().isHealthCheckEnabled()) {
    13.                 return;
    14.             }
    15.             
    16.             List<Instance> instances = service.allIPs(true);
    17.             
    18.             // first set health status of instances:
    19.             for (Instance instance : instances) {
    20.                 if (System.currentTimeMillis() - instance.getLastBeat() > instance.getInstanceHeartBeatTimeOut()) {
    21.                     if (!instance.isMarked()) {
    22.                         if (instance.isHealthy()) {
    23.                             instance.setHealthy(false);
    24.                             Loggers.EVT_LOG
    25.                                     .info("{POS} {IP-DISABLED} valid: {}:{}@{}@{}, region: {}, msg: client timeout after {}, last beat: {}",
    26.                                             instance.getIp(), instance.getPort(), instance.getClusterName(),
    27.                                             service.getName(), UtilsAndCommons.LOCALHOST_SITE,
    28.                                             instance.getInstanceHeartBeatTimeOut(), instance.getLastBeat());
    29.                             getPushService().serviceChanged(service);
    30.                             ApplicationUtils.publishEvent(new InstanceHeartbeatTimeoutEvent(this, instance));
    31.                         }
    32.                     }
    33.                 }
    34.             }
    35.             
    36.             if (!getGlobalConfig().isExpireInstance()) {
    37.                 return;
    38.             }
    39.             
    40.             // then remove obsolete instances:
    41.             for (Instance instance : instances) {
    42.                 
    43.                 if (instance.isMarked()) {
    44.                     continue;
    45.                 }
    46.                 
    47.                 if (System.currentTimeMillis() - instance.getLastBeat() > instance.getIpDeleteTimeout()) {
    48.                     // delete instance
    49.                     Loggers.SRV_LOG.info("[AUTO-DELETE-IP] service: {}, ip: {}", service.getName(),
    50.                             JacksonUtils.toJson(instance));
    51.                     deleteIp(instance);
    52.                 }
    53.             }
    54.             
    55.         } catch (Exception e) {
    56.             Loggers.SRV_LOG.warn("Exception while processing client beat time out.", e);
    57.         }
    58.         
    59.     }
    60. }

    实现了Runnable接口说明是一个线程，直接看run方法

    1.  List<Instance> instances = service.allIPs(true);
    1. public List<Instance> allIPs(boolean ephemeral) {
    2.         List<Instance> result = new ArrayList<>();
    3.         for (Map.Entry<String, Cluster> entry : clusterMap.entrySet()) {
    4.             result.addAll(entry.getValue().allIPs(ephemeral));
    5.         }
    6.         
    7.         return result;
    8.     }
    1. public List<Instance> allIPs(boolean ephemeral) {
    2.         return ephemeral ? new ArrayList<>(ephemeralInstances) : new ArrayList<>(persistentInstances);
    3.     }

    获取了service中的所有Instance,然后做一系列的心跳检查，发布事件等。到此就做完了实例初始化的动作。

  • 上述是createEmptyService(namespaceId, serviceName, instance.isEphemeral())方法：
  • 那么接下来 
    1.  Service service = getService(namespaceId, serviceName);

    肯定可以获取到service，直接看最后的addInstance(namespaceId, serviceName, instance.isEphemeral(), instance)方法：

    1. public void addInstance(String namespaceId, String serviceName, boolean ephemeral, Instance... ips)
    2.             throws NacosException {
    3.         
    4.         String key = KeyBuilder.buildInstanceListKey(namespaceId, serviceName, ephemeral);
    5.         
    6.         Service service = getService(namespaceId, serviceName);
    7.         
    8.         synchronized (service) {
    9.             List<Instance> instanceList = addIpAddresses(service, ephemeral, ips);
    10.             
    11.             Instances instances = new Instances();
    12.             instances.setInstanceList(instanceList);
    13.             
    14.             consistencyService.put(key, instances);
    15.         }
    16.     }
    1. String key = KeyBuilder.buildInstanceListKey(namespaceId, serviceName, ephemeral)这个方法通过nameSpaceId,服务名信息组建了一个key:
    1. public static String buildInstanceListKey(String namespaceId, String serviceName, boolean ephemeral) {
    2.         return ephemeral ? buildEphemeralInstanceListKey(namespaceId, serviceName)
    3.                 : buildPersistentInstanceListKey(namespaceId, serviceName);
    4.     }
    1. 可以看到是根据ephemeral来判断的，这个值是控制是否临时实例的，Instance中的默认值是true表示默认新建的就是临时实例，那么构建出来的字符串："com.alibaba.nacos.naming.iplist.ephemeral."+ namespaceId +"##"+serviceName
    1. 中间是有ephemeral.的；下面又将初始化了的Instance获取到加锁执行注册逻辑：
    1. public List<Instance> updateIpAddresses(Service service, String action, boolean ephemeral, Instance... ips)
    2.             throws NacosException {
    3.         
    4.         Datum datum = consistencyService
    5.                 .get(KeyBuilder.buildInstanceListKey(service.getNamespaceId(), service.getName(), ephemeral));
    6.         
    7.         List<Instance> currentIPs = service.allIPs(ephemeral);
    8.         Map<String, Instance> currentInstances = new HashMap<>(currentIPs.size());
    9.         Set<String> currentInstanceIds = Sets.newHashSet();
    10.         
    11.         for (Instance instance : currentIPs) {
    12.             currentInstances.put(instance.toIpAddr(), instance);
    13.             currentInstanceIds.add(instance.getInstanceId());
    14.         }
    15.         
    16.         Map<String, Instance> instanceMap;
    17.         if (datum != null && null != datum.value) {
    18.             instanceMap = setValid(((Instances) datum.value).getInstanceList(), currentInstances);
    19.         } else {
    20.             instanceMap = new HashMap<>(ips.length);
    21.         }
    22.         
    23.         for (Instance instance : ips) {
    24.             if (!service.getClusterMap().containsKey(instance.getClusterName())) {
    25.                 Cluster cluster = new Cluster(instance.getClusterName(), service);
    26.                 cluster.init();
    27.                 service.getClusterMap().put(instance.getClusterName(), cluster);
    28.                 Loggers.SRV_LOG
    29.                         .warn("cluster: {} not found, ip: {}, will create new cluster with default configuration.",
    30.                                 instance.getClusterName(), instance.toJson());
    31.             }
    32.             
    33.             if (UtilsAndCommons.UPDATE_INSTANCE_ACTION_REMOVE.equals(action)) {
    34.                 instanceMap.remove(instance.getDatumKey());
    35.             } else {
    36.                 Instance oldInstance = instanceMap.get(instance.getDatumKey());
    37.                 if (oldInstance != null) {
    38.                     instance.setInstanceId(oldInstance.getInstanceId());
    39.                 } else {
    40.                     instance.setInstanceId(instance.generateInstanceId(currentInstanceIds));
    41.                 }
    42.                 instanceMap.put(instance.getDatumKey(), instance);
    43.             }
    44.             
    45.         }
    46.         
    47.         if (instanceMap.size() <= 0 && UtilsAndCommons.UPDATE_INSTANCE_ACTION_ADD.equals(action)) {
    48.             throw new IllegalArgumentException(
    49.                     "ip list can not be empty, service: " + service.getName() + ", ip list: " + JacksonUtils
    50.                             .toJson(instanceMap.values()));
    51.         }
    52.         
    53.         return new ArrayList<>(instanceMap.values());
    54.     }

    这个方法返回值是一个List<Instance>直接看返回值是instanceMap.values()，这个map的来源是通过第一个

    1. Datum datum = consistencyService .get(KeyBuilder.buildInstanceListKey(service.getNamespaceId(), service.getName(), ephemeral))

     

     

    的结果判断的，流程是先判断这个实例是否已经注册到注册中心了，没有的话新增一个instanceMap,有将其中的旧元素都放到新的instanceMap中，再对传过来的instance做一系列的检查注册操作，返回现有的instanceMap中的元素集合。

  • 通过返回的List<Instance>构建一个Instances对象，这个对象里面的结构： 
    1. public class Instances implements Record {
    2.     
    3.     private static final long serialVersionUID = 5500823673993740145L;
    4.     
    5.     private List<Instance> instanceList = new ArrayList<>();
      }

    最后一步consistencyService.put(key, instances)方法：

    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.     }

    onPut方法就是注册的核心逻辑了：

    1.  public void onPut(String key, Record value) {
    2.         
    3.         if (KeyBuilder.matchEphemeralInstanceListKey(key)) {
    4.             Datum<Instances> datum = new Datum<>();
    5.             datum.value = (Instances) value;
    6.             datum.key = key;
    7.             datum.timestamp.incrementAndGet();
    8.             dataStore.put(key, datum);
    9.         }
    10.         
    11.         if (!listeners.containsKey(key)) {
    12.             return;
    13.         }
    14.         
    15.         notifier.addTask(key, DataOperation.CHANGE);
    16.     }

    其中notifier的结构：

    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. }

    说明又是一个线程类，成员变量有一个tasks的阻塞队列。

    1.  public void addTask(String datumKey, DataOperation action) {
    2.             
    3.             if (services.containsKey(datumKey) && action == DataOperation.CHANGE) {
    4.                 return;
    5.             }
    6.             if (action == DataOperation.CHANGE) {
    7.                 services.put(datumKey, StringUtils.EMPTY);
    8.             }
    9.             tasks.offer(Pair.with(datumKey, action));
    10.         }

    通过观察addTask方法可以看出向上面提到的阻塞队列tasks中添加了一个对象：

    1. public static <A, B> Pair<A, B> with(A value0, B value1) {
    2.         return new Pair(value0, value1);
    3.     }

    这里的Pair.with()可以不用管，只需要知道返回的一个Pair对象包含了前面生成的key和action，action代表了操作类型是新增或者编辑删除之类的。

          那么这个阻塞队列里面就包含了我们的服务信息，服务名，nameSpaceId,是否是临时实例等。再回去看notifier的run方法：

  1. public void run() {
  2.             Loggers.DISTRO.info("distro notifier started");
  3.             
  4.             for (; ; ) {
  5.                 try {
  6.                     Pair<String, DataOperation> pair = tasks.take();
  7.                     handle(pair);
  8.                 } catch (Throwable e) {
  9.                     Loggers.DISTRO.error("[NACOS-DISTRO] Error while handling notifying task", e);
  10.                 }
  11.             }
  12.         }

  可以看出是一个线程从阻塞队列中循环拿出实例的信息到后续的handle方法：

  1. private void handle(Pair<String, DataOperation> pair) {
  2.             try {
  3.                 String datumKey = pair.getValue0();
  4.                 DataOperation action = pair.getValue1();
  5.                 
  6.                 services.remove(datumKey);
  7.                 
  8.                 int count = 0;
  9.                 
  10.                 if (!listeners.containsKey(datumKey)) {
  11.                     return;
  12.                 }
  13.                 
  14.                 for (RecordListener listener : listeners.get(datumKey)) {
  15.                     
  16.                     count++;
  17.                     
  18.                     try {
  19.                         if (action == DataOperation.CHANGE) {
  20.                             listener.onChange(datumKey, dataStore.get(datumKey).value);
  21.                             continue;
  22.                         }
  23.                         
  24.                         if (action == DataOperation.DELETE) {
  25.                             listener.onDelete(datumKey);
  26.                             continue;
  27.                         }
  28.                     } catch (Throwable e) {
  29.                         Loggers.DISTRO.error("[NACOS-DISTRO] error while notifying listener of key: {}", datumKey, e);
  30.                     }
  31.                 }
  32.                 
  33.                 if (Loggers.DISTRO.isDebugEnabled()) {
  34.                     Loggers.DISTRO
  35.                             .debug("[NACOS-DISTRO] datum change notified, key: {}, listener count: {}, action: {}",
  36.                                     datumKey, count, action.name());
  37.                 }
  38.             } catch (Throwable e) {
  39.                 Loggers.DISTRO.error("[NACOS-DISTRO] Error while handling notifying task", e);
  40.             }
  41.         }

 

  1.  @Override
  2.     public void onChange(String key, Instances value) throws Exception {
  3.         
  4.         Loggers.SRV_LOG.info("[NACOS-RAFT] datum is changed, key: {}, value: {}", key, value);
  5.         
  6.         for (Instance instance : value.getInstanceList()) {
  7.             
  8.             if (instance == null) {
  9.                 // Reject this abnormal instance list:
  10.                 throw new RuntimeException("got null instance " + key);
  11.             }
  12.             
  13.             if (instance.getWeight() > 10000.0D) {
  14.                 instance.setWeight(10000.0D);
  15.             }
  16.             
  17.             if (instance.getWeight() < 0.01D && instance.getWeight() > 0.0D) {
  18.                 instance.setWeight(0.01D);
  19.             }
  20.         }
  21.         
  22.         updateIPs(value.getInstanceList(), KeyBuilder.matchEphemeralInstanceListKey(key));
  23.         
  24.         recalculateChecksum();
  25.     }

在这里拿出所有的Instance实例对象进行权重默认值的设置，之后 updateIPs(value.getInstanceList(), KeyBuilder.matchEphemeralInstanceListKey(key))方法：

  1. public void updateIPs(Collection<Instance> instances, boolean ephemeral) {
  2.         Map<String, List<Instance>> ipMap = new HashMap<>(clusterMap.size());
  3.         for (String clusterName : clusterMap.keySet()) {
  4.             ipMap.put(clusterName, new ArrayList<>());
  5.         }
  6.         
  7.         for (Instance instance : instances) {
  8.             try {
  9.                 if (instance == null) {
  10.                     Loggers.SRV_LOG.error("[NACOS-DOM] received malformed ip: null");
  11.                     continue;
  12.                 }
  13.                 
  14.                 if (StringUtils.isEmpty(instance.getClusterName())) {
  15.                     instance.setClusterName(UtilsAndCommons.DEFAULT_CLUSTER_NAME);
  16.                 }
  17.                 
  18.                 if (!clusterMap.containsKey(instance.getClusterName())) {
  19.                     Loggers.SRV_LOG
  20.                             .warn("cluster: {} not found, ip: {}, will create new cluster with default configuration.",
  21.                                     instance.getClusterName(), instance.toJson());
  22.                     Cluster cluster = new Cluster(instance.getClusterName(), this);
  23.                     cluster.init();
  24.                     getClusterMap().put(instance.getClusterName(), cluster);
  25.                 }
  26.                 
  27.                 List<Instance> clusterIPs = ipMap.get(instance.getClusterName());
  28.                 if (clusterIPs == null) {
  29.                     clusterIPs = new LinkedList<>();
  30.                     ipMap.put(instance.getClusterName(), clusterIPs);
  31.                 }
  32.                 
  33.                 clusterIPs.add(instance);
  34.             } catch (Exception e) {
  35.                 Loggers.SRV_LOG.error("[NACOS-DOM] failed to process ip: " + instance, e);
  36.             }
  37.         }
  38.         
  39.         for (Map.Entry<String, List<Instance>> entry : ipMap.entrySet()) {
  40.             //make every ip mine
  41.             List<Instance> entryIPs = entry.getValue();
  42.             clusterMap.get(entry.getKey()).updateIps(entryIPs, ephemeral);
  43.         }
  44.         
  45.         setLastModifiedMillis(System.currentTimeMillis());
  46.         getPushService().serviceChanged(this);
  47.         StringBuilder stringBuilder = new StringBuilder();
  48.         
  49.         for (Instance instance : allIPs()) {
  50.             stringBuilder.append(instance.toIpAddr()).append("_").append(instance.isHealthy()).append(",");
  51.         }
  52.         
  53.         Loggers.EVT_LOG.info("[IP-UPDATED] namespace: {}, service: {}, ips: {}", getNamespaceId(), getName(),
  54.                 stringBuilder.toString());
  55.         
  56.     }

 

其中将Instance实例放入Map中的逻辑：

  1. if (!clusterMap.containsKey(instance.getClusterName())) {
  2.                     Loggers.SRV_LOG
  3.                             .warn("cluster: {} not found, ip: {}, will create new cluster with default configuration.",
  4.                                     instance.getClusterName(), instance.toJson());
  5.                     Cluster cluster = new Cluster(instance.getClusterName(), this);
  6.                     cluster.init();
  7.                     getClusterMap().put(instance.getClusterName(), cluster);
  8.                 }

可以得出结论：Nacos中有一个线程从阻塞队列notifier中拿出新注册的Instance初始化处理之后根据是否是临时实例放到对应的HashSet中，即保存了新注册的实例信息。而每次注册的结果就是将注册信息放入到阻塞队列中去；这种异步处理的方式使得Nacos的TPS可以达到1w3+，因为其保存的服务信息对实时性要求并不高，这种场景下使用异步处理是合适的。而Nacos在更新实例信息时采用了写时复制的思想，保证了服务上下线修改Map时的效率。由于其写时复制复制的其实是Cluster中的内容，不是复制整个Map，所以它的效率也是很高的。

附一张图灵学院诸葛老师总结的流程图：

 

原文链接：https://www.cnblogs.com/ybug/p/17787741.html

 友情链接：直通硅谷 　点职佳 　北美留学生论坛

编程经验热门文章

  • Prometheus监控之SNMP Exporter介绍和数据展现
  • Intel 82599网卡异常挂死原因
  • 腾讯防水墙的分析识别与破解
  • 桌面应用自动化WinAppDriver入门
  • 文档在线预览（三）使用js前端实现word、excel、pdf、ppt 在线预览
  • 「教程」如何下载AcFun网站上的视频、A站视频下载工具
  • 基于Hi3559AV100的SVP(NNIE)开发整体流程
  • GBDT(梯度提升树)scikit-klearn中的参数说明及简汇
  • antd之RangePicker设置默认值方式
  • 决胜经典算法之插入排序

    • 编程经验推荐文章

  • wifi基础(一)：无线电波与WIFI信号干扰、衰减
  • 甲方扔给两个存在包名与类名均相同的Jar包，要在工程中同时使用怎么办？
  • 《花100块做个摸鱼小网站! 》第三篇—热搜表结构设计和热搜数据存储
  • Fluent Editor：一个基于 Quill 2.0 的富文本编辑器，功能强大、开箱即用！
  • 方法的三种调用形式
  • 《花100块做个摸鱼小网站! 》第二篇—后端应用搭建和完成第一个爬虫
  • 《熬夜整理》保姆级系列教程-玩转Wireshark抓包神器教程(4)-再识Wireshark - 北京-宏哥
  • 《软件性能测试分析与调优实践之路》(第2版) 读书笔记（二）总体介绍（下）-真正从性能分析与调优来看性能测试
  • 架构知识点（三）
  • 《花100块做个摸鱼小网站! 》第一篇—买云服务器和初始化环境

    • 本站QQ群：前端 618073944 | Java 606181507 | Python 626812652 | C/C++ 612253063 | 微信 634508462 | 苹果 692586424 | C#/.net 182808419 | PHP 305140648 | 运维 608723728

    W3xue 的所有内容仅供测试，对任何法律问题及风险不承担任何责任。通过使用本站内容随之而来的风险与本站无关。
    关于我们  |  意见建议  |  捐助我们  |  报错有奖  |  广告合作、友情链接（目前9元/月）请联系QQ：27243702 沸活量
    皖ICP备17017327号-2　皖公网安备34020702000426号