Spring Cloud Eureka源码解析之Eureka Client启动流程（二）

爱被打了一巴掌 2022-09-16 04:52 330阅读 0赞

看下EurekaClient的自动装配类EurekaClientAutoConfiguration
EurekaClientConfigBean包含了eureka.client的配置

  1. @Bean
  2. @ConditionalOnMissingBean(value = EurekaClientConfig.class, search = SearchStrategy.CURRENT)
  3. public EurekaClientConfigBean eurekaClientConfigBean(ConfigurableEnvironment env) { 
  4.     EurekaClientConfigBean client = new EurekaClientConfigBean();
  5.     if ("bootstrap".equals(this.env.getProperty("spring.config.name"))) { 
  6.         // We don't register during bootstrap by default, but there will be another
  7.         // chance later.
  8.         client.setRegisterWithEureka(false);
  9.     }
  10.     return client;
  11. }

EurekaInstanceConfigBean包含了eureka.instance的配置

  1. @Bean
  2. @ConditionalOnMissingBean(value = EurekaInstanceConfig.class, search = SearchStrategy.CURRENT)
  3. public EurekaInstanceConfigBean eurekaInstanceConfigBean(InetUtils inetUtils,
  4.                                                          ManagementMetadataProvider managementMetadataProvider) { 
  5.     String hostname = getProperty("eureka.instance.hostname");
  6.     boolean preferIpAddress = Boolean.parseBoolean(getProperty("eureka.instance.prefer-ip-address"));
  7.     String ipAddress = getProperty("eureka.instance.ip-address");
  8.     boolean isSecurePortEnabled = Boolean.parseBoolean(getProperty("eureka.instance.secure-port-enabled"));
  10.     String serverContextPath = env.getProperty("server.context-path", "/");
  11.     int serverPort = Integer.valueOf(env.getProperty("server.port", env.getProperty("port", "8080")));
  13.     Integer managementPort = env.getProperty("management.server.port", Integer.class);// nullable. should be wrapped into optional
  14.     String managementContextPath = env.getProperty("management.server.context-path");// nullable. should be wrapped into optional
  15.     Integer jmxPort = env.getProperty("com.sun.management.jmxremote.port", Integer.class);//nullable
  16.     EurekaInstanceConfigBean instance = new EurekaInstanceConfigBean(inetUtils);
  18.     instance.setNonSecurePort(serverPort);
  19.     instance.setInstanceId(getDefaultInstanceId(env));
  20.     instance.setPreferIpAddress(preferIpAddress);
  21.     instance.setSecurePortEnabled(isSecurePortEnabled);
  22.     if (StringUtils.hasText(ipAddress)) { 
  23.         instance.setIpAddress(ipAddress);
  24.     }
  26.     if(isSecurePortEnabled) { 
  27.         instance.setSecurePort(serverPort);
  28.     }
  30.     if (StringUtils.hasText(hostname)) { 
  31.         instance.setHostname(hostname);
  32.     }
  33.     String statusPageUrlPath = getProperty("eureka.instance.status-page-url-path");
  34.     String healthCheckUrlPath = getProperty("eureka.instance.health-check-url-path");
  36.     if (StringUtils.hasText(statusPageUrlPath)) { 
  37.         instance.setStatusPageUrlPath(statusPageUrlPath);
  38.     }
  39.     if (StringUtils.hasText(healthCheckUrlPath)) { 
  40.         instance.setHealthCheckUrlPath(healthCheckUrlPath);
  41.     }
  43.     ManagementMetadata metadata = managementMetadataProvider.get(instance, serverPort,
  44.                                                                  serverContextPath, managementContextPath, managementPort);
  46.     if(metadata != null) { 
  47.         instance.setStatusPageUrl(metadata.getStatusPageUrl());
  48.         instance.setHealthCheckUrl(metadata.getHealthCheckUrl());
  49.         if(instance.isSecurePortEnabled()) { 
  50.             instance.setSecureHealthCheckUrl(metadata.getSecureHealthCheckUrl());
  51.         }
  52.         Map<String, String> metadataMap = instance.getMetadataMap();
  53.         if (metadataMap.get("management.port") == null) { 
  54.             metadataMap.put("management.port", String.valueOf(metadata.getManagementPort()));
  55.         }
  56.     }
  58.     setupJmxPort(instance, jmxPort);
  59.     return instance;
  60. }

