eureka客户端源码分析

约定不等于承诺〃 2023-07-12 08:58 80阅读 0赞

eureka客户端可以完成的事情

  1. eureka客户端可以获取其他客户端的信息，用于向其他客户端发送请求。
  2. eureka客户端可以把自己注册到注册中心，这样其他客户端就可以发现本客户端，直接来调用本客户端。

相关依赖

首先要是springboot应用

  1. <dependency>
  2.             <groupId>org.springframework.cloud</groupId>
  3.             <artifactId>spring-cloud-starter-eureka</artifactId>
  4.         </dependency>

我这里使用的版本是Brixton.SR5,不同版本的实现逻辑可能不一样。

逻辑详解

启动类上需要加@EnableDiscoveryClient注解，我们打开该注解开一下代码。

  1. /**
  2.  * Annotation to enable a DiscoveryClient implementation.
  3.  * @author Spencer Gibb
  4.  */
  5. @Target(ElementType.TYPE)
  6. @Retention(RetentionPolicy.RUNTIME)
  7. @Documented
  8. @Inherited
  9. @Import(EnableDiscoveryClientImportSelector.class)
  10. public @interface EnableDiscoveryClient {
  12. }

按注释的意思就可以知道，这里是去启用DiscoveryClient的实现类。这里如何去启动的？大概的思路是通过springboot的starter机制实现的，通过扫描特定的配置类EurekaClientAutoConfiguration，在该配置类中，生成了DiscoveryClient的bean。这里贴一小段代码。

  1. @Bean(destroyMethod = "shutdown")
  2.         @ConditionalOnMissingBean(value = EurekaClient.class, search = SearchStrategy.CURRENT)
  3.         public EurekaClient eurekaClient(ApplicationInfoManager manager,
  4.                 EurekaClientConfig config) {
  5.             return new CloudEurekaClient(manager, config, this.optionalArgs,
  6.                     this.context);
  7.         }

