Spring Cloud Eureka源码解析之服务注册列表拉取（五）

红太狼 2022-09-16 04:56 227阅读 0赞

在Eureka Client启动流程一文讲到Client刚启动时会全量拉取一次，后面会开启一个定时任务，根据条件以及本地缓存情况全量拉取或增量拉取

private boolean fetchRegistry(boolean forceFullRegistryFetch) { 
        Stopwatch tracer = FETCH_REGISTRY_TIMER.start();
    
        try { 
            // 如果增量被禁用或者是第一次获取（本地缓存为空），则全量获取
            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();
    
        // 更新状态
        updateInstanceRemoteStatus();
    
        // registry was fetched successfully, so return true
        return true;
    }

## 全量拉取 ##

**EurekaClient**

private void getAndStoreFullRegistry() throws Throwable { 
        long currentUpdateGeneration = fetchRegistryGeneration.get();
    
        logger.info("Getting all instance registry info from the eureka server");
    
        Applications apps = null;
        // 发起拉取请求
        // 调用EurekaHttpClient接口，EurekaHttpClient是客户端与服务端通信的接口，通过Jersey框架完成。（Jersey是一个REST框架）
        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());
    
        if (apps == null) { 
            logger.error("The application is null for some reason. Not storing this information");
        } else if (fetchRegistryGeneration.compareAndSet(currentUpdateGeneration, currentUpdateGeneration + 1)) { 
            // 由于是全量拉取，是完整的注册信息，所以直接将本地缓存替换成拉取到的信息
            localRegionApps.set(this.filterAndShuffle(apps));
            logger.debug("Got full registry with apps hashcode {}", apps.getAppsHashCode());
        } else { 
            logger.warn("Not updating applications as another thread is updating it already");
        }
    }

这里localRegionApps就是client中的本地缓存，是一个引用类型

private final AtomicReference<Applications> localRegionApps = new AtomicReference<Applications>();

**EurekaServer**  
服务端controller接口是ApplicationsResource类的getContainers方法

@GET
    public Response getContainers(@PathParam("version") String version,
                                  @HeaderParam(HEADER_ACCEPT) String acceptHeader,
                                  @HeaderParam(HEADER_ACCEPT_ENCODING) String acceptEncoding,
                                  @HeaderParam(EurekaAccept.HTTP_X_EUREKA_ACCEPT) String eurekaAccept,
                                  @Context UriInfo uriInfo,
                                  @Nullable @QueryParam("regions") String regionsStr) { 
    
        boolean isRemoteRegionRequested = null != regionsStr && !regionsStr.isEmpty();
        String[] regions = null;
        if (!isRemoteRegionRequested) { 
            EurekaMonitors.GET_ALL.increment();
        } else { 
            regions = regionsStr.toLowerCase().split(",");
            Arrays.sort(regions); // So we don't have different caches for same regions queried in different order.
            EurekaMonitors.GET_ALL_WITH_REMOTE_REGIONS.increment();
        }
    
        // Check if the server allows the access to the registry. The server can
        // restrict access if it is not
        // ready to serve traffic depending on various reasons.
        if (!registry.shouldAllowAccess(isRemoteRegionRequested)) { 
            return Response.status(Status.FORBIDDEN).build();
        }
        CurrentRequestVersion.set(Version.toEnum(version));
        KeyType keyType = Key.KeyType.JSON;
        String returnMediaType = MediaType.APPLICATION_JSON;
        if (acceptHeader == null || !acceptHeader.contains(HEADER_JSON_VALUE)) { 
            keyType = Key.KeyType.XML;
            returnMediaType = MediaType.APPLICATION_XML;
        }
    
        Key cacheKey = new Key(Key.EntityType.Application,
                               ResponseCacheImpl.ALL_APPS,
                               keyType, CurrentRequestVersion.get(), EurekaAccept.fromString(eurekaAccept), regions
                              );
    
        Response response;
        if (acceptEncoding != null && acceptEncoding.contains(HEADER_GZIP_VALUE)) { 
            response = Response.ok(responseCache.getGZIP(cacheKey))
                .header(HEADER_CONTENT_ENCODING, HEADER_GZIP_VALUE)
                .header(HEADER_CONTENT_TYPE, returnMediaType)
                .build();
        } else { 
            response = Response.ok(responseCache.get(cacheKey))
                .build();
        }
        return response;
    }

