Spring Cloud之Eureka源码分析2

本章主要介绍Eureka Client端源码分析。
客户端主要是向Server服务端发送Http请求，主要有注册，心跳续约，获取注册信息等功能

实例配置

在分析源码之前，需要查看下客户端配置文件
application.yml

server:
  port: 8020
eureka:
  client:
    serviceUrl:
        defaultZone: http://localhost:8000/eureka/  #eureka服务端提供的注册地址 参考服务端配置的这个路径

  instance:
    instance-id: client-0 #此实例注册到eureka服务端的唯一的实例ID
    prefer-ip-address: true #是否显示IP地址
    leaseRenewalIntervalInSeconds: 10 #eureka客户需要多长时间发送心跳给eureka服务器，表明它仍然活着,默认为30 秒 (与下面配置的单位都是秒)
    leaseExpirationDurationInSeconds: 30 #Eureka服务器在接收到实例的最后一次发出的心跳后，需要等待多久才可以将此实例删除，默认为90秒

spring:
  application:
    name: service-client #此实例注册到eureka服务端的name

访问服务端localhost:8000的注册信息

localhost.png

EurekaClientAutoConfiguration

自动配置类，首先要从依赖包的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.netflix.eureka.reactive.EurekaReactiveDiscoveryClientConfiguration,\
org.springframework.cloud.netflix.eureka.loadbalancer.LoadBalancerEurekaAutoConfiguration

org.springframework.cloud.bootstrap.BootstrapConfiguration=\
org.springframework.cloud.netflix.eureka.config.EurekaDiscoveryClientConfigServiceBootstrapConfiguration

其中最重要的是EurekaClientAutoConfiguration类

EurekaClientAutoConfiguration.png

只要类中存在EurekaClientConfig类所在的依赖包eureka-client-xx.jar就可以加载这个类

EurekaClientConfig.png

初始化EurekaClientAutoConfiguration类中的方法Bean加载到spring容器中：
1、EurekaAutoServiceRegistration 用来注册

2、RefreshableEurekaClientConfiguration 用来开启定时任务

注册功能实现

1、EurekaAutoServiceRegistration
实例化EurekaAutoServiceRegistration对象，并放到spring容器中
org.springframework.cloud.netflix.eureka.serviceregistry.EurekaAutoServiceRegistration

EurekaAutoServiceRegistration.png

2、start
由于EurekaAutoServiceRegistration类实现了SmartLifecycle,SmartApplicationListener等接口，所以会在容器初始化完成之后调用EurekaAutoServiceRegistration#start方法。

start.png

调用EurekaServiceRegistry的注册方法和发布InstanceRegisteredEvent事件
3、register
org.springframework.cloud.netflix.eureka.serviceregistry.EurekaServiceRegistry#register

EurekaServiceRegistry.png

调用ApplicationInfoManager应用信息管理设置实例初始化状态信息initialStatus
4、setInstanceStatus
com.netflix.appinfo.ApplicationInfoManager#setInstanceStatus
设置实例状态信息并调用监听器notify方法

setInstanceStatus.png

5、notify
com.netflix.appinfo.ApplicationInfoManager.StatusChangeListener#notify
调用StatusChangeListener状态改变监听器的notify方法。
这个对象实现在com.netflix.discovery.DiscoveryClient#initScheduledTasks方法内

image.png

6、onDemandUpdate
com.netflix.discovery.InstanceInfoReplicator#onDemandUpdate

image.png

7、run
由于InstanceInfoReplicator 类实现了Runnable接口，所以会调用这个run方法

image.png

8、register

com.netflix.discovery.DiscoveryClient#register
这里就到了注册客户端的地方

9、register
com.netflix.discovery.shared.transport.jersey.AbstractJerseyEurekaHttpClient#register
封装http请求并调用到服务端方法

AbstractJerseyEurekaHttpClient.png

所以，封装的url是：服务端的serviceUrl:defaultZone/apps/客户端应用名
调到Eureka Server端
com.netflix.eureka.resources.ApplicationResource#addInstance

addInstance.png

定时刷新任务

包括缓存更新与心跳续约，通过RefreshableEurekaClientConfiguration类开始初始化，并最终通过initScheduledTasks方法开启定时调度器任务

RefreshableEurekaClientConfiguration

EurekaClientAutoConfiguration.RefreshableEurekaClientConfiguration