其实CloudEurekaClient里面的代码很少，基本都是父类DiscoveryClient实现的，我们直接看父类的方法。
最后是调用了如下构造方法，这里先贴一下构造方法的具体代码，不用去研究。

  1. @Inject
  2.     DiscoveryClient(ApplicationInfoManager applicationInfoManager, EurekaClientConfig config, DiscoveryClientOptionalArgs args,
  3.                     Provider<BackupRegistry> backupRegistryProvider) {
  4.         //健康校验的，跳过
  5.         if (args != null) {
  6.             this.healthCheckHandlerProvider = args.healthCheckHandlerProvider;
  7.             this.healthCheckCallbackProvider = args.healthCheckCallbackProvider;
  8.             this.eventListeners.addAll(args.getEventListeners());
  9.         } else {
  10.             this.healthCheckCallbackProvider = null;
  11.             this.healthCheckHandlerProvider = null;
  12.         }
  14.         //这主要是获取本应用信息的
  15.         this.applicationInfoManager = applicationInfoManager;
  16.         InstanceInfo myInfo = applicationInfoManager.getInfo();
  18.         clientConfig = config;
  19.         staticClientConfig = clientConfig;
  20.         transportConfig = config.getTransportConfig();
  21.         instanceInfo = myInfo;
  22.         if (myInfo != null) {
  23.             appPathIdentifier = instanceInfo.getAppName() + "/" + instanceInfo.getId();
  24.         } else {
  25.             logger.warn("Setting instanceInfo to a passed in null value");
  26.         }
  28.         this.backupRegistryProvider = backupRegistryProvider;
  30.         this.urlRandomizer = new EndpointUtils.InstanceInfoBasedUrlRandomizer(instanceInfo);
  31.         localRegionApps.set(new Applications());
  33.         fetchRegistryGeneration = new AtomicLong(0);
  35.         remoteRegionsToFetch = new AtomicReference<String>(clientConfig.fetchRegistryForRemoteRegions());
  36.         remoteRegionsRef = new AtomicReference<>(remoteRegionsToFetch.get() == null ? null : remoteRegionsToFetch.get().split(","));
  38.         //一些监视器
  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.         //如果不需要注册到eureka，也不需要从eureka读取信息的话，直接返回。
  52.         if (!config.shouldRegisterWithEureka() && !config.shouldFetchRegistry()) {
  53.             logger.info("Client configured to neither register nor query for data.");
  54.             scheduler = null;
  55.             heartbeatExecutor = null;
  56.             cacheRefreshExecutor = null;
  57.             eurekaTransport = null;
  58.             instanceRegionChecker = new InstanceRegionChecker(new PropertyBasedAzToRegionMapper(config), clientConfig.getRegion());
  60.             // This is a bit of hack to allow for existing code using DiscoveryManager.getInstance()
  61.             // to work with DI'd DiscoveryClient
  62.             DiscoveryManager.getInstance().setDiscoveryClient(this);
  63.             DiscoveryManager.getInstance().setEurekaClientConfig(config);
  65.             initTimestampMs = System.currentTimeMillis();
  67.             logger.info("Discovery Client initialized at timestamp {} with initial instances count: {}",
  68.                     initTimestampMs, this.getApplications().size());
  69.             return;  // no need to setup up an network tasks and we are done
  70.         }
  72.         //这里初始化了一些线程池，主要是给下面的task用的，
  73.         //heartbeatExecutor用于心跳,cacheRefreshExecutor用于重新获取信息
  74.         try {
  75.             scheduler = Executors.newScheduledThreadPool(3,
  76.                     new ThreadFactoryBuilder()
  77.                             .setNameFormat("DiscoveryClient-%d")
  78.                             .setDaemon(true)
  79.                             .build());
  81.             heartbeatExecutor = new ThreadPoolExecutor(
  82.                     1, clientConfig.getHeartbeatExecutorThreadPoolSize(), 0, TimeUnit.SECONDS,
  83.                     new SynchronousQueue<Runnable>(),
  84.                     new ThreadFactoryBuilder()
  85.                             .setNameFormat("DiscoveryClient-HeartbeatExecutor-%d")
  86.                             .setDaemon(true)
  87.                             .build()
  88.             );  // use direct handoff
  90.             cacheRefreshExecutor = new ThreadPoolExecutor(
  91.                     1, clientConfig.getCacheRefreshExecutorThreadPoolSize(), 0, TimeUnit.SECONDS,
  92.                     new SynchronousQueue<Runnable>(),
  93.                     new ThreadFactoryBuilder()
  94.                             .setNameFormat("DiscoveryClient-CacheRefreshExecutor-%d")
  95.                             .setDaemon(true)
  96.                             .build()
  97.             );  // use direct handoff
  99.             eurekaTransport = new EurekaTransport();
  100.             scheduleServerEndpointTask(eurekaTransport, args);
  102.             AzToRegionMapper azToRegionMapper;
  103.             if (clientConfig.shouldUseDnsForFetchingServiceUrls()) {
  104.                 azToRegionMapper = new DNSBasedAzToRegionMapper(clientConfig);
  105.             } else {
  106.                 azToRegionMapper = new PropertyBasedAzToRegionMapper(clientConfig);
  107.             }
  108.             if (null != remoteRegionsToFetch.get()) {
  109.                 azToRegionMapper.setRegionsToFetch(remoteRegionsToFetch.get().split(","));
  110.             }
  111.             instanceRegionChecker = new InstanceRegionChecker(azToRegionMapper, clientConfig.getRegion());
  112.         } catch (Throwable e) {
  113.             throw new RuntimeException("Failed to initialize DiscoveryClient!", e);
  114.         }
  116.         if (clientConfig.shouldFetchRegistry() && !fetchRegistry(false)) {
  117.             fetchRegistryFromBackup();
  118.         }
  120.         //重点，刷新缓存和心跳是在这个方法里面干的
  121.         initScheduledTasks();
  122.         try {
  123.             Monitors.registerObject(this);
  124.         } catch (Throwable e) {
  125.             logger.warn("Cannot register timers", e);
  126.         }
  128.         // This is a bit of hack to allow for existing code using DiscoveryManager.getInstance()
  129.         // to work with DI'd DiscoveryClient
  130.         DiscoveryManager.getInstance().setDiscoveryClient(this);
  131.         DiscoveryManager.getInstance().setEurekaClientConfig(config);
  133.         initTimestampMs = System.currentTimeMillis();
  134.         logger.info("Discovery Client initialized at timestamp {} with initial instances count: {}",
  135.                 initTimestampMs, this.getApplications().size());
  136.     }