RefreshableEurekaClientConfiguration.eurekaClient
在这里插入图片描述
DiscoveryClient是eureka的核心类，而CloudEurekaClient就是DiscoveryClient的子类
在这里插入图片描述
在这里插入图片描述
进入到DiscoveryClient的构造方法，这里面会初始化一些任务

  1. @Inject
  2. DiscoveryClient(ApplicationInfoManager applicationInfoManager, EurekaClientConfig config, AbstractDiscoveryClientOptionalArgs args,
  3.                 Provider<BackupRegistry> backupRegistryProvider) { 
  4.     if (args != null) { 
  5.         this.healthCheckHandlerProvider = args.healthCheckHandlerProvider;
  6.         this.healthCheckCallbackProvider = args.healthCheckCallbackProvider;
  7.         this.eventListeners.addAll(args.getEventListeners());
  8.         this.preRegistrationHandler = args.preRegistrationHandler;
  9.     } else { 
  10.         this.healthCheckCallbackProvider = null;
  11.         this.healthCheckHandlerProvider = null;
  12.         this.preRegistrationHandler = null;
  13.     }
  15.     this.applicationInfoManager = applicationInfoManager;
  16.     // 该eureka实例
  17.     InstanceInfo myInfo = applicationInfoManager.getInfo();
  19.     clientConfig = config;
  20.     staticClientConfig = clientConfig;
  21.     transportConfig = config.getTransportConfig();
  22.     instanceInfo = myInfo;
  23.     if (myInfo != null) { 
  24.         appPathIdentifier = instanceInfo.getAppName() + "/" + instanceInfo.getId();
  25.     } else { 
  26.         logger.warn("Setting instanceInfo to a passed in null value");
  27.     }
  29.     this.backupRegistryProvider = backupRegistryProvider;
  31.     this.urlRandomizer = new EndpointUtils.InstanceInfoBasedUrlRandomizer(instanceInfo);
  32.     localRegionApps.set(new Applications());
  34.     fetchRegistryGeneration = new AtomicLong(0);
  36.     remoteRegionsToFetch = new AtomicReference<String>(clientConfig.fetchRegistryForRemoteRegions());
  37.     remoteRegionsRef = new AtomicReference<>(remoteRegionsToFetch.get() == null ? null : remoteRegionsToFetch.get().split(","));
  39.     if (config.shouldFetchRegistry()) { 
  40.         this.registryStalenessMonitor = new ThresholdLevelsMetric(this, METRIC_REGISTRY_PREFIX + "lastUpdateSec_", new long[]{ 15L, 30L, 60L, 120L, 240L, 480L});
  41.     } else { 
  42.         this.registryStalenessMonitor = ThresholdLevelsMetric.NO_OP_METRIC;
  43.     }
  45.     if (config.shouldRegisterWithEureka()) { 
  46.         this.heartbeatStalenessMonitor = new ThresholdLevelsMetric(this, METRIC_REGISTRATION_PREFIX + "lastHeartbeatSec_", new long[]{ 15L, 30L, 60L, 120L, 240L, 480L});
  47.     } else { 
  48.         this.heartbeatStalenessMonitor = ThresholdLevelsMetric.NO_OP_METRIC;
  49.     }
  51.     logger.info("Initializing Eureka in region {}", clientConfig.getRegion());
  53.     // 不注册到eureka server && 不从eureka server获取注册信息服务列表，则就返回了
  54.     if (!config.shouldRegisterWithEureka() && !config.shouldFetchRegistry()) { 
  55.         logger.info("Client configured to neither register nor query for data.");
  56.         scheduler = null;
  57.         heartbeatExecutor = null;
  58.         cacheRefreshExecutor = null;
  59.         eurekaTransport = null;
  60.         instanceRegionChecker = new InstanceRegionChecker(new PropertyBasedAzToRegionMapper(config), clientConfig.getRegion());
  62.         // This is a bit of hack to allow for existing code using DiscoveryManager.getInstance()
  63.         // to work with DI'd DiscoveryClient
  64.         DiscoveryManager.getInstance().setDiscoveryClient(this);
  65.         DiscoveryManager.getInstance().setEurekaClientConfig(config);
  67.         initTimestampMs = System.currentTimeMillis();
  68.         logger.info("Discovery Client initialized at timestamp {} with initial instances count: {}",
  69.                     initTimestampMs, this.getApplications().size());
  71.         return;  // no need to setup up an network tasks and we are done
  72.     }
  74.     try { 
  75.         // default size of 2 - 1 each for heartbeat and cacheRefresh
  76.         scheduler = Executors.newScheduledThreadPool(2,
  77.                                                      new ThreadFactoryBuilder()
  78.                                                      .setNameFormat("DiscoveryClient-%d")
  79.                                                      .setDaemon(true)
  80.                                                      .build());
  82.         // 发送心跳线程池
  83.         heartbeatExecutor = new ThreadPoolExecutor(
  84.             1, clientConfig.getHeartbeatExecutorThreadPoolSize(), 0, TimeUnit.SECONDS,
  85.             new SynchronousQueue<Runnable>(),
  86.             new ThreadFactoryBuilder()
  87.             .setNameFormat("DiscoveryClient-HeartbeatExecutor-%d")
  88.             .setDaemon(true)
  89.             .build()
  90.         );  // use direct handoff
  92.         // 刷新注册信息缓存线程池
  93.         cacheRefreshExecutor = new ThreadPoolExecutor(
  94.             1, clientConfig.getCacheRefreshExecutorThreadPoolSize(), 0, TimeUnit.SECONDS,
  95.             new SynchronousQueue<Runnable>(),
  96.             new ThreadFactoryBuilder()
  97.             .setNameFormat("DiscoveryClient-CacheRefreshExecutor-%d")
  98.             .setDaemon(true)
  99.             .build()
  100.         );  // use direct handoff
  102.         eurekaTransport = new EurekaTransport();
  103.         scheduleServerEndpointTask(eurekaTransport, args);
  105.         AzToRegionMapper azToRegionMapper;
  106.         if (clientConfig.shouldUseDnsForFetchingServiceUrls()) { 
  107.             azToRegionMapper = new DNSBasedAzToRegionMapper(clientConfig);
  108.         } else { 
  109.             azToRegionMapper = new PropertyBasedAzToRegionMapper(clientConfig);
  110.         }
  111.         if (null != remoteRegionsToFetch.get()) { 
  112.             azToRegionMapper.setRegionsToFetch(remoteRegionsToFetch.get().split(","));
  113.         }
  114.         instanceRegionChecker = new InstanceRegionChecker(azToRegionMapper, clientConfig.getRegion());
  115.     } catch (Throwable e) { 
  116.         throw new RuntimeException("Failed to initialize DiscoveryClient!", e);
  117.     }
  119.     // 第一次拉取，如果拉取失败就从备份注册中心拉取
  120.     if (clientConfig.shouldFetchRegistry() && !fetchRegistry(false)) { 
  121.         fetchRegistryFromBackup();
  122.     }
  124.     // call and execute the pre registration handler before all background tasks (inc registration) is started
  125.     if (this.preRegistrationHandler != null) { 
  126.         this.preRegistrationHandler.beforeRegistration();
  127.     }
  129.     // 需要注册到eureka server && 在初始化时强制注册（默认false，所以这里的register方法默认不执行）
  130.     if (clientConfig.shouldRegisterWithEureka() && clientConfig.shouldEnforceRegistrationAtInit()) { 
  131.         try { 
  132.             if (!register() ) { 
  133.                 throw new IllegalStateException("Registration error at startup. Invalid server response.");
  134.             }
  135.         } catch (Throwable th) { 
  136.             logger.error("Registration error at startup: {}", th.getMessage());
  137.             throw new IllegalStateException(th);
  138.         }
  139.     }
  141.     // 初始化定时任务，服务心跳、服务列表获取等功能在此处完成
  142.     initScheduledTasks();
  144.     try { 
  145.         Monitors.registerObject(this);
  146.     } catch (Throwable e) { 
  147.         logger.warn("Cannot register timers", e);
  148.     }
  150.     // This is a bit of hack to allow for existing code using DiscoveryManager.getInstance()
  151.     // to work with DI'd DiscoveryClient
  152.     DiscoveryManager.getInstance().setDiscoveryClient(this);
  153.     DiscoveryManager.getInstance().setEurekaClientConfig(config);
  155.     initTimestampMs = System.currentTimeMillis();
  156.     logger.info("Discovery Client initialized at timestamp {} with initial instances count: {}",
  157.                 initTimestampMs, this.getApplications().size());
  158. }
  160. // 初始化eureka client所有的定时任务
  161. private void initScheduledTasks() { 
  162.     if (clientConfig.shouldFetchRegistry()) { 
  163.         // 缓存刷新定时任务，拉取注册信息
  164.         int registryFetchIntervalSeconds = clientConfig.getRegistryFetchIntervalSeconds();
  165.         int expBackOffBound = clientConfig.getCacheRefreshExecutorExponentialBackOffBound();
  166.         scheduler.schedule(
  167.             new TimedSupervisorTask(
  168.                 "cacheRefresh",
  169.                 scheduler,
  170.                 cacheRefreshExecutor,
  171.                 registryFetchIntervalSeconds,
  172.                 TimeUnit.SECONDS,
  173.                 expBackOffBound,
  174.                 new CacheRefreshThread()
  175.             ),
  176.             registryFetchIntervalSeconds, TimeUnit.SECONDS);
  177.     }
  179.     if (clientConfig.shouldRegisterWithEureka()) { 
  180.         int renewalIntervalInSecs = instanceInfo.getLeaseInfo().getRenewalIntervalInSecs();
  181.         int expBackOffBound = clientConfig.getHeartbeatExecutorExponentialBackOffBound();
  182.         logger.info("Starting heartbeat executor: " + "renew interval is: {}", renewalIntervalInSecs);
  184.         // 发送心跳定时器
  185.         scheduler.schedule(
  186.             new TimedSupervisorTask(
  187.                 "heartbeat",
  188.                 scheduler,
  189.                 heartbeatExecutor,
  190.                 renewalIntervalInSecs,
  191.                 TimeUnit.SECONDS,
  192.                 expBackOffBound,
  193.                 new HeartbeatThread()
  194.             ),
  195.             renewalIntervalInSecs, TimeUnit.SECONDS);
  197.         // 实例信息复制器，不知道为啥叫复制器，里面其实是注册的逻辑
  198.         instanceInfoReplicator = new InstanceInfoReplicator(
  199.             this,
  200.             instanceInfo,
  201.             clientConfig.getInstanceInfoReplicationIntervalSeconds(),
  202.             2); // burstSize
  204.         // 实例状态变更的监听器
  205.         statusChangeListener = new ApplicationInfoManager.StatusChangeListener() { 
  206.             @Override
  207.             public String getId() { 
  208.                 return "statusChangeListener";
  209.             }
  211.             @Override
  212.             public void notify(StatusChangeEvent statusChangeEvent) { 
  213.                 if (InstanceStatus.DOWN == statusChangeEvent.getStatus() ||
  214.                     InstanceStatus.DOWN == statusChangeEvent.getPreviousStatus()) { 
  215.                     // log at warn level if DOWN was involved
  216.                     logger.warn("Saw local status change event {}", statusChangeEvent);
  217.                 } else { 
  218.                     logger.info("Saw local status change event {}", statusChangeEvent);
  219.                 }
  220.                 // 有可能会重新注册
  221.                 instanceInfoReplicator.onDemandUpdate();
  222.             }
  223.         };
  225.         if (clientConfig.shouldOnDemandUpdateStatusChange()) { 
  226.             applicationInfoManager.registerStatusChangeListener(statusChangeListener);
  227.         }
  229.         // 启动复制器，包含注册逻辑
  230.         instanceInfoReplicator.start(clientConfig.getInitialInstanceInfoReplicationIntervalSeconds());
  231.     } else { 
  232.         logger.info("Not registering with Eureka server per configuration");
  233.     }
  234. }

发表评论

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

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

相关阅读