第二章Eureka注册发现原理解析笔记

快来打我* 2023-07-13 12:10 5阅读 0赞

# **一、EurekaClient运行流程分析** #

EurekaClient为了简化开发人员的工作量，将很多与EurekaServer交互的工作隐藏起来，自主完成，具体完成的工作如下，代码见spring-cloud-netflix-eureka-client-2.0.0.RELEASE.jar。

代码Git地址：https://gitee.com/hankin\_chj/springcloud-micro-service.git

## **1、应用启动阶段：** ##

*  读取与Eureka Server交互的配置信息，封装成EurekaClientConfig；
 *  读取自身服务实例配置信息，封装成EurekalnstanceConfig；
 *  从Eureka server拉取注册表信息并缓存到本地；
 *  服务注册；
 *  通过定时任务初始化发送心跳、缓存刷新(拉取注册表信息更新本地缓存)和按需注册(监控服务实例信息变化，决定是否重新发起注册，更新注册表中的服务实例元数据)定时任务。

## **2、应用执行阶段** ##

*  定时发送心跳到Eureka Server中维持在注册表的租约；
 *  定时从Eureka Server中拉取注册表信息，更新本地注册表缓存；
 *  监控应用自身信息变化，若发生变化，需要重新发起服务注册。

## **3、应用销毁阶段** ##

从Eureka Server注销自身服务实例。

## **4、应用启动与运行阶段分析** ##

Eureka Client通过Starter的方式引人依赖，Spring Boot将会为项目使用以下的自动配置类：

*  **EurekaClientAutoConfiguration：**EurekeClient自动配置类，负责Eureka关键Beans的配置和初始化，如AppplicationInfoManager和EurekaClientConfig等。
 *  **RibbonEurekaAutoConfiguration：**Ribbon负载均衡相关配置。
 *  **EurekaDiscoveryClientConfiguration：**配置自动注册和应用的健康检查器。

**这些注解对应的配置信息：**

spring-cloud-netflix-eureka-client-2.0.0.RELEASE.jar/META-INF/spring.factories

**org.springframework.boot.autoconfigure.EnableAutoConfiguration**=**\\**  
**org.springframework.cloud.netflix.eureka.config.EurekaClientConfigServerAutoConfiguration,****\\**  
**org.springframework.cloud.netflix.eureka.config.EurekaDiscoveryClientConfigServiceAutoConfiguration,****\\**  
**org.springframework.cloud.netflix.eureka.EurekaClientAutoConfiguration,\\**  
**org.springframework.cloud.netflix.ribbon.eureka.RibbonEurekaAutoConfiguration,\\**  
**org.springframework.cloud.netflix.eureka.EurekaDiscoveryClientConfiguration**  
**org.springframework.cloud.bootstrap.BootstrapConfiguration**=**\\**  
**org.springframework.cloud.netflix.eureka.config.EurekaDiscoveryClientConfigServiceBootstrapConfiguration**

### **4.1、读取应用自身配置** ###

通过**EurekaClientAutoConfiguration**配置类，Spring boot帮助Eureka Client完成很多必要Bean的属性读取和配置。

**1）****EurekaClientConfig****：**封装Eureka Client与Eureka Server交互所需要的配置信息。Spring Cloud为其提供了一个默认配置类的EurekaClientConfigBean，可以在配置文件中通过前缀eureka.client属性名进行属性覆盖。源码见org.springframework.cloud.netflix.eureka.EurekaClientConfigBean：

@ConfigurationProperties(EurekaClientConfigBean.***PREFIX***)  
**public class** EurekaClientConfigBean **implements** EurekaClientConfig \{   
   **public static final** String ***PREFIX*** = **"eureka.client"**;

**2）****ApplicationInfoManager****：****作为应用信息管理器**，管理服务实例的信息类InstanceInfo和服务实例的配置信息类EurekaInstanceConfig。

它的构造方法如下：

@Inject  
**public** ApplicationInfoManager(EurekaInstanceConfig config, InstanceInfo instanceInfo, OptionalArgs optionalArgs) \{   
    **this**.**config** = config;  
    **this**.**instanceInfo** = instanceInfo;  
    **this**.**listeners** = **new** ConcurrentHashMap<String, StatusChangeListener>();  
    **if** (optionalArgs != **null**) \{   
        **this**.**instanceStatusMapper** = optionalArgs.getInstanceStatusMapper();  
    \} **else** \{   
        **this**.**instanceStatusMapper** = ***NO\_OP\_MAPPER***;  
    \}  
*    **instance* = **this**;  
\}

**3）****InstanceInfo****：****封装将被发送到EurekaServer进行服务注册的服务实例元数据**。它在EurekServer的注册表中代表一个服务实例，其他服务实例可以通过Instancelnfo了解该服的相关信息从而发起服务请求。

**4）****EurekaInstanceConfig****：**封装EurekaClient自身服务实例的配置信息，主要用于构建InstanceInfo通常这些信息在配置文件中的eureka.instance前缀下进行设置， SpringCloud通过EurekalnstanceConfigBean配置类提供了默认配置。

**5）****DiscoveryClient****：**Spring Cloud中定义用来服务发现的客户端接口，对于DiscoveryClient可以具体查看EurekaDiscoveryClient，EurekaDiscoveryClient又借助EurekaClient来实现。

另外netflix包里面还有一个DiscoveryClient，按名字翻译其实就是服务发现客户端，他是整个EurekaClient的核心，是与EurekaServer进行交互的核心所在。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM3NjYxNDU4_size_16_color_FFFFFF_t_70][]