关键就是最下面几行代码  
![在这里插入图片描述][00e64325bb97f2d10291f4bdab9b8b7a.png_pic_center]  
先创建一个缓存键  
![在这里插入图片描述][a272d6a4918a00d4fe236be84e71c02e.png_pic_center]  
这几个参数都是固定的  
![在这里插入图片描述][8fc29bb8316ce4ed1cd011949b0bfc27.png_pic_center]  
Key内部有一个hashKey，根据创建时传进去的参数生成的  
根据Key从响应缓存中读取数据，ResponseCacheImpl是ResponseCache的实现，负责缓存将由客户端查询的注册表信息的类  
![在这里插入图片描述][94448baba66eb43c31ed3c6a461e8896.png_pic_center]  
shouldUseReadOnlyResponseCache表示是否从只读缓存读取，默认是true

@VisibleForTesting
    Value getValue(final Key key, boolean useReadOnlyCache) { 
        Value payload = null;
        try { 
            // 默认从只读缓存读取
            if (useReadOnlyCache) { 
                final Value currentPayload = readOnlyCacheMap.get(key);
                if (currentPayload != null) { 
                    payload = currentPayload;
                } else { 
                    // 只读缓存没读取到就到读写缓存读取，然后设置到只读缓存中
                    payload = readWriteCacheMap.get(key);
                    readOnlyCacheMap.put(key, payload);
                }
            } else { 
                payload = readWriteCacheMap.get(key);
            }
        } catch (Throwable t) { 
            logger.error("Cannot get value for key : {}", key, t);
        }
        return payload;
    }

先从readOnlyCacheMap只读缓存，没读到再去readWriteCacheMap读写缓存读取，然后设置进readOnlyCacheMap。**注意：如果读写缓存中该key不存在值，将调用CacheLoader的load方法加载新值到该key中**  
readOnlyCacheMap是使用java的ConcurrentMap，而readWriteCacheMap读写缓存使用了guava中的LoadingCache  
![在这里插入图片描述][26f2947e680ef074a625ff2d76406872.png_pic_center]  
readWriteCacheMap是在ResponseCacheImpl构造方法中构建的

ResponseCacheImpl(EurekaServerConfig serverConfig, ServerCodecs serverCodecs, AbstractInstanceRegistry registry) { 
        this.serverConfig = serverConfig;
        this.serverCodecs = serverCodecs;
        this.shouldUseReadOnlyResponseCache = serverConfig.shouldUseReadOnlyResponseCache();
        this.registry = registry;
    
        long responseCacheUpdateIntervalMs = serverConfig.getResponseCacheUpdateIntervalMs();
        this.readWriteCacheMap =
            // 初始大小1000
            CacheBuilder.newBuilder().initialCapacity(1000)
            // 默认写入后180s后过期
            .expireAfterWrite(serverConfig.getResponseCacheAutoExpirationInSeconds(), TimeUnit.SECONDS)
            // 移除监听器
            .removalListener(new RemovalListener<Key, Value>() { 
                @Override
                public void onRemoval(RemovalNotification<Key, Value> notification) { 
                    Key removedKey = notification.getKey();
                    if (removedKey.hasRegions()) { 
                        Key cloneWithNoRegions = removedKey.cloneWithoutRegions();
                        regionSpecificKeys.remove(cloneWithNoRegions, removedKey);
                    }
                }
            })
            .build(new CacheLoader<Key, Value>() { 
                // 获取key对应的value没值时执行的方法逻辑，加载值到cache中
                @Override
                public Value load(Key key) throws Exception { 
                    if (key.hasRegions()) { 
                        Key cloneWithNoRegions = key.cloneWithoutRegions();
                        regionSpecificKeys.put(cloneWithNoRegions, key);
                    }
                    // 去加载key对应的value到读写缓存中
                    Value value = generatePayload(key);
                    return value;
                }
            });
    
        // 每隔30s从读写缓存同步到只读缓存
        if (shouldUseReadOnlyResponseCache) { 
            timer.schedule(getCacheUpdateTask(),
                           new Date(((System.currentTimeMillis() / responseCacheUpdateIntervalMs) * responseCacheUpdateIntervalMs)
                                    + responseCacheUpdateIntervalMs),
                           responseCacheUpdateIntervalMs);
        }
    
        try { 
            Monitors.registerObject(this);
        } catch (Throwable e) { 
            logger.warn("Cannot register the JMX monitor for the InstanceRegistry", e);
        }
    }