主要逻辑在initScheduledTasks()里面，我们直接看这个里面的实现。

  1. /**
  2.      * Initializes all scheduled tasks.
  3.      */
  4.     private void initScheduledTasks() {
  5.         //如果需要获取注册中心的信息，则执行里面的定时任务。
  6.         if (clientConfig.shouldFetchRegistry()) {
  7.             // registry cache refresh timer
  8.             int registryFetchIntervalSeconds = clientConfig.getRegistryFetchIntervalSeconds();
  9.             int expBackOffBound = clientConfig.getCacheRefreshExecutorExponentialBackOffBound();
  10.             //定时去获取注册中心的信息，这里主要的逻辑是通过CacheRefreshThread去实现的
  11.             scheduler.schedule(
  12.                     new TimedSupervisorTask(
  13.                             "cacheRefresh",
  14.                             scheduler,
  15.                             cacheRefreshExecutor,
  16.                             registryFetchIntervalSeconds,
  17.                             TimeUnit.SECONDS,
  18.                             expBackOffBound,
  19.                             new CacheRefreshThread()
  20.                     ),
  21.                     registryFetchIntervalSeconds, TimeUnit.SECONDS);
  22.         }
  24.         //如果需要把自身注册到注册中心的话，执行里面的逻辑
  25.         if (clientConfig.shouldRegisterWithEureka()) {
  26.             int renewalIntervalInSecs = instanceInfo.getLeaseInfo().getRenewalIntervalInSecs();
  27.             int expBackOffBound = clientConfig.getHeartbeatExecutorExponentialBackOffBound();
  28.             logger.info("Starting heartbeat executor: " + "renew interval is: " + renewalIntervalInSecs);
  30.             //每隔一段时间，发送心跳信息到注册中心，具体的逻辑由HeartbeatThread线程实现。
  31.             // Heartbeat timer
  32.             scheduler.schedule(
  33.                     new TimedSupervisorTask(
  34.                             "heartbeat",
  35.                             scheduler,
  36.                             heartbeatExecutor,
  37.                             renewalIntervalInSecs,
  38.                             TimeUnit.SECONDS,
  39.                             expBackOffBound,
  40.                             new HeartbeatThread()
  41.                     ),
  42.                     renewalIntervalInSecs, TimeUnit.SECONDS);
  44.             // InstanceInfo replicator
  45.             instanceInfoReplicator = new InstanceInfoReplicator(
  46.                     this,
  47.                     instanceInfo,
  48.                     clientConfig.getInstanceInfoReplicationIntervalSeconds(),
  49.                     2); // burstSize
  51.             statusChangeListener = new ApplicationInfoManager.StatusChangeListener() {
  52.                 @Override
  53.                 public String getId() {
  54.                     return "statusChangeListener";
  55.                 }
  57.                 @Override
  58.                 public void notify(StatusChangeEvent statusChangeEvent) {
  59.                     if (InstanceStatus.DOWN == statusChangeEvent.getStatus() ||
  60.                             InstanceStatus.DOWN == statusChangeEvent.getPreviousStatus()) {
  61.                         // log at warn level if DOWN was involved
  62.                         logger.warn("Saw local status change event {}", statusChangeEvent);
  63.                     } else {
  64.                         logger.info("Saw local status change event {}", statusChangeEvent);
  65.                     }
  66.                     instanceInfoReplicator.onDemandUpdate();
  67.                 }
  68.             };
  70.             if (clientConfig.shouldOnDemandUpdateStatusChange()) {
  71.                 applicationInfoManager.registerStatusChangeListener(statusChangeListener);
  72.             }
  74.             //用于更新本地instanceinfo并将其复制到远程服务器的任务
  75.             instanceInfoReplicator.start(clientConfig.getInitialInstanceInfoReplicationIntervalSeconds());
  76.         } else {
  77.             logger.info("Not registering with Eureka server per configuration");
  78.         }
  79.     }

这里一共有三个逻辑

  1. 获取注册中心信息
  2. 心跳
  3. 更新本地信息并同步至远程

更新本地信息并同步至远程

这里看下InstanceInfoReplicator的run方法

  1. public void run() {
  2.         try {
  3.             //刷新本地配置信息
  4.             discoveryClient.refreshInstanceInfo();
  6.             Long dirtyTimestamp = instanceInfo.isDirtyWithTime();
  7.             //如果本地信息已经更新了且没有同步到远程，才有dirtyTimestamp时间
  8.             if (dirtyTimestamp != null) {
  9.                 //向远程注册
  10.                 discoveryClient.register();
  12.                 //已经同步到远程过了，这里需要更新状态，表明下次不需要重新注册
  13.                 instanceInfo.unsetIsDirty(dirtyTimestamp);
  14.             }
  15.         } catch (Throwable t) {
  16.             logger.warn("There was a problem with the instance info replicator", t);
  17.         } finally {
  18.             //这里会继续轮询，发现本地信息更新了，就会同步到远程
  19.             Future next = scheduler.schedule(this, replicationIntervalSeconds, TimeUnit.SECONDS);
  20.             scheduledPeriodicRef.set(next);
  21.         }
  22.     }