@Configuration(proxyBeanMethods = false)
    @ConditionalOnRefreshScope
    protected static class RefreshableEurekaClientConfiguration {

        @Autowired
        private ApplicationContext context;

        @Autowired
        private AbstractDiscoveryClientOptionalArgs<?> optionalArgs;

        @Bean(destroyMethod = "shutdown")
        @ConditionalOnMissingBean(value = EurekaClient.class,
                search = SearchStrategy.CURRENT)
        @org.springframework.cloud.context.config.annotation.RefreshScope
        @Lazy
        public EurekaClient eurekaClient(ApplicationInfoManager manager,
                EurekaClientConfig config, EurekaInstanceConfig instance,
                @Autowired(required = false) HealthCheckHandler healthCheckHandler) {
            // If we use the proxy of the ApplicationInfoManager we could run into a
            // problem
            // when shutdown is called on the CloudEurekaClient where the
            // ApplicationInfoManager bean is
            // requested but wont be allowed because we are shutting down. To avoid this
            // we use the
            // object directly.
            ApplicationInfoManager appManager;
            if (AopUtils.isAopProxy(manager)) {
                appManager = ProxyUtils.getTargetObject(manager);
            }
            else {
                appManager = manager;
            }
            CloudEurekaClient cloudEurekaClient = new CloudEurekaClient(appManager,
                    config, this.optionalArgs, this.context);
            cloudEurekaClient.registerHealthCheck(healthCheckHandler);
            return cloudEurekaClient;
        }
}

这个类是程序实现定时刷新任务开始的地方，主要是通过new CloudEurekaClient()方法创建Cloud客户端类（CloudEurekaClient）

CloudEurekaClient

下面主要查看CloudEurekaClient的调用链
1、CloudEurekaClient#CloudEurekaClient
org.springframework.cloud.netflix.eureka.CloudEurekaClient#CloudEurekaClient
调用CloudEurekaClient类的父类，和对applicationInfoManager、publisher、eurekaTransportField等属性值赋值

CloudEurekaClient.png

2、DiscoveryClient#DiscoveryClient
com.netflix.discovery.DiscoveryClient#DiscoveryClient

DiscoveryClient.png

3、DiscoveryClient#DiscoveryClient
com.netflix.discovery.DiscoveryClient#DiscoveryClient
构造器类调用，传入获取备用服务器backupRegistryInstance实例的get方法。不在这里调用

public DiscoveryClient(ApplicationInfoManager applicationInfoManager, final EurekaClientConfig config, AbstractDiscoveryClientOptionalArgs args, EndpointRandomizer randomizer) {
        this(applicationInfoManager, config, args, new Provider<BackupRegistry>() {
            private volatile BackupRegistry backupRegistryInstance;

            @Override
            public synchronized BackupRegistry get() {
                if (backupRegistryInstance == null) {
                    String backupRegistryClassName = config.getBackupRegistryImpl();
                    if (null != backupRegistryClassName) {
                        try {
                            backupRegistryInstance = (BackupRegistry) Class.forName(backupRegistryClassName).newInstance();
                            logger.info("Enabled backup registry of type {}", backupRegistryInstance.getClass());
                        } catch (InstantiationException e) {
                            logger.error("Error instantiating BackupRegistry.", e);
                        } catch (IllegalAccessException e) {
                            logger.error("Error instantiating BackupRegistry.", e);
                        } catch (ClassNotFoundException e) {
                            logger.error("Error instantiating BackupRegistry.", e);
                        }
                    }

                    if (backupRegistryInstance == null) {
                        logger.warn("Using default backup registry implementation which does not do anything.");
                        backupRegistryInstance = new NotImplementedRegistryImpl();
                    }
                }

                return backupRegistryInstance;
            }
        }, randomizer);
    }

4、DiscoveryClient#DiscoveryClient
定义调度器类、心跳执行器和缓存刷新执行器等的定义。
com.netflix.discovery.DiscoveryClient#DiscoveryClient

DiscoveryClient(ApplicationInfoManager applicationInfoManager, EurekaClientConfig config, AbstractDiscoveryClientOptionalArgs args,
                    Provider<BackupRegistry> backupRegistryProvider, EndpointRandomizer endpointRandomizer) {
        //  ........前半部分省略
        try {
            // 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

            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);
        }

        // 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());
    }