![在这里插入图片描述][a32ce75730635c2a6545ea6d24391b55.png_pic_center]  
所以如果上面从读写缓存中也没获取到，会调用generatePayload(key)方法从registry缓存加载

private Value generatePayload(Key key) { 
        Stopwatch tracer = null;
        try { 
            String payload;
            // 获取注册表的类型，全量、增量或只获取某个应用的注册信息
            switch (key.getEntityType()) { 
                case Application:
                    boolean isRemoteRegionRequested = key.hasRegions();
    
                    if (ALL_APPS.equals(key.getName())) {  // 全量获取
                        if (isRemoteRegionRequested) { 
                            tracer = serializeAllAppsWithRemoteRegionTimer.start();
                            payload = getPayLoad(key, registry.getApplicationsFromMultipleRegions(key.getRegions()));
                        } else { 
                            tracer = serializeAllAppsTimer.start();
                            payload = getPayLoad(key, registry.getApplications());
                        }
                    } else if (ALL_APPS_DELTA.equals(key.getName())) {  // 增量获取
                        if (isRemoteRegionRequested) { 
                            tracer = serializeDeltaAppsWithRemoteRegionTimer.start();
                            versionDeltaWithRegions.incrementAndGet();
                            versionDeltaWithRegionsLegacy.incrementAndGet();
                            payload = getPayLoad(key,
                                                 registry.getApplicationDeltasFromMultipleRegions(key.getRegions()));
                        } else { 
                            tracer = serializeDeltaAppsTimer.start();
                            versionDelta.incrementAndGet();
                            versionDeltaLegacy.incrementAndGet();
                            payload = getPayLoad(key, registry.getApplicationDeltas());
                        }
                    } else {  // 只获取某个应用的注册信息
                        tracer = serializeOneApptimer.start();
                        payload = getPayLoad(key, registry.getApplication(key.getName()));
                    }
                    break;
                case VIP:
                case SVIP:
                    tracer = serializeViptimer.start();
                    payload = getPayLoad(key, getApplicationsForVip(key, registry));
                    break;
                default:
                    logger.error("Unidentified entity type: {} found in the cache key.", key.getEntityType());
                    payload = "";
                    break;
            }
            return new Value(payload);
        } finally { 
            if (tracer != null) { 
                tracer.stop();
            }
        }
    }

从AbstractInstanceRegistry中获取所有应用实例，AbstractInstanceRegistry缓存了所有的实例租约信息，实时更新，key是应用名，value就是每个应用对应的实例租约信息列表  
![在这里插入图片描述][279c425396bd43ca1608bf6ebfcb2c2a.png_pic_center]  
其中InstanceInfo就是代表着EurekaClient实例，客户端向服务端请求注册时就会携带着这个实例，这个实例中包含了客户端的一些基本信息，Lease是服务端用来保存客户端的租约信息的，其中holder中持有的就是InstanceInfo对象

public class Lease<T> { 
    
        enum Action { 
            Register, Cancel, Renew
        };
    
        // 租约过期时间
        public static final int DEFAULT_DURATION_IN_SECS = 90;
    
