Nacos系列【26】源码分析篇之客户端自动注册

た 入场券 2024-04-08 11:16 238阅读 0赞

有道无术，术尚可求，有术无道，止于术。

版本說明：

spring-boot：2.6.3

spring-cloud：2021.0.1

spring-cloud-alibaba：2.2.8.RELEASE

Nacos：2.1.1

文章目录

    • 前言
      1. 加载日志系统
      1. 启动Grpc 客户端
      1. 自动配置类
      1. NacosWatch
      1. 自动注册

前言

紧接上篇，接下来分析下客户端自动注册的流程。

1. 加载日志系统

Spring Boot开启启动，会加载spring.factories中的自动配置类：

  • NacosDiscoveryAutoConfiguration
  • RibbonNacosAutoConfiguration
  • NacosDiscoveryEndpointAutoConfiguration
  • NacosServiceRegistryAutoConfiguration
  • NacosReactiveDiscoveryClientConfiguration
  • NacosConfigServerAutoConfiguration
  • NacosDiscoveryClientConfigServiceBootstrapConfiguration
  • NacosDiscoveryAutoConfiguration

在环境准备阶段，NacosLoggingListener监听器首先工作，因为它监听的事件是ApplicationEnvironmentPreparedEvent，其作用是加载日志系统：

  1. public void onApplicationEvent(ApplicationEvent applicationEvent) {
  3.         // 如果存在`logback`相关的类，加载logback日志系统LogbackNacosLogging，否则使用Log4J2NacosLogging，然后加载配置
  4.         NacosLogging.getInstance().loadConfiguration();
  5.     }

在加载配置的最后阶段，可以看到其读取了classpath:nacos-logback.xm日志配置文件，并创建了一个LoggerContext日志上下文：

  1. private LoggerContext loadConfigurationOnStart() {
  3.         String location = this.getLocation("classpath:nacos-logback.xml");
  4.         try {
  6.             LoggerContext loggerContext = (LoggerContext)StaticLoggerBinder.getSingleton().getLoggerFactory();
  7.             NacosJoranConfigurator configurator = new NacosJoranConfigurator();
  8.             configurator.setContext(loggerContext);
  9.             configurator.doNacosConfigure(ResourceUtils.getResourceUrl(location));
  10.             return loggerContext;
  11.         } catch (Exception var4) {
  13.             throw new IllegalStateException("Could not initialize Logback Nacos logging from " + location, var4);
  14.         }
  15.     }

而这个时候，当前应用对应的日志系统都还没有加载，所以Nacos客户端的一些日志级别配置，需要在源码配置文件中修改，或者通过JVM启动参数才能起效：
在这里插入图片描述

2. 启动Grpc 客户端

Nacos 2.x采用Grpc通信，gRPC是由google开发的一个高性能、通用的开源RPC框架，主要面向移动应用开发且基于HTTP/2协议标准而设计，同时支持大多数流行的编程语言。

使用gRPC框架，需要开启客户端和服务端，可以在GitHub学习下，这里就不再研究了。

GrpcSdkClient就是Nacos 进行通信的客户端，利用Grpc来实现服务器连接、消息处理等功能，启动逻辑在其start()方法中。在方法中会创建一些定时任务，以及和Nacos服务端中的GrpcSdkServer建立长连接。

3. 自动配置类

接着开始进行自动配置，注入配置类及Bean 对象，首先是服务管理器：
在这里插入图片描述
接着是配置类：
在这里插入图片描述
服务发现功能类：
在这里插入图片描述
服务发现客户端：
在这里插入图片描述
整个生命周期监测者NacosWatch：
在这里插入图片描述
等等。。。其他类就不一一介绍了。

4. NacosWatch