这个方法主要流程：
①、这个方法前半部分是初始化属性值。
②、根据客户端client配置文件，config.shouldFetchRegistry()是否获取注册表信息和
config.shouldRegisterWithEureka()是否注册到eureka上来对属性赋值，或直接返回
③、初始化调度器scheduler、两个线程池执行器heartbeatExecutor（心跳续约）和cacheRefreshExecutor（缓存刷新，定时获取注册信息表）
④、在获取服务注册信息条件下，没有获取到信息或异常即fetchRegistry(false)返回false。可以从备用服务器获取调用fetchRegistryFromBackup()方法，内部实现方法调用备用服务器类的get方法backupRegistryProvider.get()
⑤、初始化调度器任务方法initScheduledTasks()

初始化调度器任务initScheduledTasks

调度器任务包括：
1、定时刷新缓存注册表信息，分为全量获取和增量获取
2、定时向服务端发送心跳续约
3、状态改变监听器执行
这里不仅包括这些定时任务，注册也是在这里调用状态改变监听器StatusChangeListener的notify方法
com.netflix.discovery.DiscoveryClient#initScheduledTasks

 /**
     * Initializes all scheduled tasks.
     */
    private void initScheduledTasks() {
        if (clientConfig.shouldFetchRegistry()) {
            // registry cache refresh timer
            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
            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");
        }
    }

1、心跳续约

1、initScheduledTasks
com.netflix.discovery.DiscoveryClient#initScheduledTasks
TimedSupervisorTask继承了TimerTask，TimerTask实现了Runnable
TimedSupervisorTask类的构造方法

public TimedSupervisorTask(String name, ScheduledExecutorService scheduler, ThreadPoolExecutor executor,
                               int timeout, TimeUnit timeUnit, int expBackOffBound, Runnable task) {
        this.scheduler = scheduler;
        this.executor = executor;
        this.timeoutMillis = timeUnit.toMillis(timeout);
        this.task = task;
        this.delay = new AtomicLong(timeoutMillis);
        this.maxDelay = timeoutMillis * expBackOffBound;
        Monitors.registerObject(name, this);
    }

heartbeatExecutor.png

2、HeartbeatThread
执行TimedSupervisorTask的task任务，在给定的间隔内执行心跳续约任务
com.netflix.discovery.DiscoveryClient.HeartbeatThread

HeartbeatThread.png

3、renew
续约任务，续约成功更新lastSuccessfulHeartbeatTimestamp参数。通过REST方式进行续订
com.netflix.discovery.DiscoveryClient#renew

renew.png

4、sendHeartBeat
拼接http请求，发送心跳
com.netflix.discovery.shared.transport.jersey.AbstractJerseyEurekaHttpClient#sendHeartBeat

sendHeartBeat.png

客户端appName=SERVICE-CLIENT，id是instance-id属性值client-0

服务端调用到renewLease方法续约，appName和id与客户端传过来的相同
com.netflix.eureka.resources.InstanceResource#renewLease

image.png

全量拉取和增量拉取

在定时刷新缓存实现获取注册信息，分为全量拉取和增量拉取
创建TimedSupervisorTask调度任务类，传入cacheRefreshExecutor执行器、CacheRefreshThread任务类、从服务端获取注册信息的时间间隔RegistryFetchIntervalSeconds等参数信息

image.png

1、run
定时执行CacheRefreshThread类的run方法

CacheRefreshThread.png

2、refreshRegistry
首先对remoteRegionsModified参数进行判断，这样可以确保对远程区域进行动态更改时可以获取数据。如果更改则remoteRegionsModified=true，只进行全量拉取
com.netflix.discovery.DiscoveryClient#refreshRegistry

refreshRegistry.png

3、fetchRegistry
com.netflix.discovery.DiscoveryClient#fetchRegistry
这个方法是决定了使用那种方式拉取

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)) //Client application does not have latest library supporting delta
            {
                logger.info("Disable delta property : {}", clientConfig.shouldDisableDelta());
                logger.info("Single vip registry refresh property : {}", clientConfig.getRegistryRefreshSingleVipAddress());
                logger.info("Force full registry fetch : {}", forceFullRegistryFetch);
                logger.info("Application is null : {}", (applications == null));
                logger.info("Registered Applications size is zero : {}",
                        (applications.getRegisteredApplications().size() == 0));
                logger.info("Application version is -1: {}", (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();
            }
        }

        // Notify about cache refresh before updating the instance remote status
        onCacheRefreshed();

        // Update remote status based on refreshed data held in the cache
        updateInstanceRemoteStatus();

        // registry was fetched successfully, so return true
        return true;
    }