我们再看下注册方法，注册方法也很简单，只是调用一下http请求，把自身信息告诉远程。

  1. boolean register() throws Throwable {
  2.         logger.info(PREFIX + appPathIdentifier + ": registering service...");
  3.         EurekaHttpResponse<Void> httpResponse;
  4.         try {
  5.             httpResponse = eurekaTransport.registrationClient.register(instanceInfo);
  6.         } catch (Exception e) {
  7.             logger.warn("{} - registration failed {}", PREFIX + appPathIdentifier, e.getMessage(), e);
  8.             throw e;
  9.         }
  10.         if (logger.isInfoEnabled()) {
  11.             logger.info("{} - registration status: {}", PREFIX + appPathIdentifier, httpResponse.getStatusCode());
  12.         }
  13.         return httpResponse.getStatusCode() == 204;
  14.     }

获取注册中心的信息

这里主要看一下CacheRefreshThread的run方法，最后调用了fetchRegistry方法

  1. private boolean fetchRegistry(boolean forceFullRegistryFetch) {
  2.         Stopwatch tracer = FETCH_REGISTRY_TIMER.start();
  4.         try {
  5.             //获取原来注册中心获取的信息
  6.             Applications applications = getApplications();
  8.             //新增的情况
  9.             if (clientConfig.shouldDisableDelta()
  10.                     || (!Strings.isNullOrEmpty(clientConfig.getRegistryRefreshSingleVipAddress()))
  11.                     || forceFullRegistryFetch
  12.                     || (applications == null)
  13.                     || (applications.getRegisteredApplications().size() == 0)
  14.                     || (applications.getVersion() == -1)) //Client application does not have latest library supporting delta
  15.             {
  16.                 logger.info("Disable delta property : {}", clientConfig.shouldDisableDelta());
  17.                 logger.info("Single vip registry refresh property : {}", clientConfig.getRegistryRefreshSingleVipAddress());
  18.                 logger.info("Force full registry fetch : {}", forceFullRegistryFetch);
  19.                 logger.info("Application is null : {}", (applications == null));
  20.                 logger.info("Registered Applications size is zero : {}",
  21.                         (applications.getRegisteredApplications().size() == 0));
  22.                 logger.info("Application version is -1: {}", (applications.getVersion() == -1));
  23.                 getAndStoreFullRegistry();
  24.             } else {
  25.                 //更新的情况
  26.                 getAndUpdateDelta(applications);
  27.             }
  28.             applications.setAppsHashCode(applications.getReconcileHashCode());
  29.             logTotalInstances();
  30.         } catch (Throwable e) {
  31.             logger.error(PREFIX + appPathIdentifier + " - was unable to refresh its cache! status = " + e.getMessage(), e);
  32.             return false;
  33.         } finally {
  34.             if (tracer != null) {
  35.                 tracer.stop();
  36.             }
  37.         }
  39.         // Notify about cache refresh before updating the instance remote status
  40.         onCacheRefreshed();
  42.         // Update remote status based on refreshed data held in the cache
  43.         updateInstanceRemoteStatus();
  45.         // registry was fetched successfully, so return true
  46.         return true;
  47.     }