@ImplementedBy(DiscoveryClient.**class**)  
**public interface** EurekaClient **extends** LookupService \{

@Singleton  
**public class** DiscoveryClient **implements** EurekaClient \{

**public class** EurekaDiscoveryClient **implements** DiscoveryClient \{   
   **public static final** String ***DESCRIPTION*** = **"Spring Cloud Eureka Discovery Client"**;  
   **private final** EurekaInstanceConfig **config**;  
   **private final** EurekaClient **eurekaClient**;  
   **public** EurekaDiscoveryClient(EurekaInstanceConfig config, EurekaClient eurekaClient) \{   
      **this**.**config** = config;  
      **this**.**eurekaClient** = eurekaClient;  
   \}

### **4.2、运行服务** ###

DiscoveryClient核心方法代码如下所示：

DiscoveryClient(ApplicationInfoManager applicationInfoManager, EurekaClientConfig config,  
                AbstractDiscoveryClientOptionalArgs args,Provider<BackupRegistry> backupRegistryProvider) \{   
    **if** (args != **null**) \{   
        **this**.healthCheckHandlerProvider = args.healthCheckHandlerProvider;  
        **this**.healthCheckCallbackProvider = args.healthCheckCallbackProvider;  
        **this**.eventListeners.addAll(args.getEventListeners());  
        **this**.preRegistrationHandler = args.preRegistrationHandler;  
    \} **else** \{   
        **this**.healthCheckCallbackProvider = **null**;  
        **this**.healthCheckHandlerProvider = **null**;  
        **this**.preRegistrationHandler = **null**;  
    \}  
    **this**.applicationInfoManager = applicationInfoManager;  
    InstanceInfo myInfo = applicationInfoManager.getInfo();  
    clientConfig = config;  
    staticClientConfig = clientConfig;  
    transportConfig = config.getTransportConfig();  
    instanceInfo = myInfo;  
    **if** (myInfo != **null**) \{   
        appPathIdentifier = instanceInfo.getAppName() + **"/"** \+ instanceInfo.getId();  
    \} **else** \{   
        logger.warn(**"Setting instanceInfo to a passed in null value"**);  
    \}  
    *//传入BackupRegistry（NotImplementedRegistryImpl）备份注册中心*  
*    ***this**.backupRegistryProvider = backupRegistryProvider;  
    **this**.urlRandomizer = **new** EndpointUtils.InstanceInfoBasedUrlRandomizer(instanceInfo);  
    localRegionApps.set(**new** Applications());  
    fetchRegistryGeneration = **new** AtomicLong(0);  
    remoteRegionsToFetch = **new** AtomicReference<String>(clientConfig.fetchRegistryForRemoteRegions());  
    remoteRegionsRef = **new** AtomicReference<>(remoteRegionsToFetch.get() == **null** ? **null** : remoteRegionsToFetch.get().split(**","**));  
   ** *****//从eureka server拉起注册表信息  eureka.client.fetch-register***  
***    *****if (config.shouldFetchRegistry()) \{**   
        **this**.registryStalenessMonitor = **new** ThresholdLevelsMetric(**this**, METRIC\_REGISTRY\_PREFIX

\+ **"lastUpdateSec\_"**, **new long**\[\]\{ 15L, 30L, 60L, 120L, 240L, 480L\});  
    \} **else** \{   
        **this**.registryStalenessMonitor = ThresholdLevelsMetric.NO\_OP\_METRIC;  
    \}  
**    *****// 当前的客户端是否应该注册到erueka中******：******eureka.client.register-with-eureka  ***  
*    ***if** (config.shouldRegisterWithEureka()) \{   
        **this**.heartbeatStalenessMonitor = **new** ThresholdLevelsMetric(**this**, METRIC\_REGISTRATION\_PREFIX

+ **"lastHeartbeatSec\_"**, **new long**\[\]\{ 15L, 30L, 60L, 120L, 240L, 480L\});  
    \} **else** \{   
        **this**.heartbeatStalenessMonitor = ThresholdLevelsMetric.NO\_OP\_METRIC;  
    \}  
    logger.info(**"Initializing Eureka in region \{\}"**, clientConfig.getRegion());  
**    *****//如果既不需要注册，也不需要拉去数据，直接返回,初始结束***  
*    ***if** (!config.shouldRegisterWithEureka() && !config.shouldFetchRegistry()) \{   
        logger.info(**"Client configured to neither register nor query for data."**);  
        scheduler = **null**;  
        heartbeatExecutor = **null**;  
        cacheRefreshExecutor = **null**;  
        eurekaTransport = **null**;  
        instanceRegionChecker = **new** InstanceRegionChecker(**new** PropertyBasedAzToRegionMapper(config), clientConfig.getRegion());  
        *// This is a bit of hack to allow for existing code using DiscoveryManager.getInstance()*  
*        // to work with DI'd DiscoveryClient*  
*        *DiscoveryManager.getInstance().setDiscoveryClient(**this**);  
        DiscoveryManager.getInstance().setEurekaClientConfig(config);  
        initTimestampMs = System.*currentTimeMillis*();  
        logger.info(**"Discovery Client initialized at timestamp \{\} with initial instances count: \{\}"**,  
                initTimestampMs, **this**.getApplications().size());  
        **return**;  *// no need to setup up an network tasks and we are done*  
*    *\}  
    **try** \{   
      **  *****//线程池大小为2，一个用户发送心跳，另外******1******个缓存刷新***  
***        // default size of 2 - 1 each for heartbeat and cacheRefresh***  
***        *****scheduler = Executors.newScheduledThreadPool(2,**  
**                new ThreadFactoryBuilder().setNameFormat("DiscoveryClient-%d").setDaemon(true).build());**  
        heartbeatExecutor = **new** ThreadPoolExecutor(  
                1, clientConfig.getHeartbeatExecutorThreadPoolSize(), 0, TimeUnit.SECONDS,  
                **new** SynchronousQueue<Runnable>(),  
                **new** ThreadFactoryBuilder().setNameFormat(**"DiscoveryClient-HeartbeatExecutor-%d"**).setDaemon(**true**) .build()  
        );  *// use direct handoff*  
*      **  *cacheRefreshExecutor = **new** ThreadPoolExecutor(  
                1, clientConfig.getCacheRefreshExecutorThreadPoolSize(), 0, TimeUnit.SECONDS,  
                **new** SynchronousQueue<Runnable>(),  
                **new** ThreadFactoryBuilder().setNameFormat(**"DiscoveryClient-CacheRefreshExecutor-%d"**).setDaemon(**true**).build()  
        );  *// use direct handoff*  
*        /****/初始化client与server交互的jersey客户端***  
***        *****eurekaTransport = new EurekaTransport();**  
        scheduleServerEndpointTask(eurekaTransport, args);  
        AzToRegionMapper azToRegionMapper;  
        **if** (clientConfig.shouldUseDnsForFetchingServiceUrls()) \{   
            azToRegionMapper = **new** DNSBasedAzToRegionMapper(clientConfig);  
        \} **else** \{   
            azToRegionMapper = **new** PropertyBasedAzToRegionMapper(clientConfig);  
        \}  
        **if** (**null** != remoteRegionsToFetch.get()) \{   
            azToRegionMapper.setRegionsToFetch(remoteRegionsToFetch.get().split(**","**));  
        \}  
        instanceRegionChecker = **new** InstanceRegionChecker(azToRegionMapper, clientConfig.getRegion());  
    \} **catch** (Throwable e) \{   
        **throw new** RuntimeException(**"Failed to initialize DiscoveryClient!"**, e);  
    \}  
  **  *****//拉取注册表的信息***  
***    *****if (clientConfig.shouldFetchRegistry() && !fetchRegistry(false)) \{**   
**        fetchRegistryFromBackup();**  
**    \}**  
    ***//将服务实例进行注册***  
*    // call and execute the pre registration handler before all background tasks (inc registration) is started*  
*   ** ***if** (**this**.preRegistrationHandler != **null**) \{   
        **this**.preRegistrationHandler.beforeRegistration();  
    \}  
    **if** (clientConfig.shouldRegisterWithEureka() && clientConfig.shouldEnforceRegistrationAtInit()) \{   
        **try** \{   
            **if** (!register() ) \{   
                **throw new** IllegalStateException(**"Registration error at startup. Invalid server response."**);  
            \}  
        \} **catch** (Throwable th) \{   
            logger.error(**"Registration error at startup: \{\}"**, th.getMessage());  
            **throw new** IllegalStateException(th);  
        \}  
    \}  
    *// finally, init the schedule tasks (e.g. cluster resolvers, heartbeat, instanceInfo replicator, fetch*  
*  ****  //初始心跳定时任务，缓存刷新***  
***    *****initScheduledTasks();**  
**    try \{**   
**        Monitors.registerObject(this);**  
    \} **catch** (Throwable e) \{   
        logger.warn(**"Cannot register timers"**, e);  
    \}  
    DiscoveryManager.getInstance().setDiscoveryClient(**this**);  
    DiscoveryManager.getInstance().setEurekaClientConfig(config);  
    initTimestampMs = System.*currentTimeMillis*();  
    logger.info(**"Discovery Client initialized at timestamp \{\} with initial instances count: \{\}"**,  
    initTimestampMs, **this**.getApplications().size());  
\}

### **4.3、总结** ###

DiscoveryClient构造函数做的事情：

*  相关配置赋值
 *  备份注册中心的初始化，实际什么事都没做
 *  拉取Server注册表中的信息
 *  注册前的预处理
 *  向Server注册自身
 *  初始心跳定时任务，缓存刷新等定时任务

## **5、Eureka客户端流程源码分析** ##

### **5.1、拉取Server注册表中的信息代码分析（重要步骤）** ###

**if** (**clientConfig**.shouldFetchRegistry() && !fetchRegistry(**false**)) \{   
    fetchRegistryFromBackup();  
\}

**判断是否全量拉去方法源码fetchRegistry(false)：**

*//是否全量拉去*  
**private boolean** fetchRegistry(**boolean** forceFullRegistryFetch) \{   
    Stopwatch tracer = FETCH\_REGISTRY\_TIMER.start();  
    **try** \{  *// If the delta is disabled or if it is the first time, get all applications*  
*        //如果增量拉取被禁止全量拉去*  
*        *Applications applications = getApplications();  
        **if** (clientConfig.shouldDisableDelta()  
                || (!Strings.isNullOrEmpty(clientConfig.getRegistryRefreshSingleVipAddress()))  
                || forceFullRegistryFetch  
                || (applications == **null**)  
                || (applications.getRegisteredApplications().size() == 0)  
                || (applications.getVersion() == -1)) * *\{   
           ** *****//全量拉取***  
***            *****getAndStoreFullRegistry();**  
        \} **else** \{   
            *//增量拉取*  
*            *getAndUpdateDelta(applications);  
        \}  
        applications.setAppsHashCode(applications.getReconcileHashCode());  
    **    *****//打印注册表上所有服务实例信息***  
***        *****logTotalInstances();**  
    \} **catch** (Throwable e) \{   
 logger.error(PREFIX + **"\{\} - was unable to refresh its cache! status = \{\}"**, appPathIdentifier, e.getMessage(), e);  
        **return false**;  
    \} **finally** \{   
        **if** (tracer != **null**) \{   
            tracer.stop();  
        \}  
    \}  
    onCacheRefreshed();  
    *// Update remote status based on refreshed data held in the cache*  
*    *updateInstanceRemoteStatus();  
    *// registry was fetched successfully, so return true*  
*    ***return true**;  
\}

### **5.2、全量拉取源码分析** ###

一般只有在第一次拉去注册表信息的时候，全量拉取调用 getAndStoreFullRegistry()方法：

**private void** getAndStoreFullRegistry() **throws** Throwable \{   
   ** *****//拉取注册表的版本信息***  
***    *****long currentUpdateGeneration = fetchRegistryGeneration.get();**  
    logger.info(**"Getting all instance registry info from the eureka server"**);  
    Applications apps = **null**;  
    EurekaHttpResponse<Applications> httpResponse = clientConfig.getRegistryRefreshSingleVipAddress() == **null**  
**            **? eurekaTransport.queryClient.getApplications(remoteRegionsRef.get())  
            : eurekaTransport.queryClient.getVip(clientConfig.getRegistryRefreshSingleVipAddress(), remoteRegionsRef.get());  
   ** *****//拉取成功***  
***    *****if (httpResponse.getStatusCode() == Status.OK.getStatusCode()) \{**   
**        apps = httpResponse.getEntity();**  
**    \}**  
    logger.info(**"The response status is \{\}"**, httpResponse.getStatusCode());  
\}

### **5.3、增量拉取代码分析** ###

增量拉取，首先根据拉去信息进行判断是否为空，如果为空则进行全量拉去，反之更新本地缓存，getAndUpdateDelta(applications)代码如下：

**private void** getAndUpdateDelta(Applications applications) **throws** Throwable \{   
    **long** currentUpdateGeneration = fetchRegistryGeneration.get();  
**    *****//拉取信息***  
***    *****Applications delta = null;**  
    EurekaHttpResponse<Applications> httpResponse = eurekaTransport.queryClient.getDelta(remoteRegionsRef.get());  
    **if** (httpResponse.getStatusCode() == Status.OK.getStatusCode()) \{   
        **delta = httpResponse.getEntity();**  
    \}  
   ** *****//如果拉取失败进行全量拉取***  
***    *****if (delta == null) \{**   
        logger.warn(**"The server does not allow the delta revision to be applied because it is not safe. "**  
**                **\+ **"Hence got the full registry."**);  
        **getAndStoreFullRegistry();**  
    \} **else if** (fetchRegistryGeneration.compareAndSet(currentUpdateGeneration, currentUpdateGeneration + 1)) \{   
        logger.debug(**"Got delta update with apps hashcode \{\}"**, delta.getAppsHashCode());  
        String reconcileHashCode = **""**;  
        **if** (fetchRegistryUpdateLock.tryLock()) \{   
            **try** \{   
               ** *****//跟新本地缓存***  
***                *****updateDelta(delta);**  
                reconcileHashCode = getReconcileHashCode(applications);  
            \} **finally** \{   
                fetchRegistryUpdateLock.unlock();  
            \}  
        \} **else** \{   
            logger.warn(**"Cannot acquire update lock, aborting getAndUpdateDelta"**);  
        \}  
        *// There is a diff in number of instances for some reason*  
*        ***if** (!reconcileHashCode.equals(delta.getAppsHashCode()) || clientConfig.shouldLogDeltaDiff()) \{   
            reconcileAndLogDifference(delta, reconcileHashCode);  *// this makes a remoteCall*  
*        *\}  
    \} **else** \{   
        logger.warn(**"Not updating application delta as another thread is updating it already"**);  
        logger.debug(**"Ignoring delta update with apps hashcode \{\}, as another thread is updating it already"**, delta.getAppsHashCode());  
    \}  
\}

### **5.4、服务注册（重要步骤）** ###

**if** (**clientConfig**.shouldRegisterWithEureka() && **clientConfig**.shouldEnforceRegistrationAtInit()) \{   
    **try** \{   
      **  if (!register() ) \{** ** // 注册方法**  
            **throw new** IllegalStateException(**"Registration error at startup. Invalid server response."**);  
        \}  
    \} **catch** (Throwable th) \{   
        ***logger***.error(**"Registration error at startup: \{\}"**, th.getMessage());  
        **throw new** IllegalStateException(th);  
    \}  
\}

**register()****方法负责服务的注册源码分析：**

**boolean** register() **throws** Throwable \{   
    ***logger***.info(***PREFIX*** \+ **"\{\}: registering service..."**, **appPathIdentifier**);  
    EurekaHttpResponse<Void> httpResponse;  
    **try** \{

**// 把自身的实例发送给服务端**  
        **httpResponse = eurekaTransport.registrationClient.register(instanceInfo);**  
    \} **catch** (Exception e) \{   
        ***logger***.warn(***PREFIX*** \+ **"\{\} - registration failed \{\}"**, **appPathIdentifier**, e.getMessage(), e);  
        **throw** e;  
    \}  
    **if** (***logger***.isInfoEnabled()) \{   
        ***logger***.info(***PREFIX*** \+ **"\{\} - registration status: \{\}"**, **appPathIdentifier**, httpResponse.getStatusCode());  
    \}  
    **return** httpResponse.getStatusCode() == 204;  
\}

### **5.5、定时任务(重要步骤)** ###

initScheduledTasks()是负责定时任务的相关方法，具体源码实现如下：

**private void** initScheduledTasks() \{   
    **if** (**clientConfig**.shouldFetchRegistry()) \{   
**      ****   // 拉取服务默认30秒，eureka.client.register-fetch-interval-seconds**  
*        ***int** registryFetchIntervalSeconds = **clientConfig**.getRegistryFetchIntervalSeconds();  
        **int** expBackOffBound = **clientConfig**.getCacheRefreshExecutorExponentialBackOffBound();  
        **scheduler**.schedule( **new** TimedSupervisorTask(  **"cacheRefresh"**, **scheduler**, **cacheRefreshExecutor**,  
                 registryFetchIntervalSeconds, TimeUnit.***SECONDS***,expBackOffBound, **new** CacheRefreshThread()),  
                registryFetchIntervalSeconds, TimeUnit.***SECONDS***);  
    \}  
    **if** (**clientConfig**.shouldRegisterWithEureka()) \{   
        **int** renewalIntervalInSecs = **instanceInfo**.getLeaseInfo().getRenewalIntervalInSecs();  
        **int** expBackOffBound = **clientConfig**.getHeartbeatExecutorExponentialBackOffBound();  
        ***logger***.info(**"Starting heartbeat executor: "** \+ **"renew interval is: \{\}"**, renewalIntervalInSecs);  
       ** *****// Heartbeat timer****** 心跳服务，默认30秒***  
***       **** *scheduler.schedule(new TimedSupervisorTask( "heartbeat", scheduler, heartbeatExecutor,  
                        renewalIntervalInSecs,TimeUnit.*SECONDS*,expBackOffBound,new HeartbeatThread()),  
                renewalIntervalInSecs, TimeUnit.*SECONDS*);  
        *// InstanceInfo replicator*  
*        ***instanceInfoReplicator** = **new** InstanceInfoReplicator(**this**, **instanceInfo**,  
                **clientConfig**.getInstanceInfoReplicationIntervalSeconds(), 2); *// burstSize*  
*        ***statusChangeListener** = **new** ApplicationInfoManager.StatusChangeListener() \{   
            @Override  
            **public** String getId() \{   
                **return** **"statusChangeListener"**;  
            \}  
            @Override  
            **public void** notify(StatusChangeEvent statusChangeEvent) \{   
                **if** (InstanceStatus.***DOWN*** == statusChangeEvent.getStatus() ||  
                        InstanceStatus.***DOWN*** == statusChangeEvent.getPreviousStatus()) \{   
                    *// log at warn level if DOWN was involved*  
*                    ****logger***.warn(**"Saw local status change event \{\}"**, statusChangeEvent);  
                \} **else** \{   
                    ***logger***.info(**"Saw local status change event \{\}"**, statusChangeEvent);  
                \}  
                **instanceInfoReplicator**.onDemandUpdate();  
            \}  
        \};  
        **if** (**clientConfig**.shouldOnDemandUpdateStatusChange()) \{   
            **applicationInfoManager**.registerStatusChangeListener(**statusChangeListener**);  
        \}  
        **instanceInfoReplicator**.start(**clientConfig**.getInitialInstanceInfoReplicationIntervalSeconds());  
    \} **else** \{   
        ***logger***.info(**"Not registering with Eureka server per configuration"**);  
    \}  
\}

### **5.6、服务下线** ###

com.netflix.discovery.DiscoveryClient \#shutdown

@PreDestroy  
@Override  
**public synchronized void** shutdown() \{   
    **if** (isShutdown.compareAndSet(**false**, **true**)) \{   
        logger.info(**"Shutting down DiscoveryClient ..."**);  
        **if** (statusChangeListener != **null** && applicationInfoManager != **null**) \{   
            *//注销状态监听器*  
*            *applicationInfoManager.unregisterStatusChangeListener(statusChangeListener.getId());  
        \}  
        *//取消定时任务*  
*        *cancelScheduledTasks();  
        *// If APPINFO was registered*  
*        ***if** (applicationInfoManager != **null** && clientConfig.shouldRegisterWithEureka()  
                && clientConfig.shouldUnregisterOnShutdown()) \{   
            applicationInfoManager.setInstanceStatus(InstanceStatus.DOWN);  
            unregister();  
        \}  
        */**/关闭与server连接的客户端*  
*        ***if** (eurekaTransport != **null**) \{   
            eurekaTransport.shutdown();  
        \}  
        *//关闭相关监控*  
*        *heartbeatStalenessMonitor.shutdown();  
        registryStalenessMonitor.shutdown();  
        logger.info(**"Completed shut down of DiscoveryClient"**);  
    \}  
\}

# **二、EurekaServer运行流程分析** #

## **1、流程总览** ##

EurekaServer是服务的注册中心，负责Eureka Client的相关信息注册，主要职责：

*  服务注册
 *  接受心跳服务（client端通过定时任务初始化发送心跳）
 *  服务剔除
 *  服务下线
 *  集群同步

### **1.1、服务配置启动信息：** ###

**spring-cloud-netflix-eureka-server-2.0.0.RELEASE.jar!\\META-INF\\spring.factories**

**org.springframework.boot.autoconfigure.EnableAutoConfiguration**=**\\**  
**org.springframework.cloud.netflix.eureka.server.EurekaServerAutoConfiguration**

EurekaServerAutoConfiguration类代码：

@Configuration  
@Import(EurekaServerInitializerConfiguration.**class**)  
@ConditionalOnBean(EurekaServerMarkerConfiguration.Marker.**class**)  
@EnableConfigurationProperties(\{ EurekaDashboardProperties.**class**,InstanceRegistryProperties.**class** \})  
@PropertySource(**"classpath:/eureka/server.properties"**)  
**public class** EurekaServerAutoConfiguration **extends** WebMvcConfigurerAdapter \{   
*   ***private static final** String\[\] ***EUREKA\_PACKAGES*** = **new** String\[\] \{ **"com.netflix.discovery"**,**"com.netflix.eureka"** \};  
   @Autowired  
   **private** ApplicationInfoManager **applicationInfoManager**;  
   @Autowired  
   **private** EurekaServerConfig **eurekaServerConfig**;  
   @Autowired  
   **private** EurekaClientConfig **eurekaClientConfig**;  
   @Autowired  
   **private** EurekaClient **eurekaClient**;  
   @Autowired  
   **private** InstanceRegistryProperties **instanceRegistryProperties**;

EurekaServerAutoConfiguration 是通过配置文件注册的，注册方法入口：

@Bean  
**public** PeerAwareInstanceRegistry peerAwareInstanceRegistry( ServerCodecs serverCodecs) \{   
   **this**.**eurekaClient**.getApplications(); *// force initialization*  
*   ***return new** **InstanceRegistry**(**this**.**eurekaServerConfig**, **this**.**eurekaClientConfig**,  
         serverCodecs, **this**.**eurekaClient**,  
         **this**.**instanceRegistryProperties**.getExpectedNumberOfRenewsPerMin(),  
         **this**.**instanceRegistryProperties**.getDefaultOpenForTrafficCount());  
\}

这里面有个**InstanceRegistry****()**方法就是重点需要关注的了。

### **1.2、整个接口关系如下如所示：** ###

![20200311212032580.png][]

**所有业务的核心方式实现主要集中在PeerAwareInstanceRegistryImpl与AbstractInstanceRegistry这两个类中，****其中服务注册、接受心跳服务、服务剔除都是在AbstractInstanceRegistry中处理的，服务下线与集群同步是在PeerAwareInstanceRegistryImpl类中处理的。**

### **1.3、首先看下最上层的接口** ###

**public interface** LeaseManager<T> \{   
    ***//注册***  
*    ***void** register(T var1, **int** var2, **boolean** var3);  
  ***  //下线***  
*    ***boolean** cancel(String var1, String var2, **boolean** var3);  
    ***//跟新***  
*    ***boolean** renew(String var1, String var2, **boolean** var3);  
 ***   //服务剔除***  
*    ***void** evict();  
\}

PeerAwareInstanceRegistryImpl是一个子类的实现，在上面的基础上扩展对集群的同步操作，使Eureka Server集群信息保持一致。

## **2、服务注册** ##

com.netflix.eureka.registry.AbstractInstanceRegistry\#register(); 这方法是负责服务的注册的:

**public void** register(InstanceInfo registrant, **int** leaseDuration, **boolean** isReplication) \{   
    **try** \{   
        *//获取读锁*  
*        ***this**.read.lock();  
        *//* ***TODO gMap  其实可以发现，这里注册中心其实是个ConcurrentHashMap***  
***        ***Map<String, Lease<InstanceInfo>> gMap = (Map)**this**.registry.get(registrant.getAppName());  
        EurekaMonitors.REGISTER.increment(isReplication);  
        **if**(gMap == **null**) \{   
            ConcurrentHashMap<String, Lease<InstanceInfo>> gNewMap = **new** ConcurrentHashMap();  
            *//****TODO  key为appName,如果存在，返回存在的值，否则添加，返回null***  
***           ****** *****gMap = (Map)this.registry.putIfAbsent(registrant.getAppName(), gNewMap);**  
            **if**(gMap == **null**) \{   
                gMap = gNewMap;  
            \}  
        \}  
     **   *****//TODO 根据instanceId获取实例的租约***  
***        *****Lease<InstanceInfo> existingLease = (Lease)((Map)gMap).get(registrant.getId());**  
        **if**(existingLease != **null** && existingLease.getHolder() != **null**) \{   
            Long existingLastDirtyTimestamp = ((InstanceInfo)existingLease.getHolder()).getLastDirtyTimestamp();  
            Long registrationLastDirtyTimestamp = registrant.getLastDirtyTimestamp();  
            *//* ***TODO 如果该实例的租约已经存在，比较最后的更新时间戳大小，取最大值的注册信息信息***  
***           ****** *****if**(existingLastDirtyTimestamp.longValue() > registrationLastDirtyTimestamp.longValue()) \{   
                registrant = (InstanceInfo)existingLease.getHolder();  
            \}  
        \} **else** \{   
            Object var6 = **this**.lock;  
            *//****TODO 如果租约不存在，注册一个新的实例***  
***            *****synchronized**(**this**.lock) \{   
                **if**(**this**.expectedNumberOfRenewsPerMin > 0) \{   
                    **this**.expectedNumberOfRenewsPerMin += 2;  
                    **this**.numberOfRenewsPerMinThreshold = (**int**)((**double**)**this**.expectedNumberOfRenewsPerMin

\* **this**.serverConfig.getRenewalPercentThreshold());  
                \}  
            \}  
            logger.debug(**"No previous lease information found; it is new registration"**);  
        \}  
        *//****TODO 创建新的租约***  
***        ***Lease<InstanceInfo> lease = **new** Lease(registrant, leaseDuration);  
        **if**(existingLease != **null**) \{   
            lease.setServiceUpTimestamp(existingLease.getServiceUpTimestamp());  
        \}  
        *//****TODO 保存租约到map中***  
***       ****** ***((Map)gMap).put(registrant.getId(), lease);  
        *//****TODO 获得最近注册队列***  
***        ***AbstractInstanceRegistry.CircularQueue var20 = **this**.recentRegisteredQueue;  
        **synchronized**(**this**.recentRegisteredQueue) \{   
            **this**.recentRegisteredQueue.add(**new** Pair(Long.*valueOf*(System.*currentTimeMillis*()), registrant.getAppName() + **"("** \+ registrant.getId() + **")"**));  
        \}  
        **if**(!InstanceStatus.UNKNOWN.equals(registrant.getOverriddenStatus())) \{   
            logger.debug(**"Found overridden status \{\} for instance \{\}. Checking to see if needs to be add to the overrides"**, registrant.getOverriddenStatus(), registrant.getId());  
            **if**(!**this**.overriddenInstanceStatusMap.containsKey(registrant.getId())) \{   
                logger.info(**"Not found overridden id \{\} and hence adding it"**, registrant.getId());  
                **this**.overriddenInstanceStatusMap.put(registrant.getId(), registrant.getOverriddenStatus());  
            \}  
        \}  
        InstanceStatus overriddenStatusFromMap = (InstanceStatus)**this**.overriddenInstanceStatusMap.get(registrant.getId());  
        **if**(overriddenStatusFromMap != **null**) \{   
            logger.info(**"Storing overridden status \{\} from map"**, overriddenStatusFromMap);  
            registrant.setOverriddenStatus(overriddenStatusFromMap);  
        \}  
        InstanceStatus overriddenInstanceStatus = **this**.getOverriddenInstanceStatus(registrant, existingLease, isReplication);  
        registrant.setStatusWithoutDirty(overriddenInstanceStatus);  
        **if**(InstanceStatus.UP.equals(registrant.getStatus())) \{   
            lease.serviceUp();  
        \}  
        registrant.setActionType(ActionType.ADDED);  
        **this**.recentlyChangedQueue.add(**new** AbstractInstanceRegistry.RecentlyChangedItem(lease));  
        registrant.setLastUpdatedTimestamp();  
        **this**.invalidateCache(registrant.getAppName(), registrant.getVIPAddress(), registrant.getSecureVipAddress());  
    \} **finally** \{ *//释放锁*  
*        ***this**.read.unlock();  
    \}  
\}

## **3、接受心跳服务** ##

在Eureka Client完成服务的注册后，需要定时向Eureka Server发送心跳请求（默认30s），维持自己在EurekaServer的租约有效性com.netflix.eureka.registry.AbstractInstanceRegistry.renew(appName, id,

isReplication);方法源码如下所示：

**public boolean** renew(String appName, String id, **boolean** isReplication) \{   
    EurekaMonitors.RENEW.increment(isReplication);  
    *//****TODO 根据appName获取服务集群租约集合***  
***   ****** ***Map<String, Lease<InstanceInfo>> gMap = (Map) **this**.registry.get(appName);  
    Lease<InstanceInfo> leaseToRenew = **null**;  
    **if** (gMap != **null**) \{   
        leaseToRenew = (Lease) gMap.get(id);  
    \}  
    *//****TODO 如果租约不存在，直接返回false***  
***    *****if** (leaseToRenew == **null**) \{   
        EurekaMonitors.RENEW\_NOT\_FOUND.increment(isReplication);  
        logger.warn(**"DS: Registry: lease doesn't exist, registering resource: \{\} - \{\}"**, appName, id);  
        **return false**;  
    \} **else** \{   
        InstanceInfo instanceInfo = (InstanceInfo) leaseToRenew.getHolder();  
        **if** (instanceInfo != **null**) \{   
            *//****TODO 得到服务的最终状态***  
***            ***InstanceStatus overriddenInstanceStatus = **this**.getOverriddenInstanceStatus(instanceInfo, leaseToRenew, isReplication);  
            **if** (overriddenInstanceStatus == InstanceStatus.UNKNOWN) \{   
                *//****TODO 如果状态为UNKNOWN，取消续约***  
***                ***logger.info(**"Instance status UNKNOWN possibly due to deleted override for instance \{\}; re-register required"**, instanceInfo.getId());  
                EurekaMonitors.RENEW\_NOT\_FOUND.increment(isReplication);  
                **return false**;  
            \}  
            **if** (!instanceInfo.getStatus().equals(overriddenInstanceStatus)) \{   
                logger.info(**"The instance status \{\} is different from overridden instance status \{\} for instance \{\}. Hence setting the status to overridden status"**, **new** Object\[\]\{instanceInfo.getStatus().name(), instanceInfo.getOverriddenStatus().name(), instanceInfo.getId()\});  
                instanceInfo.setStatusWithoutDirty(overriddenInstanceStatus);  
            \}  
        \}  
        **this**.renewsLastMin.increment();  
        *//****TODO 跟新续约有效时间***  
***        ***leaseToRenew.renew();  
        **return true**;  
    \}  
\}

## **4、服务剔除** ##

如果Eureka Client在注册后，由于服务的崩溃或网络异常导致既没有续约，也没有下线，那么服务就处于不可知的状态，需要剔除这些服务。

代码见com.netflix.eureka.registry.AbstractInstanceRegistry\#evict(long)方法。

这是个定时任务调用的方法：com.netflix.eureka.registry.AbstractInstanceRegistry\#postInit()中使用

AbstractInstanceRegistry.EvictionTask()负责调用(默认60s)。

1）AbstractInstanceRegistry\#evict(long)剔除方法源码：

**public void** evict(**long** additionalLeaseMs) \{   
    logger.debug(**"Running the evict task"**);  
    *//****TODO 如果自我保护状态，不允许剔除服务******（******前面配置关系自我保护机制代码判断******）***  
***   ****** *****if**(!**this**.isLeaseExpirationEnabled()) \{   
        logger.debug(**"DS: lease expiration is currently disabled."**);  
    \} **else** \{   
        List<Lease<InstanceInfo>> expiredLeases = **new** ArrayList();  
        *//****TODO 遍历注册表registry，获取所有过期的租约***  
***        ***Iterator var4 = **this**.registry.entrySet().iterator();  
        **while**(**true**) \{   
            Map leaseMap;  
            **do** \{   
                **if**(!var4.hasNext()) \{   
                    *//****TODO 获取注册表租约总数***  
***                    *****int** registrySize = (**int**)**this**.getLocalRegistrySize();  
                    **int** registrySizeThreshold = (**int**)((**double**)registrySize \* **this**.serverConfig.getRenewalPercentThreshold());  
                    *//****TODO 计算最多允许剔除的阈值***  
***                    *****int** evictionLimit = registrySize - registrySizeThreshold;  
                    *//****TODO 两者中取小的值，为本常剔除的数量***  
***                  ******  *****int** toEvict = Math.min(expiredLeases.size(), evictionLimit);  
                    **if**(toEvict > 0) \{   
                        logger.info(**"Evicting \{\} items (expired=\{\}, evictionLimit=\{\})"**, **new** Object\[\]\{Integer.*valueOf*(toEvict), Integer.valueOf(expiredLeases.size()), Integer.*valueOf*(evictionLimit)\});  
                        Random random = **new** Random(System.*currentTimeMillis*());  
                        *//****TODO 逐个剔除***  
***                        *****for**(**int** i = 0; i < toEvict; ++i) \{   
                            **int** next = i + random.nextInt(expiredLeases.size() - i);  
                            Collections.swap(expiredLeases, i, next);  
                            Lease<InstanceInfo> lease = (Lease)expiredLeases.get(i);  
                            String appName = ((InstanceInfo)lease.getHolder()).getAppName();  
                            String id = ((InstanceInfo)lease.getHolder()).getId();  
                            EurekaMonitors.EXPIRED.increment();  
                            logger.warn(**"DS: Registry: expired lease for \{\}/\{\}"**, appName, id);  
                            *//****TODO 剔除***  
***                            *****this**.internalCancel(appName, id, **false**);  
                        \}  
                    \}  
                    **return**;  
                \}  
                Entry<String, Map<String, Lease<InstanceInfo>>> groupEntry = (Entry)var4.next();  
                leaseMap = (Map)groupEntry.getValue();  
            \} **while**(leaseMap == **null**);  
            Iterator var7 = leaseMap.entrySet().iterator();  
            **while**(var7.hasNext()) \{   
                Entry<String, Lease<InstanceInfo>> leaseEntry = (Entry)var7.next();  
                Lease<InstanceInfo> lease = (Lease)leaseEntry.getValue();  
                **if**(lease.isExpired(additionalLeaseMs) && lease.getHolder() != **null**) \{   
                    expiredLeases.add(lease);  
                \}  
            \}  
        \}  
    \}  
\}

## **5、服务下线** ##

EurekaClient在应用销毁时候，会向Eureka Server发送下线请求，对于服务端的服务下线，其主要代码对应在com.netflix.eureka.registry.AbstractInstanceRegistry\#cancel()方法中

PeerAwareInstanceRegistryImpl\#cancel()方法调用父类方法AbstractInstanceRegistry\#cancel()方法实现：

@Override  
**public boolean** cancel(**final** String appName, **final** String id, **final boolean** isReplication) \{   
    **if** (**super**.cancel(appName, id, isReplication)) \{   
        replicateToPeers(Action.***Cancel***, appName, id, **null**, **null**, isReplication);  
        **synchronized** (**lock**) \{   
            **if** (**this**.**expectedNumberOfRenewsPerMin** > 0) \{   
                *// Since the client wants to cancel it, reduce the threshold (1 for 30 seconds, 2 for a minute)*  
*                ***this**.**expectedNumberOfRenewsPerMin** = **this**.**expectedNumberOfRenewsPerMin** \- 2;  
                **this**.**numberOfRenewsPerMinThreshold** =  
                        (**int**) (**this**.**expectedNumberOfRenewsPerMin** \* **serverConfig**.getRenewalPercentThreshold());  
            \}  
        \}  
        **return true**;  
    \}  
    **return false**;  
\}

com.netflix.eureka.registry.AbstractInstanceRegistry\#cancel方法源码：

**public boolean** cancel(String appName, String id, **boolean** isReplication) \{   
    **return this**.internalCancel(appName, id, isReplication);  
\}

**protected boolean** internalCancel(String appName, String id, **boolean** isReplication) \{   
    **boolean** var10;  
    **try** \{   
        *//读锁，防止被其他线程进行修改*  
*        ***this**.read.lock();  
        EurekaMonitors.CANCEL.increment(isReplication);  
        *//根据appName获取服务实例集群*  
*        *Map<String, Lease<InstanceInfo>> gMap = (Map)**this**.registry.get(appName);  
        Lease<InstanceInfo> leaseToCancel = **null**;  
        *//****TODO 移除服务实例租约***  
***        *****if**(gMap != **null**) \{   
            leaseToCancel = (Lease)gMap.remove(id);  
        \}  
        AbstractInstanceRegistry.CircularQueue var6 = **this**.recentCanceledQueue;  
        **synchronized**(**this**.recentCanceledQueue) \{   
            **this**.recentCanceledQueue.add(**new** Pair(Long.*valueOf*(System.*currentTimeMillis*()), appName + **"("** \+ id + **")"**));  
        \}  
        InstanceStatus instanceStatus = (InstanceStatus)**this**.overriddenInstanceStatusMap.remove(id);  
        **if**(instanceStatus != **null**) \{   
            logger.debug(**"Removed instance id \{\} from the overridden map which has value \{\}"**, id, instanceStatus.name());  
        \}  
        *//****TODO 租约不存在，返回false***  
***        *****if**(leaseToCancel == **null**) \{   
            EurekaMonitors.CANCEL\_NOT\_FOUND.increment(isReplication);  
            logger.warn(**"DS: Registry: cancel failed because Lease is not registered for: \{\}/\{\}"**, appName, id);  
            **boolean** var17 = **false**;  
            **return** var17;  
        \}  
        *//****TODO 设置租约的下线时间***  
***        ***leaseToCancel.cancel();  
        InstanceInfo instanceInfo = (InstanceInfo)leaseToCancel.getHolder();  
        String vip = **null**;  
        String svip = **null**;  
        **if**(instanceInfo != **null**) \{   
            instanceInfo.setActionType(ActionType.DELETED);  
            **this**.recentlyChangedQueue.add(**new** AbstractInstanceRegistry.RecentlyChangedItem(leaseToCancel));  
            instanceInfo.setLastUpdatedTimestamp();  
            vip = instanceInfo.getVIPAddress();  
            svip = instanceInfo.getSecureVipAddress();  
        \}  
        *//****TODO 设置缓存过期***  
***        *****this**.invalidateCache(appName, vip, svip);  
        logger.info(**"Cancelled instance \{\}/\{\} (replication=\{\})"**, **new** Object\[\]\{appName, id, Boolean.*valueOf*(isReplication)\});  
        var10 = **true**;  
    \} **finally** \{ *//释放锁*  
*        ***this**.read.unlock();  
    \}  
    **return** var10;  
\}

## **6、集群同步** ##

如果Eureka Server是通过集群方式进行部署，为了为维护整个集群中注册表数据一致性所以集群同步也是非常重要得事情。

**集群同步分为两部分：**

*  EurekaServer在启动过程中从他的peer节点中拉取注册表信息，并讲这些服务实例注册到本地注册表中；
 *  另一部分是eureka server每次对本地注册表进行操作时，同时会讲操作同步到他的peer节点中，达到数据一致；

### **6.1、Eureka Server初始化本地注册表信息** ###

在eureka server启动过程中，会从它的peer节点中拉取注册表来初始化本地注册表，这部分主要通过

com.netflix.eureka.registry.PeerAwareInstanceRegistryImpl\#syncUp()方法实现，他从可能存在的peer节点中，拉取peer节点中的注册表信息，并将其中的服务实例的信息注册到本地注册表中。

**public int** syncUp() \{   
    *// Copy entire entry from neighboring DS node*  
*    ***int** count = 0;  
    **for** (**int** i = 0; ((i < serverConfig.getRegistrySyncRetries()) && (count == 0)); i++) \{   
        **if** (i > 0) \{   
            **try** \{   
                *//****TODO 根据配置休停下***  
***                ***Thread.*sleep*(serverConfig.getRegistrySyncRetryWaitMs());  
            \} **catch** (InterruptedException e) \{   
                logger.warn(**"Interrupted during registry transfer.."**);  
                **break**;  
            \}  
        \}  
        *//****TODO 获取所有的服务实例***  
***      ******  ***Applications apps = eurekaClient.getApplications();  
        **for** (Application app : apps.getRegisteredApplications()) \{   
            **for** (InstanceInfo instance : app.getInstances()) \{   
                **try** \{   
                    *//****TODO 判断是否可以注册***  
***                    *****if** (isRegisterable(instance)) \{   
                        *//****TODO 注册到自身的注册表中***  
***                    ******    ***register(instance, instance.getLeaseInfo().getDurationInSecs(), **true**);  
                        count++;  
                    \}  
                \} **catch** (Throwable t) \{   
                    logger.error(**"During DS init copy"**, t);  
                \}  
            \}  
        \}  
    \}  
    **return** count;  
\}

通过这一步保证了eureka启动时的数据一致性。

### **6.2、Eureka Server之间注册表信息同步复制** ###

为了保证Eureka Server集群运行时候注册表的信息一致性，每个eureka server在对本地注册表进行管理操作时，会将相应的信息同步到peer节点中。其中以下几个方法：

com.netflix.eureka.registry.PeerAwareInstanceRegistryImpl\#cancel();

com.netflix.eureka.registry.PeerAwareInstanceRegistryImpl\#register();

com.netflix.eureka.registry.PeerAwareInstanceRegistryImpl\#renew();

等方法中，都回调用com.netflix.eureka.registry.PeerAwareInstanceRegistryImpl\#replicateToPeers()方法：

**private void** replicateToPeers(Action action, String appName, String id,InstanceInfo info */\* optional \*/*,  
                              InstanceStatus newStatus */\* optional \*/*, **boolean** isReplication) \{   
    Stopwatch tracer = action.getTimer().start();  
    **try** \{   
        **if** (isReplication) \{   
            numberOfReplicationsLastMin.increment();  
        \}  
        *// If it is a replication already, do not replicate again as this will create a poison replication*  
*        ***if** (peerEurekaNodes == Collections.EMPTY\_LIST || isReplication) \{   
            **return**;  
        \}  
        *//****TODO 向peer集群中的每一个peer进行同步***  
***       ****** *****for** (**final** PeerEurekaNode node : peerEurekaNodes.getPeerEurekaNodes()) \{   
            *// If the url represents this host, do not replicate to yourself.*  
*            ***if** (peerEurekaNodes.isThisMyUrl(node.getServiceUrl())) \{   
                **continue**;  
            \}  
            *//****TODO 根据action调用不同的同步请求***  
***           ****** ***replicateInstanceActionsToPeers(action, appName, id, info, newStatus, node);  
        \}  
    \} **finally** \{   
        tracer.stop();  
    \}  
\}

根据action调用不同的同步请求：

**private void** replicateInstanceActionsToPeers(Action action, String appName, String id,   
                InstanceInfo info, InstanceStatus newStatus,PeerEurekaNode node) \{   
    **try** \{   
        InstanceInfo infoFromRegistry = **null**;  
        CurrentRequestVersion.*set*(Version.***V2***);  
        **switch** (action) \{   
            **case** ***Cancel***:  
                node.cancel(appName, id);  
                **break**;  
            **case** ***Heartbeat***:  
                InstanceStatus overriddenStatus = **overriddenInstanceStatusMap**.get(id);  
                infoFromRegistry = getInstanceByAppAndId(appName, id, **false**);  
                node.heartbeat(appName, id, infoFromRegistry, overriddenStatus, **false**);  
                **break**;  
            **case** ***Register***:  
                node.register(info);  
                **break**;  
            **case** ***StatusUpdate***:  
                infoFromRegistry = getInstanceByAppAndId(appName, id, **false**);  
                node.statusUpdate(appName, id, newStatus, infoFromRegistry);  
                **break**;  
            **case** ***DeleteStatusOverride***:  
                infoFromRegistry = getInstanceByAppAndId(appName, id, **false**);  
                node.deleteStatusOverride(appName, id, infoFromRegistry);  
                **break**;  
        \}  
    \} **catch** (Throwable t) \{   
        ***logger***.error(**"Cannot replicate information to \{\} for action \{\}"**, node.getServiceUrl(), action.name(), t);  
    \}  
\}

**public void** heartbeat(**final** String appName, **final** String id,**final** InstanceInfo info,  
                       **final** InstanceStatus overriddenStatus, **boolean** primeConnection) **throws** Throwable \{   
    **if** (primeConnection) \{   
        *// We do not care about the result for priming request.*  
*        ***replicationClient**.sendHeartBeat(appName, id, info, overriddenStatus);  
        **return**;  
    \}  
    ReplicationTask replicationTask = **new** InstanceReplicationTask(**targetHost**, Action.***Heartbeat***, info, overriddenStatus, **false**) \{   
        @Override  
        **public** EurekaHttpResponse<InstanceInfo> execute() **throws** Throwable \{   
            **return replicationClient**.sendHeartBeat(appName, id, info, overriddenStatus);  
        \}  
        @Override  
        **public void** handleFailure(**int** statusCode, Object responseEntity) **throws** Throwable \{   
            **super**.handleFailure(statusCode, responseEntity);  
            **if** (statusCode == 404) \{   
                ***logger***.warn(**"\{\}: missing entry."**, getTaskName());  
                **if** (info != **null**) \{   
                    ***logger***.warn(**"\{\}: cannot find instance id \{\} and hence replicating the instance with status \{\}"**,  
                            getTaskName(), info.getId(), info.getStatus());  
                    register(info);  
                \}  
            \} **else if** (**config**.shouldSyncWhenTimestampDiffers()) \{   
                InstanceInfo peerInstanceInfo = (InstanceInfo) responseEntity;  
                **if** (peerInstanceInfo != **null**) \{   
                    syncInstancesIfTimestampDiffers(appName, id, info, peerInstanceInfo);  
                \}  
            \}  
        \}  
    \};  
    **long** expiryTime = System.*currentTimeMillis*() + *getLeaseRenewalOf*(info);  
    **batchingDispatcher**.process(*taskId*(**"heartbeat"**, info), replicationTask, expiryTime);  
\}

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM3NjYxNDU4_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20200311212151826.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM3NjYxNDU4,size_16,color_FFFFFF,t_70
[20200311212032580.png]: https://img-blog.csdnimg.cn/20200311212032580.png