全量拉取条件（任意一个）
①、disable-delta属性值是true 关闭增量拉取
②、registry-refresh-single-vip-address 属性vip地址的值不为空
③、forceFullRegistryFetch 为true 传过来的变量值
④、localRegionApps的applications是null 当前区域应用
⑤、applications的数量是0
⑥、applications的版本是-1

增量拉取

实现增量拉取的条件是不符合全量拉取，调用getAndUpdateDelta方法
com.netflix.discovery.DiscoveryClient#getAndUpdateDelta

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());
        }
    }

这个方法实现了增量拉取的请求实现，及对拉取增量结果的处理
1、getDelta
eurekaTransport.queryClient.getDelta(remoteRegionsRef.get())的具体实现是通过AbstractJerseyEurekaHttpClient类实现的
com.netflix.discovery.shared.transport.jersey.AbstractJerseyEurekaHttpClient#getDelta

getDelta.png

拼接apps/delta开头的请求
2、getApplicationsInternal
com.netflix.discovery.shared.transport.jersey.AbstractJerseyEurekaHttpClient#getApplicationsInternal

image.png

拼接的http请求是：http://localhost:8000/eureka/apps/delta然后调用到Eureka Server服务端
3、getContainerDifferential
服务端通过getContainerDifferential接收
com.netflix.eureka.resources.ApplicationsResource#getContainerDifferential

getContainerDifferential1.png

getContainerDifferential2.png

无论是调用到responseCache.getGZIP(cacheKey)方法，还是responseCache.get(cacheKey)方法。最终都是调用到Eureka Server的查询的三种缓存，最终会经过读写缓存readWriteCacheMap处理。
4、readWriteCacheMap
调用到ResponseCacheImpl构造方法中readWriteCacheMap读写缓存的创建中，如果缓存中没有会调用generatePayload方法
com.netflix.eureka.registry.ResponseCacheImpl#readWriteCacheMap

image.png

5、generatePayload
com.netflix.eureka.registry.ResponseCacheImpl#generatePayload
根据传入的Key参数处理不同的情况

generatePayload.png

key = { eurekaAccept = "full"，entityType = "Application"，hashKey = "ApplicationALL_APPS_DELTAJSONV2full"， requestVersion = "V2"， requestType = "JSON"， regions = null， entityName = "ALL_APPS_DELTA"}
entityType 可分为Application、VIP、SVIP、default等情况
在entityType 是Application情况下EntityName分为："ALL_APPS"（全量拉取）和"ALL_APPS_DELTA"（增量拉取）
在全量或增量拉取下有isRemoteRegionRequested参数判断是否具有远程区域请求
本次处理是：增量拉取的没有远程区域请求，并调用getPayLoad方法加载是否含有增量数据
6、getApplicationDeltas
获取增量实例
com.netflix.eureka.registry.AbstractInstanceRegistry#getApplicationDeltas

public Applications getApplicationDeltas() {
        GET_ALL_CACHE_MISS_DELTA.increment();
        Applications apps = new Applications();
        apps.setVersion(responseCache.getVersionDelta().get());
        Map<String, Application> applicationInstancesMap = new HashMap<String, Application>();
        try {
            write.lock();
            Iterator<RecentlyChangedItem> iter = this.recentlyChangedQueue.iterator();
            logger.debug("The number of elements in the delta queue is : {}",
                    this.recentlyChangedQueue.size());
            while (iter.hasNext()) {
                Lease<InstanceInfo> lease = iter.next().getLeaseInfo();
                InstanceInfo instanceInfo = lease.getHolder();
                logger.debug(
                        "The instance id {} is found with status {} and actiontype {}",
                        instanceInfo.getId(), instanceInfo.getStatus().name(), instanceInfo.getActionType().name());
                Application app = applicationInstancesMap.get(instanceInfo
                        .getAppName());
                if (app == null) {
                    app = new Application(instanceInfo.getAppName());
                    applicationInstancesMap.put(instanceInfo.getAppName(), app);
                    apps.addApplication(app);
                }
                app.addInstance(new InstanceInfo(decorateInstanceInfo(lease)));
            }

            boolean disableTransparentFallback = serverConfig.disableTransparentFallbackToOtherRegion();

            if (!disableTransparentFallback) {
                Applications allAppsInLocalRegion = getApplications(false);

                for (RemoteRegionRegistry remoteRegistry : this.regionNameVSRemoteRegistry.values()) {
                    Applications applications = remoteRegistry.getApplicationDeltas();
                    for (Application application : applications.getRegisteredApplications()) {
                        Application appInLocalRegistry =
                                allAppsInLocalRegion.getRegisteredApplications(application.getName());
                        if (appInLocalRegistry == null) {
                            apps.addApplication(application);
                        }
                    }
                }
            }

            Applications allApps = getApplications(!disableTransparentFallback);
            apps.setAppsHashCode(allApps.getReconcileHashCode());
            return apps;
        } finally {
            write.unlock();
        }
    }