        // 持有的实例
        private T holder;
        // 服务下线时间
        private long evictionTimestamp;
        // 注册起始时间
        private long registrationTimestamp;
        private long serviceUpTimestamp;
        // Make it volatile so that the expiration task would see this quicker
        // 最后更新时间
        private volatile long lastUpdateTimestamp;
        // 租约持续时间
        private long duration;
    }

接着上面的方法，走全量拉取分支中的registry.getApplications()  
![在这里插入图片描述][8d26227bbbd3d35911d8e15475ea9906.png_pic_center]  
![在这里插入图片描述][382c947e28eae1173e03e76b5980e788.png_pic_center]

public Applications getApplicationsFromMultipleRegions(String[] remoteRegions) { 
    
        boolean includeRemoteRegion = null != remoteRegions && remoteRegions.length != 0;
    
        logger.debug("Fetching applications registry with remote regions: {}, Regions argument {}",
                     includeRemoteRegion, remoteRegions);
    
        if (includeRemoteRegion) { 
            GET_ALL_WITH_REMOTE_REGIONS_CACHE_MISS.increment();
        } else { 
            GET_ALL_CACHE_MISS.increment();
        }
        Applications apps = new Applications();
        apps.setVersion(1L);
        // 循环所有应用对应的租约信息
        for (Entry<String, Map<String, Lease<InstanceInfo>>> entry : registry.entrySet()) { 
            Application app = null;
    
            if (entry.getValue() != null) { 
                // 将相同应用下的服务实例封装为Application
                for (Entry<String, Lease<InstanceInfo>> stringLeaseEntry : entry.getValue().entrySet()) { 
                    Lease<InstanceInfo> lease = stringLeaseEntry.getValue();
                    if (app == null) { 
                        app = new Application(lease.getHolder().getAppName());
                    }
                    app.addInstance(decorateInstanceInfo(lease));
                }
            }
            // 将每个Application添加到Applications中
            if (app != null) { 
                apps.addApplication(app);
            }
        }
        if (includeRemoteRegion) { 
            // 从远程获取。。。
        }
        // 设置hashcode为所有注册表信息计算出来的hashcode值，主要给增量拉取使用，在全量这边没什么用
        apps.setAppsHashCode(apps.getReconcileHashCode());
        return apps;
    }

在ResponseCacheImpl构造方法的最后eureka启动了一个定时任务，每隔30s从读写缓存更新数据到只读缓存  
![在这里插入图片描述][1b79117f0f7fdecba21a0b8a256a0fb5.png_pic_center]

private TimerTask getCacheUpdateTask() { 
        return new TimerTask() { 
            @Override
            public void run() { 
                logger.debug("Updating the client cache from response cache");
                // 循环只读缓存，与读写缓存做比较，不相同则更新
                for (Key key : readOnlyCacheMap.keySet()) { 
                    if (logger.isDebugEnabled()) { 
                        logger.debug("Updating the client cache from response cache for key : {} {} {} {}",
                                     key.getEntityType(), key.getName(), key.getVersion(), key.getType());
                    }
                    try { 
                        CurrentRequestVersion.set(key.getVersion());
                        Value cacheValue = readWriteCacheMap.get(key);
                        Value currentCacheValue = readOnlyCacheMap.get(key);
                        if (cacheValue != currentCacheValue) { 
                            readOnlyCacheMap.put(key, cacheValue);
                        }
                    } catch (Throwable th) { 
                        logger.error("Error while updating the client cache from response cache for key {}", key.toStringCompact(), th);
                    }
                }
            }
        };
    }

## 增量拉取 ##

**EurekaClient**  
Eureka Client启动时会创建一个每隔30s刷新本地缓存的定时任务  
![在这里插入图片描述][00a95ab92efe0e8c13cd6e02b5d387cd.png_pic_center]