在NacosDiscoveryClientConfiguration自动配置类中，注册了NacosWatch。
在这里插入图片描述
调用NacosWatch构造方法：

  1. public NacosWatch(NacosServiceManager nacosServiceManager, NacosDiscoveryProperties properties) {
  3.         // 服务管理器
  4.         this.nacosServiceManager = nacosServiceManager;
  5.         // YML 配置
  6.         this.properties = properties;
  7.         // 多线程定时任务
  8.         this.taskScheduler = getTaskScheduler();
  9.     }

NacosWatch实现了ApplicationEventPublisherAware接口，作用是获取初始化事件发布器：
在这里插入图片描述
NacosWatch还实现了SmartLifecycle，该接口主要是作用是所有的bean都创建完成之后，可以执行初始化操作，在退出时执行资源销毁工作。
在这里插入图片描述
所以在应用启动后，执行重写了SmartLifecycle的start()方法：

  1. public void start() {
  3.         if (this.running.compareAndSet(false, true)) {
  5.             // 添加名称为app-service001:DEFAULT_GROUP 的监听器 
  6.             EventListener eventListener = (EventListener)this.listenerMap.computeIfAbsent(this.buildKey(), (event) -> {
  8.                 return new EventListener() {
  10.                     public void onEvent(Event event) {
  12.                         if (event instanceof NamingEvent) {
  14.                             List instances = ((NamingEvent)event).getInstances();
  15.                             Optional instanceOptional = NacosWatch.this.selectCurrentInstance(instances);
  16.                             instanceOptional.ifPresent((currentInstance) -> {
  18.                                 NacosWatch.this.resetIfNeeded(currentInstance);
  19.                             });
  20.                         }
  22.                     }
  23.                 };
  24.             });
  25.             // 获取NamingService
  26.             NamingService namingService = this.nacosServiceManager.getNamingService(this.properties.getNacosProperties());
  28.             try {
  30.                 // 订阅事件
  31.                 namingService.subscribe(this.properties.getService(), this.properties.getGroup(), Arrays.asList(this.properties.getClusterName()), eventListener);
  32.             } catch (Exception var4) {
  34.                 log.error("namingService subscribe failed, properties:{}", this.properties, var4);
  35.             }
  36.             // 定时任务去发布心跳事件
  37.             this.watchFuture = this.taskScheduler.scheduleWithFixedDelay(this::nacosServicesWatch, this.properties.getWatchDelay());
  38.         }
  40.     }

5. 自动注册

自动注册由NacosAutoServiceRegistration类来完成，继承了AbstractAutoServiceRegistration抽象类，类关系图如下：
在这里插入图片描述
实现了监听器接口，监听WebServerInitializedEvent事件，所以在Web容器初始化完成后，会进入到onApplicationEvent 方法：

  1. public void onApplicationEvent(WebServerInitializedEvent event) {
  3.         this.bind(event);
  4.     }

在这里插入图片描述
接着进入到AbstractAutoServiceRegistration的bind()方法：

  1. @Deprecated
  2.     public void bind(WebServerInitializedEvent event) {
  4.         // 获取上下文
  5.         ApplicationContext context = event.getApplicationContext();
  6.         if (!(context instanceof ConfigurableWebServerApplicationContext) || !"management".equals(((ConfigurableWebServerApplicationContext)context).getServerNamespace())) {
  8.             // 设置端口为程序启动端口
  9.             this.port.compareAndSet(0, event.getWebServer().getPort());
  10.             // 调用开始方法：
  11.             this.start();
  12.         }
  13.     }

在start()方法中，调用注册方法，并发布一些事件和日志：

  1. public void start() {
  3.         // 没有开启自动注册，打印debug 日志
  4.         if (!this.isEnabled()) {
  6.             if (logger.isDebugEnabled()) {
  8.                 logger.debug("Discovery Lifecycle disabled. Not starting");
  9.             }
  11.         } else {
  13.             if (!this.running.get()) {
  14.      // 开始标记，最初为false
  15.                 // 发布实例预注册事件
  16.                 this.context.publishEvent(new InstancePreRegisteredEvent(this, this.getRegistration()));
  17.                 // 开始注册
  18.                 this.register();
  19.                 if (this.shouldRegisterManagement()) {
  21.                     this.registerManagement();
  22.                 }
  23.                 // 发布实例注册完成事件
  24.                 this.context.publishEvent(new InstanceRegisteredEvent(this, this.getConfiguration()));
  25.                 this.running.compareAndSet(false, true);
  26.             }
  28.         }
  29.     }
  30.     // 调用服务注册器进行注册
  31.     protected void register() {
  33.         this.serviceRegistry.register(this.getRegistration());
  34.     }