这个方法主要是遍历recentlyChangedQueue存在的数据放入到Applications对象中。所以recentlyChangedQueue队列中存在什么数据就很重要，因此我们需要了解最新更新队列recentlyChangedQueue是如何放入的及放入那些数据，及其的移除的原理。
在这个方法最后 apps.setAppsHashCode设置了当前服务端所有注册信息的HashCode，所以这个增量对象存储了最新的状态HashCode值。
7、客户端获取增量数据的处理
还是在getAndUpdateDelta方法内，对服务端传输过来数据，获取当前服务端的增量数据部分
com.netflix.discovery.DiscoveryClient#getAndUpdateDelta

getAndUpdateDelta.png

这个方法的主要过程是：
如果增量数据部分为空，则执行全量拉取。
对当前服务的注册信息表执行updateDelta(delta)方法，对当前注册实例的增加删除或修改操作
当前更新后的服务注册表的HashCode值与增量对象存储的最新的状态HashCode值比较，如果不相等则执行全量拉取

recentlyChangedQueue

最新更新队列ConcurrentLinkedQueue<RecentlyChangedItem> recentlyChangedQueue
com.netflix.eureka.registry.AbstractInstanceRegistry#AbstractInstanceRegistry类在构建创建注册表时创建了recentlyChangedQueue队列，并创建了一个增量调度任务方法getDeltaRetentionTask方法

AbstractInstanceRegistry.png

com.netflix.eureka.registry.AbstractInstanceRegistry#getDeltaRetentionTask
对recentlyChangedQueue队列中对最近改变的队列在一定时间范围retentionTimeInMSInDeltaQueue=180000ms(3分钟)外的进行定时清除(30s清除一次)

getDeltaRetentionTask.png

recentlyChangedQueue队列添加条件：
1、注册时register
2、下线时Cancel

3、statusUpdate
4、deleteStatusOverride
getDeltaRetentionTask进行定时清除

全量拉取

全量拉取与增量拉取过程类似
全量拉取调用getAndStoreFullRegistry方法
1、getAndStoreFullRegistry
com.netflix.discovery.DiscoveryClient#getAndStoreFullRegistry

getAndStoreFullRegistry.png

2、getApplications
com.netflix.discovery.shared.transport.EurekaHttpClient#getApplications

EurekaHttpClient.png

3、getApplications
com.netflix.discovery.shared.transport.jersey.AbstractJerseyEurekaHttpClient#getApplications

getApplications.png

拼接的Http请求是:apps/
4、getApplicationsInternal

com.netflix.discovery.shared.transport.jersey.AbstractJerseyEurekaHttpClient#getApplicationsInternal

getApplicationsInternal.png

5、getContainers
调用到Eureka Server服务端的getContainers方法，这里调用与增量拉取类似。查询经过三层缓存结构，具体过程可参考增量查询与三层缓存结构。
com.netflix.eureka.resources.ApplicationsResource#getContainers

getContainers.png

如果acceptEncoding包含值gzip将会调用ResponseCacheImpl#getGZIP方法获取

ResponseCacheImpl.png

否则调用ResponseCacheImpl#get(Key)方法
最终会调用到读写缓存的generatePayload方法处理
6、generatePayload
com.netflix.eureka.registry.ResponseCacheImpl#generatePayload
Key结构中的EntityName是"ALL_APPS"全量查询。

generatePayload.png

7、getApplications
获取注册表所有的注册信息

com.netflix.eureka.registry.AbstractInstanceRegistry#getApplications()

getApplications.png

8、结果处理
放入到当前对象localRegionApps缓存中

image.png

如果含有获取远程区域注册信息FetchingRemoteRegionRegistries，只需要分到不同的索引位置

image.png

总结：

Eureka Client客户端与Eureka Server服务端是关联的，不能分开处理。关键点有以下几点：
客户端主要了解的是如何向服务端发起请求及服务端接收接口的对应
客户端的定时任务有哪些：心跳及缓存获取
客户端注册的实现，缓存获取原理
缓存获取中的增量拉取与全量拉取的区别
增量拉取在服务端的实现原理，由一个状态更改队列实现的