前面分析了Eureka Server的功能，下面看下Eureka Client的功能实现

  • 注册到Eureka Server
  • 从Eureka Server中获取服务列表

1.注册到Eureka Server

我们在使用client的时候，需要使用@EnableDiscoveryClient注解表明自己是一个client，我们通过追踪这个注解类的时候发现和DiscoveryClient绑定了。

  1. Annotation to enable a DiscoveryClient implementation.

查看DiscoveryClient类有一个register方法

  1. /**
  2. * Register with the eureka service by making the appropriate REST call.
  3. * 通过REST调用向eureka服务注册
  4. */
  5. boolean register() throws Throwable {
  6.     logger.info(PREFIX + "{}: registering service...", appPathIdentifier);
  7.     EurekaHttpResponse<Void> httpResponse;
  8.     try {
  9.         httpResponse = eurekaTransport.registrationClient.register(instanceInfo);
  10.     } catch (Exception e) {
  11.         logger.warn(PREFIX + "{} - registration failed {}", appPathIdentifier, e.getMessage(), e);
  12.         throw e;
  13.     }
  14.     if (logger.isInfoEnabled()) {
  15.         logger.info(PREFIX + "{} - registration status: {}", appPathIdentifier, httpResponse.getStatusCode());
  16.     }
  17.     return httpResponse.getStatusCode() == 204;
  18. }

这个 registrationClient 有很多实现，默认的实现是 SessionedEurekaHttpClient。当返回 204 的时候，表示注册成功。