@VisibleForTesting
    void refreshRegistry() { 
        try { 
            boolean isFetchingRemoteRegionRegistries = isFetchingRemoteRegionRegistries();
    
            boolean remoteRegionsModified = false;
            // 省略。。。
    
            // 增量拉取
            boolean success = fetchRegistry(remoteRegionsModified);
            if (success) { 
                registrySize = localRegionApps.get().size();
                lastSuccessfulRegistryFetchTimestamp = System.currentTimeMillis();
            }
    
            if (logger.isDebugEnabled()) { 
                StringBuilder allAppsHashCodes = new StringBuilder();
                allAppsHashCodes.append("Local region apps hashcode: ");
                allAppsHashCodes.append(localRegionApps.get().getAppsHashCode());
                allAppsHashCodes.append(", is fetching remote regions? ");
                allAppsHashCodes.append(isFetchingRemoteRegionRegistries);
                for (Map.Entry<String, Applications> entry : remoteRegionVsApps.entrySet()) { 
                    allAppsHashCodes.append(", Remote region: ");
                    allAppsHashCodes.append(entry.getKey());
                    allAppsHashCodes.append(" , apps hashcode: ");
                    allAppsHashCodes.append(entry.getValue().getAppsHashCode());
                }
                logger.debug("Completed cache refresh task for discovery. All Apps hash code is {} ",
                             allAppsHashCodes);
            }
        } catch (Throwable e) { 
            logger.error("Cannot fetch registry from server", e);
        }        
    }

![在这里插入图片描述][240958f0bc8c07520d0fe325e8ed8cac.png_pic_center]  
![在这里插入图片描述][caf5dd6c6eafe8d6362ad0467798c623.png_pic_center]  
接下来看服务端  
**EurekaServer**

@Path("delta")
    @GET
    public Response getContainerDifferential(
        @PathParam("version") String version,
        @HeaderParam(HEADER_ACCEPT) String acceptHeader,
        @HeaderParam(HEADER_ACCEPT_ENCODING) String acceptEncoding,
        @HeaderParam(EurekaAccept.HTTP_X_EUREKA_ACCEPT) String eurekaAccept,
        @Context UriInfo uriInfo, @Nullable @QueryParam("regions") String regionsStr) { 
    
        boolean isRemoteRegionRequested = null != regionsStr && !regionsStr.isEmpty();
    
        // If the delta flag is disabled in discovery or if the lease expiration
        // has been disabled, redirect clients to get all instances
        if ((serverConfig.shouldDisableDelta()) || (!registry.shouldAllowAccess(isRemoteRegionRequested))) { 
            return Response.status(Status.FORBIDDEN).build();
        }
    
        String[] regions = null;
        if (!isRemoteRegionRequested) { 
            EurekaMonitors.GET_ALL_DELTA.increment();
        } else { 
            regions = regionsStr.toLowerCase().split(",");
            Arrays.sort(regions); // So we don't have different caches for same regions queried in different order.
            EurekaMonitors.GET_ALL_DELTA_WITH_REMOTE_REGIONS.increment();
        }
    
        CurrentRequestVersion.set(Version.toEnum(version));
        KeyType keyType = Key.KeyType.JSON;
        String returnMediaType = MediaType.APPLICATION_JSON;
        if (acceptHeader == null || !acceptHeader.contains(HEADER_JSON_VALUE)) { 
            keyType = Key.KeyType.XML;
            returnMediaType = MediaType.APPLICATION_XML;
        }
    
        // 这里跟全量拉取的区别只是这里EntityType传的是ResponseCacheImpl.ALL_APPS_DELTA，
        // 代表增量拉取，其它参数都一样
        Key cacheKey = new Key(Key.EntityType.Application,
                               ResponseCacheImpl.ALL_APPS_DELTA,
                               keyType, CurrentRequestVersion.get(), EurekaAccept.fromString(eurekaAccept), regions
                              );
    
        if (acceptEncoding != null
            && acceptEncoding.contains(HEADER_GZIP_VALUE)) { 
            return Response.ok(responseCache.getGZIP(cacheKey))
                .header(HEADER_CONTENT_ENCODING, HEADER_GZIP_VALUE)
                .header(HEADER_CONTENT_TYPE, returnMediaType)
                .build();
        } else { 
            return Response.ok(responseCache.get(cacheKey))
                .build();
        }
    }