这里直接看一个更新的情况getAndUpdateDelta

  1. private void getAndUpdateDelta(Applications applications) throws Throwable {
  2.         long currentUpdateGeneration = fetchRegistryGeneration.get();
  4.         Applications delta = null;
  5.         //调用远程获取信息
  6.         EurekaHttpResponse<Applications> httpResponse = eurekaTransport.queryClient.getDelta(remoteRegionsRef.get());
  7.         if (httpResponse.getStatusCode() == Status.OK.getStatusCode()) {
  8.             delta = httpResponse.getEntity();
  9.         }
  11.         if (delta == null) {
  12.             logger.warn("The server does not allow the delta revision to be applied because it is not safe. "
  13.                     + "Hence got the full registry.");
  14.             getAndStoreFullRegistry();
  15.         } else if (fetchRegistryGeneration.compareAndSet(currentUpdateGeneration, currentUpdateGeneration + 1)) {
  16.             logger.debug("Got delta update with apps hashcode {}", delta.getAppsHashCode());
  17.             String reconcileHashCode = "";
  18.             if (fetchRegistryUpdateLock.tryLock()) {
  19.                 try {
  20.                     //调用该方法更新了注册信息。
  21.                     updateDelta(delta);
  22.                     reconcileHashCode = getReconcileHashCode(applications);
  23.                 } finally {
  24.                     fetchRegistryUpdateLock.unlock();
  25.                 }
  26.             } else {
  27.                 logger.warn("Cannot acquire update lock, aborting getAndUpdateDelta");
  28.             }
  29.             // There is a diff in number of instances for some reason
  30.             if (!reconcileHashCode.equals(delta.getAppsHashCode()) || clientConfig.shouldLogDeltaDiff()) {
  31.                 reconcileAndLogDifference(delta, reconcileHashCode);  // this makes a remoteCall
  32.             }
  33.         } else {
  34.             logger.warn("Not updating application delta as another thread is updating it already");
  35.             logger.debug("Ignoring delta update with apps hashcode {}, as another thread is updating it already", delta.getAppsHashCode());
  36.         }
  37.     }

心跳

下面再看下心跳，主要看HeartbeatThread的run方法,最后调用了renew方法。

  1. boolean renew() {
  2.         EurekaHttpResponse<InstanceInfo> httpResponse;
  3.         try {
  4.             //发送http请求，instanceInfo为客户端信息。
  5.             httpResponse = eurekaTransport.registrationClient.sendHeartBeat(instanceInfo.getAppName(), instanceInfo.getId(), instanceInfo, null);
  6.             logger.debug("{} - Heartbeat status: {}", PREFIX + appPathIdentifier, httpResponse.getStatusCode());
  7.             if (httpResponse.getStatusCode() == 404) {
  8.                 REREGISTER_COUNTER.increment();
  9.                 logger.info("{} - Re-registering apps/{}", PREFIX + appPathIdentifier, instanceInfo.getAppName());
  10.                 return register();
  11.             }
  12.             return httpResponse.getStatusCode() == 200;
  13.         } catch (Throwable e) {
  14.             logger.error("{} - was unable to send heartbeat!", PREFIX + appPathIdentifier, e);
  15.             return false;
  16.         }
  17.     }

注册信息

这里贴一个Applications类的代码，该类用于保存注册中心返回的信息，具体信息内容可以自己打断点查看。
里面保存了很多数据结构，主要是为了后期数据更方便的使用。

  1. /**
  2.  * The class that wraps all the registry information returned by eureka server.
  3.  *
  4.  */
  5. @Serializer("com.netflix.discovery.converters.EntityBodyConverter")
  6. @XStreamAlias("applications")
  7. @JsonRootName("applications")
  8. public class Applications {
  9.     private static final String APP_INSTANCEID_DELIMITER = "$$";
  10.     private static final Logger logger = LoggerFactory.getLogger(Applications.class);
  11.     private static final String STATUS_DELIMITER = "_";
  13.     private Long versionDelta = Long.valueOf(-1);
  15.     @XStreamImplicit
  16.     private AbstractQueue<Application> applications;
  18.     private Map<String, Application> appNameApplicationMap = new ConcurrentHashMap<String, Application>();
  19.     private Map<String, AbstractQueue<InstanceInfo>> virtualHostNameAppMap = new ConcurrentHashMap<String, AbstractQueue<InstanceInfo>>();
  20.     private Map<String, AbstractQueue<InstanceInfo>> secureVirtualHostNameAppMap = new ConcurrentHashMap<String, AbstractQueue<InstanceInfo>>();
  21.     private Map<String, AtomicLong> virtualHostNameIndexMap = new ConcurrentHashMap<String, AtomicLong>();
  22.     private Map<String, AtomicLong> secureVirtualHostNameIndexMap = new ConcurrentHashMap<String, AtomicLong>();
  24.     private Map<String, AtomicReference<List<InstanceInfo>>> shuffleVirtualHostNameMap = new ConcurrentHashMap<String, AtomicReference<List<InstanceInfo>>>();
  25.     private Map<String, AtomicReference<List<InstanceInfo>>> shuffledSecureVirtualHostNameMap = new ConcurrentHashMap<String, AtomicReference<List<InstanceInfo>>>();
  27.     //其他代码省略。
  28. }

发表评论

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

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

相关阅读