DiscoveryClient里面维护了一个两个定时任务，

  • 定时向注册中心拉取服务（包括全量拉取、增量拉取）（默认30秒）

  • 定时发送心跳连接（默认30秒）

    1. private void initScheduledTasks() {
    2.       if (clientConfig.shouldFetchRegistry()) {
    3.           // registry cache refresh timer
    4.           int registryFetchIntervalSeconds = clientConfig.getRegistryFetchIntervalSeconds();
    5.           int expBackOffBound = clientConfig.getCacheRefreshExecutorExponentialBackOffBound();
    6.           scheduler.schedule(
    7.                   new TimedSupervisorTask(
    8.                           "cacheRefresh",
    9.                           scheduler,
    10.                           cacheRefreshExecutor,
    11.                           registryFetchIntervalSeconds,
    12.                           TimeUnit.SECONDS,
    13.                           expBackOffBound,
    14.                           // 向注册中心拉取服务
    15.                           new CacheRefreshThread()
    16.                   ),
    17.                   registryFetchIntervalSeconds, TimeUnit.SECONDS);
    18.       }
    20.       if (clientConfig.shouldRegisterWithEureka()) {
    21.           int renewalIntervalInSecs = instanceInfo.getLeaseInfo().getRenewalIntervalInSecs();
    22.           int expBackOffBound = clientConfig.getHeartbeatExecutorExponentialBackOffBound();
    23.           logger.info("Starting heartbeat executor: " + "renew interval is: {}", renewalIntervalInSecs);
    25.           // Heartbeat timer
    26.           scheduler.schedule(
    27.                   new TimedSupervisorTask(
    28.                           "heartbeat",
    29.                           scheduler,
    30.                           heartbeatExecutor,
    31.                           renewalIntervalInSecs,
    32.                           TimeUnit.SECONDS,
    33.                           expBackOffBound,
    34.                           // 心跳连接
    35.                           new HeartbeatThread()
    36.                   ),
    37.                   renewalIntervalInSecs, TimeUnit.SECONDS);
    39.           // InstanceInfo replicator
    40.           instanceInfoReplicator = new InstanceInfoReplicator(
    41.                   this,
    42.                   instanceInfo,
    43.                   clientConfig.getInstanceInfoReplicationIntervalSeconds(),
    44.                   2); // burstSize
    46.           statusChangeListener = new ApplicationInfoManager.StatusChangeListener() {
    47.               @Override
    48.               public String getId() {
    49.                   return "statusChangeListener";
    50.               }
    52.               @Override
    53.               public void notify(StatusChangeEvent statusChangeEvent) {
    54.                   if (InstanceStatus.DOWN == statusChangeEvent.getStatus() ||
    55.                           InstanceStatus.DOWN == statusChangeEvent.getPreviousStatus()) {
    56.                       // log at warn level if DOWN was involved
    57.                       logger.warn("Saw local status change event {}", statusChangeEvent);
    58.                   } else {
    59.                       logger.info("Saw local status change event {}", statusChangeEvent);
    60.                   }
    61.                   instanceInfoReplicator.onDemandUpdate();
    62.               }
    63.           };
    65.           if (clientConfig.shouldOnDemandUpdateStatusChange()) {
    66.               applicationInfoManager.registerStatusChangeListener(statusChangeListener);
    67.           }
    69.           instanceInfoReplicator.start(clientConfig.getInitialInstanceInfoReplicationIntervalSeconds());
    70.       } else {
    71.           logger.info("Not registering with Eureka server per configuration");
    72.       }
    73.   }

    2.心跳连接

    下面看下心跳连接

    1. private class HeartbeatThread implements Runnable {
    3.   public void run() {
    4.       // 通过适当的REST调用续订eureka服务
    5.       if (renew()) {
    6.           lastSuccessfulHeartbeatTimestamp = System.currentTimeMillis();
    7.       }
    8.   }
    9. }
    1. boolean renew() {
    2.   EurekaHttpResponse<InstanceInfo> httpResponse;
    3.   try {
    4.       httpResponse = eurekaTransport.registrationClient.sendHeartBeat(instanceInfo.getAppName(), instanceInfo.getId(), instanceInfo, null);
    5.       logger.debug(PREFIX + "{} - Heartbeat status: {}", appPathIdentifier, httpResponse.getStatusCode());
    6.       if (httpResponse.getStatusCode() == 404) {
    7.           REREGISTER_COUNTER.increment();
    8.           logger.info(PREFIX + "{} - Re-registering apps/{}", appPathIdentifier, instanceInfo.getAppName());
    9.           long timestamp = instanceInfo.setIsDirtyWithTime();
    10.           // 向注册中心发送注册服务
    11.           boolean success = register();
    12.           if (success) {
    13.               instanceInfo.unsetIsDirty(timestamp);
    14.           }
    15.           return success;
    16.       }
    17.       return httpResponse.getStatusCode() == 200;
    18.   } catch (Throwable e) {
    19.       logger.error(PREFIX + "{} - was unable to send heartbeat!", appPathIdentifier, e);
    20.       return false;
    21.   }
    22. }

    3.向注册中心拉取服务

    客户端向注册中心拉取服务分为全量拉取和增量拉取
    下面看下拉取服务代码

    1. // 向注册中心拉取服务
    2. boolean success = fetchRegistry(remoteRegionsModified);
    1. private boolean fetchRegistry(boolean forceFullRegistryFetch) {
    2.       Stopwatch tracer = FETCH_REGISTRY_TIMER.start();
    4.       try {
    5.           // If the delta is disabled or if it is the first time, get all
    6.           // applications
    7.           // 拿到本地Eureka注册信息缓存
    8.           Applications applications = getApplications();
    10.           if (clientConfig.shouldDisableDelta()
    11.                   || (!Strings.isNullOrEmpty(clientConfig.getRegistryRefreshSingleVipAddress()))
    12.                   || forceFullRegistryFetch
    13.                   || (applications == null)
    14.                   || (applications.getRegisteredApplications().size() == 0)
    15.                   || (applications.getVersion() == -1)) //Client application does not have latest library supporting delta
    16.           {
    17.               logger.info("Disable delta property : {}", clientConfig.shouldDisableDelta());
    18.               logger.info("Single vip registry refresh property : {}", clientConfig.getRegistryRefreshSingleVipAddress());
    19.               logger.info("Force full registry fetch : {}", forceFullRegistryFetch);
    20.               logger.info("Application is null : {}", (applications == null));
    21.               logger.info("Registered Applications size is zero : {}",
    22.                       (applications.getRegisteredApplications().size() == 0));
    23.               logger.info("Application version is -1: {}", (applications.getVersion() == -1));
    24.               // 全量拉取
    25.               getAndStoreFullRegistry();
    26.           } else {
    27.               // 增量拉取
    28.               getAndUpdateDelta(applications);
    29.           }
    30.           applications.setAppsHashCode(applications.getReconcileHashCode());
    31.           logTotalInstances();
    32.       } catch (Throwable e) {
    33.           logger.error(PREFIX + "{} - was unable to refresh its cache! status = {}", appPathIdentifier, e.getMessage(), e);
    34.           return false;
    35.       } finally {
    36.           if (tracer != null) {
    37.               tracer.stop();
    38.           }
    39.       }
    41.       // Notify about cache refresh before updating the instance remote status
    42.       onCacheRefreshed();
    44.       // Update remote status based on refreshed data held in the cache
    45.       updateInstanceRemoteStatus();
    47.       // registry was fetched successfully, so return true
    48.       return true;
    49.   }

    3.1全量拉取

    当客户端第一启动时会向注册中心全量拉取所有的服务

    1. private void getAndStoreFullRegistry() throws Throwable {
    2.       long currentUpdateGeneration = fetchRegistryGeneration.get();
    4.       logger.info("Getting all instance registry info from the eureka server");
    6.       Applications apps = null;
    7.       EurekaHttpResponse<Applications> httpResponse = clientConfig.getRegistryRefreshSingleVipAddress() == null
    8.               ? eurekaTransport.queryClient.getApplications(remoteRegionsRef.get())
    9.               : eurekaTransport.queryClient.getVip(clientConfig.getRegistryRefreshSingleVipAddress(), remoteRegionsRef.get());
    10.       if (httpResponse.getStatusCode() == Status.OK.getStatusCode()) {
    11.           apps = httpResponse.getEntity();
    12.       }
    13.       logger.info("The response status is {}", httpResponse.getStatusCode());
    15.       if (apps == null) {
    16.           logger.error("The application is null for some reason. Not storing this information");
    17.       } else if (fetchRegistryGeneration.compareAndSet(currentUpdateGeneration, currentUpdateGeneration + 1)) {
    18.           localRegionApps.set(this.filterAndShuffle(apps));
    19.           logger.debug("Got full registry with apps hashcode {}", apps.getAppsHashCode());
    20.       } else {
    21.           logger.warn("Not updating applications as another thread is updating it already");
    22.       }
    23.   }

3.2增量拉取

把Eureka Server注册中心最近（3分钟内）更新的注册信息拉取下来

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