跟全量拉取一样，也是先创建该实例的Key，Key中的EntityType是ResponseCacheImpl.ALL\_APPS\_DELTA，代表是增量拉取，根据Key从响应缓存responseCache中获取，接下来的流程跟全量拉取一样，也是从只读缓存到读写缓存，只是如果读写缓存没有的话，会去最近变更记录队列recentlyChangedQueue读取加载  
**ResponseCacheImpl.generatePayload**  
![在这里插入图片描述][c1d799d63cbca014b8f13d95e8373415.png_pic_center]

public Applications getApplicationDeltas() { 
        GET_ALL_CACHE_MISS_DELTA.increment();
        // 返回的应用，会将最近变更的实例设置到该apps中
        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(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);
                        }
                    }
                }
            }
    
            // 设置hashcode为所有注册表信息计算出来的hashcode值
            // 用来给client端判断增量更新本地缓存后的注册表信息是否跟server端的注册表信息是否一致，
            // 若不一致，则需要重新全量拉取一次
            Applications allApps = getApplications(!disableTransparentFallback);
            apps.setAppsHashCode(allApps.getReconcileHashCode());
            return apps;
        } finally { 
            write.unlock();
        }
    }

**处理增量拉取结果**  
![在这里插入图片描述][7907613b18b6038b7d9f6e94484046a4.png_pic_center]  
更新从EurekaServer获取到的增量信息到本地缓存

private void updateDelta(Applications delta) { 
        int deltaCount = 0;
        for (Application app : delta.getRegisteredApplications()) { 
            for (InstanceInfo instance : app.getInstances()) { 
                // 本地缓存
                Applications applications = getApplications();
                String instanceRegion = instanceRegionChecker.getInstanceRegion(instance);
                if (!instanceRegionChecker.isLocalRegion(instanceRegion)) { 
                    Applications remoteApps = remoteRegionVsApps.get(instanceRegion);
                    if (null == remoteApps) { 
                        remoteApps = new Applications();
                        remoteRegionVsApps.put(instanceRegion, remoteApps);
                    }
                    applications = remoteApps;
                }
    
                // 记录获取到的全部实例数
                ++deltaCount;
                // 新注册的服务
                if (ActionType.ADDED.equals(instance.getActionType())) { 
                    Application existingApp = applications.getRegisteredApplications(instance.getAppName());
                    if (existingApp == null) { 
                        applications.addApplication(app);
                    }
                    logger.debug("Added instance {} to the existing apps in region {}", instance.getId(), instanceRegion);
                    applications.getRegisteredApplications(instance.getAppName()).addInstance(instance);
                } else if (ActionType.MODIFIED.equals(instance.getActionType())) {  // 修改的服务
                    Application existingApp = applications.getRegisteredApplications(instance.getAppName());
                    if (existingApp == null) { 
                        applications.addApplication(app);
                    }
                    logger.debug("Modified instance {} to the existing apps ", instance.getId());
    
                    applications.getRegisteredApplications(instance.getAppName()).addInstance(instance);
    
                } else if (ActionType.DELETED.equals(instance.getActionType())) {  // 已删除的服务
                    Application existingApp = applications.getRegisteredApplications(instance.getAppName());
                    if (existingApp == null) { 
                        applications.addApplication(app);
                    }
                    logger.debug("Deleted instance {} to the existing apps ", instance.getId());
                    applications.getRegisteredApplications(instance.getAppName()).removeInstance(instance);
                }
            }
        }
        logger.debug("The total number of instances fetched by the delta processor : {}", deltaCount);
    
        getApplications().setVersion(delta.getVersion());
        getApplications().shuffleInstances(clientConfig.shouldFilterOnlyUpInstances());
    
        for (Applications applications : remoteRegionVsApps.values()) { 
            applications.setVersion(delta.getVersion());
            applications.shuffleInstances(clientConfig.shouldFilterOnlyUpInstances());
        }
    }