注册方法，调用的是NacosServiceRegistry的register(Registration registration)方法：

  1. public void register(Registration registration) {
  3.         // 服务名为空，打印警告日志
  4.         if (StringUtils.isEmpty(registration.getServiceId())) {
  6.             log.warn("No service to register for nacos client...");
  7.         } else {
  9.             NamingService namingService = this.namingService();// // NamingService
  10.             String serviceId = registration.getServiceId(); // 服务名=>app-service001
  11.             String group = this.nacosDiscoveryProperties.getGroup();// 配置的分组，默认 DEFAULT_GROUP
  12.             // 创建实例对象
  13.             Instance instance = this.getNacosInstanceFromRegistration(registration);
  14.             try {
  16.                 // 调用NamingService 注册实例
  17.                 namingService.registerInstance(serviceId, group, instance);
  18.                 log.info("nacos registry, {} {} {}:{} register finished", new Object[]{
  19.     group, serviceId, instance.getIp(), instance.getPort()});
  20.             } catch (Exception var7) {
  22.                 if (this.nacosDiscoveryProperties.isFailFast()) {
  24.                     log.error("nacos registry, {} register failed...{},", new Object[]{
  25.     serviceId, registration.toString(), var7});
  26.                     ReflectionUtils.rethrowRuntimeException(var7);
  27.                 } else {
  29.                     log.warn("Failfast is false. {} register failed...{},", new Object[]{
  30.     serviceId, registration.toString(), var7});
  31.                 }
  32.             }
  34.         }
  35.     }

在获取实例的方法中，可以看到设置了很多实例信息：

  1. private Instance getNacosInstanceFromRegistration(Registration registration) {
  3.         Instance instance = new Instance();
  4.         instance.setIp(registration.getHost()); // 当前服务IP=》192.168.142.1
  5.         instance.setPort(registration.getPort()); // 启动端口=》 9005
  6.         instance.setWeight((double)this.nacosDiscoveryProperties.getWeight()); // 权重=》1.0
  7.         instance.setClusterName(this.nacosDiscoveryProperties.getClusterName()); // 集群名称=》DEFAULT
  8.         instance.setEnabled(this.nacosDiscoveryProperties.isInstanceEnabled()); // 是否主动注册的实例
  9.         instance.setMetadata(registration.getMetadata()); // 元数据
  10.         instance.setEphemeral(this.nacosDiscoveryProperties.isEphemeral()); // 临时实例还是永久实例
  11.         return instance;
  12.     }

在注册实例registerInstance方法中，会调用客户端代理进行注册：

  1. public void registerInstance(String serviceName, String groupName, Instance instance) throws NacosException 
  2.         // 检查是否合法：心跳时间、服务剔除时间、集群名称是否合法
  3.         // 心跳机制：
  4.         //  Nacos Server会开启一个定时任务用来检查注册服务实例的健康状况，
  5.         //  对于超过15s没有收到客户端心跳的实例会将它的healthy属性设置为false（客户端服务发现时不会发现）。
  6.         //  如果某个实例超过30秒没有收到心跳，直接剔除该实例（被剔除的实例如果恢复发送心跳则会重新注册）
  7.         NamingUtils.checkInstanceIsLegal(instance);
  8.         // 调用客户端代理进行注册
  9.         this.clientProxy.registerService(serviceName, groupName, instance);
  10.     }
  11.         // 获取客户端执行代理进行注册
  12.         public void registerService(String serviceName, String groupName, Instance instance) throws NacosException {
  14.             this.getExecuteClientProxy(instance).registerService(serviceName, groupName, instance);
  15.     }