最后比对本地所有的注册表数据与server端所有注册数据的hash值，如果哈希值不一致，则通过这个reconcileAndLogDifference()来进行一次全量同步。因为Eureka client的增量更新，其实获取的是Eureka server最近三分钟内的变更，如果Eureka client有超过三分钟没有做增量更新的话（例如网络问题），这就造成了Eureka server和Eureka client之间的数据不一致。正常情况下，Eureka client每次增量更新后，最终的服务列表数据应该跟Eureka server保持一致，但如果期间发生异常，可能导致和Eureka server的数据不一致，为了暴露这个问题，Eureka server每次返回的增量更新数据中，会带有一致性哈希码，Eureka client用本地服务列表数据算出的一致性哈希码应该和Eureka server返回的一致，若不一致就证明增量更新出了问题导致Eureka client和Eureka server上的服务列表信息不一致了，此时需要全量更新。  
![在这里插入图片描述][88eb0bea6ff143d48a99a4316056692f.png_pic_center]  
全量拉取到后就直接赋值给本地缓存  
![在这里插入图片描述][562b03d828c7404c2185e2db9f5ebaaf.png_pic_center]

## 参考资料 ##

[SpringCloud 源码系列（2）—— 注册中心 Eureka（中）][SpringCloud _2_ _ Eureka]

[00e64325bb97f2d10291f4bdab9b8b7a.png_pic_center]: /images/20220828/4ef559dd518848c48234b3e1ba497486.png
[a272d6a4918a00d4fe236be84e71c02e.png_pic_center]: /images/20220828/8784fc22c6dc42a1ae47ee1815471f15.png
[8fc29bb8316ce4ed1cd011949b0bfc27.png_pic_center]: /images/20220828/697b05e717714bffa0e5ee6ecaeaa888.png
[94448baba66eb43c31ed3c6a461e8896.png_pic_center]: /images/20220828/618b62821c2a448bba38114d3eebf911.png
[26f2947e680ef074a625ff2d76406872.png_pic_center]: /images/20220828/ccbe39dd6a654eb7aa59151b870ad17a.png
[a32ce75730635c2a6545ea6d24391b55.png_pic_center]: /images/20220828/80e25dde73d54f4aa6765a017d631741.png
[279c425396bd43ca1608bf6ebfcb2c2a.png_pic_center]: /images/20220828/662b5c345de24cdebb1561d1dc1e9d4c.png
[8d26227bbbd3d35911d8e15475ea9906.png_pic_center]: /images/20220828/8c821bd6ff2143529784ce197f2e8ede.png
[382c947e28eae1173e03e76b5980e788.png_pic_center]: /images/20220828/5df8c44e091443c5bb93db4b631a27ff.png
[1b79117f0f7fdecba21a0b8a256a0fb5.png_pic_center]: /images/20220828/e6506fe5494242b0a69e1bc1436a009a.png
[00a95ab92efe0e8c13cd6e02b5d387cd.png_pic_center]: /images/20220828/e6af792b37b441c8876918241f5acb8d.png
[240958f0bc8c07520d0fe325e8ed8cac.png_pic_center]: /images/20220828/333e7b6657054d16a5e0f86e30fdd482.png
[caf5dd6c6eafe8d6362ad0467798c623.png_pic_center]: /images/20220828/f823c1ba1d3f47b6989f0208eec1a24b.png
[c1d799d63cbca014b8f13d95e8373415.png_pic_center]: /images/20220828/4b5f71581de447bbb4a2a27d8a98d27f.png
[7907613b18b6038b7d9f6e94484046a4.png_pic_center]: /images/20220828/ddc5a0d962e84a418adc64e22fbef424.png
[88eb0bea6ff143d48a99a4316056692f.png_pic_center]: /images/20220828/9869535c05104b1387306bccd1596e98.png
[562b03d828c7404c2185e2db9f5ebaaf.png_pic_center]: /images/20220828/4b2b1cadbe05477ab9951289b82e4662.png
[SpringCloud _2_ _ Eureka]: https://www.cnblogs.com/chiangchou/p/eureka-2.html