获取到的通信注册客户端为NamingGrpcClientProxy：

  1. // 获取执行客户端代理，如果是临时实例，则会使用RPC，不是则使用HTTP。
  2.     // 通信层统一到 gRPC 协议，同时完善了客户端和服务端的流量控制和负载均衡能力，提升的整体吞吐。
  3.     // 由于通信使用了 RPC 方式，因此某一客户端的所有请求（无论是注册还是订阅）都通过同一个链接和同一个服务节点进行，不像之前通过 HTTP 连接可能每次请求都请求在不同的 Nacos 节点上，这就导致了服务发现的数据内容由原来的无状态化变为了与连接状态绑定的一种有状态数据。
  4.     private NamingClientProxy getExecuteClientProxy(Instance instance) {
  6.             return (NamingClientProxy)(instance.isEphemeral() ? this.grpcClientProxy : this.httpClientProxy);
  7.     }

NamingGrpcClientProxy 缓存实例信息，接着继续注册：

  1. public void registerService(String serviceName, String groupName, Instance instance) throws NacosException {
  3.         LogUtils.NAMING_LOGGER.info("[REGISTER-SERVICE] {} registering service {} with instance {}", new Object[]{
  4.     this.namespaceId, serviceName, instance});
  5.         // 缓存放入内存ConcurrentMap，DEFAULT_GROUP@@app-service001=》实例信息
  6.         this.redoService.cacheInstanceForRedo(serviceName, groupName, instance);
  7.         // 调用客户端注册
  8.         this.doRegisterService(serviceName, groupName, instance);
  9.     }

最终，发送gprc请求到服务端，客户端完成注册：

  1. public void doRegisterService(String serviceName, String groupName, Instance instance) throws NacosException {
  3.         // 创建请求对象
  4.         InstanceRequest request = new InstanceRequest(this.namespaceId, serviceName, groupName, "registerInstance", instance);
  5.         // 发送请求到服务端
  6.         this.requestToServer(request, Response.class);
  7.         this.redoService.instanceRegistered(serviceName, groupName);
  8.     }
  9.     private <T extends Response> T requestToServer(AbstractNamingRequest request, Class<T> responseClass) throws NacosException {
  11.         try {
  13.             // 处理注册中心设置权限时，需要携带认证消息头的情况
  14.             request.putAllHeader(this.getSecurityHeaders(request.getNamespace(), request.getGroupName(), request.getServiceName()));
  15.             Response response = this.requestTimeout < 0L ? this.rpcClient.request(request) : this.rpcClient.request(request, this.requestTimeout);
  16.             // 处理响应结果
  17.             if (ResponseCode.SUCCESS.getCode() != response.getResultCode()) {
  19.                 throw new NacosException(response.getErrorCode(), response.getMessage());
  20.             }
  21.             if (responseClass.isAssignableFrom(response.getClass())) {
  23.                 return response;
  24.             }
  26.             LogUtils.NAMING_LOGGER.error("Server return unexpected response '{}', expected response should be '{}'", response.getClass().getName(), responseClass.getName());
  27.         } catch (NacosException var4) {
  29.             throw var4;
  30.         } catch (Exception var5) {
  32.             throw new NacosException(500, "Request nacos server failed: ", var5);
  33.         }
  35.         throw new NacosException(500, "Server return invalid response");
  36.     }

可以在控制台看到打印的注册日志：
在这里插入图片描述

发表评论

表情：
评论列表 （有 0 条评论，238人围观）

还没有评论，来说两句吧...

相关阅读

    相关 Nacos客户端源码解读

    背景：         如今的网络服务大多都是分布式的集群部署，采用的架构也是面向服务的架构（SOA），大家都在服务治理领域奋斗，这不，今天要说的Nacos也是一个

    短命女短命女/ 0 赞/ 537 阅读