##序 eureka server定时任务去判断lease是否过期，然后设置lease对应的instance的状态，然后更新到ResponseCache，供客户端来获取刷新，这里讲下client端的刷新过程中是如何剔除实例的。
##client更新task eureka-client-1.4.12-sources.jar!/com/netflix/discovery/DiscoveryClient.java
1if (clientConfig.shouldFetchRegistry()) { 2 // registry cache refresh timer 3 int registryFetchIntervalSeconds = clientConfig.getRegistryFetchIntervalSeconds(); 4 int expBackOffBound = clientConfig.getCacheRefreshExecutorExponentialBackOffBound(); 5 scheduler.schedule( 6 new TimedSupervisorTask( 7 "cacheRefresh", 8 scheduler, 9 cacheRefreshExecutor, 10 registryFetchIntervalSeconds, 11 TimeUnit.SECONDS, 12 expBackOffBound, 13 new CacheRefreshThread() 14 ), 15 registryFetchIntervalSeconds, TimeUnit.SECONDS); 16 }
##CacheRefreshThread
1/** 2 * The task that fetches the registry information at specified intervals. 3 * 4 */ 5 class CacheRefreshThread implements Runnable { 6 public void run() { 7 refreshRegistry(); 8 } 9 }
##refreshRegistry
1@VisibleForTesting 2 void refreshRegistry() { 3 try { 4 boolean isFetchingRemoteRegionRegistries = isFetchingRemoteRegionRegistries(); 5 6 boolean remoteRegionsModified = false; 7 // This makes sure that a dynamic change to remote regions to fetch is honored. 8 String latestRemoteRegions = clientConfig.fetchRegistryForRemoteRegions(); 9 if (null != latestRemoteRegions) { 10 String currentRemoteRegions = remoteRegionsToFetch.get(); 11 if (!latestRemoteRegions.equals(currentRemoteRegions)) { 12 // Both remoteRegionsToFetch and AzToRegionMapper.regionsToFetch need to be in sync 13 synchronized (instanceRegionChecker.getAzToRegionMapper()) { 14 if (remoteRegionsToFetch.compareAndSet(currentRemoteRegions, latestRemoteRegions)) { 15 String[] remoteRegions = latestRemoteRegions.split(","); 16 remoteRegionsRef.set(remoteRegions); 17 instanceRegionChecker.getAzToRegionMapper().setRegionsToFetch(remoteRegions); 18 remoteRegionsModified = true; 19 } else { 20 logger.info("Remote regions to fetch modified concurrently," + 21 " ignoring change from {} to {}", currentRemoteRegions, latestRemoteRegions); 22 } 23 } 24 } else { 25 // Just refresh mapping to reflect any DNS/Property change 26 instanceRegionChecker.getAzToRegionMapper().refreshMapping(); 27 } 28 } 29 30 boolean success = fetchRegistry(remoteRegionsModified); 31 if (success) { 32 registrySize = localRegionApps.get().size(); 33 lastSuccessfulRegistryFetchTimestamp = System.currentTimeMillis(); 34 } 35 36 if (logger.isDebugEnabled()) { 37 StringBuilder allAppsHashCodes = new StringBuilder(); 38 allAppsHashCodes.append("Local region apps hashcode: "); 39 allAppsHashCodes.append(localRegionApps.get().getAppsHashCode()); 40 allAppsHashCodes.append(", is fetching remote regions? "); 41 allAppsHashCodes.append(isFetchingRemoteRegionRegistries); 42 for (Map.Entry<String, Applications> entry : remoteRegionVsApps.entrySet()) { 43 allAppsHashCodes.append(", Remote region: "); 44 allAppsHashCodes.append(entry.getKey()); 45 allAppsHashCodes.append(" , apps hashcode: "); 46 allAppsHashCodes.append(entry.getValue().getAppsHashCode()); 47 } 48 logger.debug("Completed cache refresh task for discovery. All Apps hash code is {} ", 49 allAppsHashCodes.toString()); 50 } 51 } catch (Throwable e) { 52 logger.error("Cannot fetch registry from server", e); 53 } 54 }
##fetchRegistry
1/** 2 * Fetches the registry information. 3 * 4 * <p> 5 * This method tries to get only deltas after the first fetch unless there 6 * is an issue in reconciling eureka server and client registry information. 7 * </p> 8 * 9 * @param forceFullRegistryFetch Forces a full registry fetch. 10 * 11 * @return true if the registry was fetched 12 */ 13 private boolean fetchRegistry(boolean forceFullRegistryFetch) { 14 Stopwatch tracer = FETCH_REGISTRY_TIMER.start(); 15 16 try { 17 // If the delta is disabled or if it is the first time, get all 18 // applications 19 Applications applications = getApplications(); 20 21 if (clientConfig.shouldDisableDelta() 22 || (!Strings.isNullOrEmpty(clientConfig.getRegistryRefreshSingleVipAddress())) 23 || forceFullRegistryFetch 24 || (applications == null) 25 || (applications.getRegisteredApplications().size() == 0) 26 || (applications.getVersion() == -1)) //Client application does not have latest library supporting delta 27 { 28 logger.info("Disable delta property : {}", clientConfig.shouldDisableDelta()); 29 logger.info("Single vip registry refresh property : {}", clientConfig.getRegistryRefreshSingleVipAddress()); 30 logger.info("Force full registry fetch : {}", forceFullRegistryFetch); 31 logger.info("Application is null : {}", (applications == null)); 32 logger.info("Registered Applications size is zero : {}", 33 (applications.getRegisteredApplications().size() == 0)); 34 logger.info("Application version is -1: {}", (applications.getVersion() == -1)); 35 getAndStoreFullRegistry(); 36 } else { 37 getAndUpdateDelta(applications); 38 } 39 applications.setAppsHashCode(applications.getReconcileHashCode()); 40 logTotalInstances(); 41 } catch (Throwable e) { 42 logger.error(PREFIX + appPathIdentifier + " - was unable to refresh its cache! status = " + e.getMessage(), e); 43 return false; 44 } finally { 45 if (tracer != null) { 46 tracer.stop(); 47 } 48 } 49 50 // Notify about cache refresh before updating the instance remote status 51 onCacheRefreshed(); 52 53 // Update remote status based on refreshed data held in the cache 54 updateInstanceRemoteStatus(); 55 56 // registry was fetched successfully, so return true 57 return true; 58 }
##getAndUpdateDelta
1/** 2 * Get the delta registry information from the eureka server and update it locally. 3 * When applying the delta, the following flow is observed: 4 * 5 * if (update generation have not advanced (due to another thread)) 6 * atomically try to: update application with the delta and get reconcileHashCode 7 * abort entire processing otherwise 8 * do reconciliation if reconcileHashCode clash 9 * fi 10 * 11 * @return the client response 12 * @throws Throwable on error 13 */ 14 private void getAndUpdateDelta(Applications applications) throws Throwable { 15 long currentUpdateGeneration = fetchRegistryGeneration.get(); 16 17 Applications delta = null; 18 EurekaHttpResponse<Applications> httpResponse = eurekaTransport.queryClient.getDelta(remoteRegionsRef.get()); 19 if (httpResponse.getStatusCode() == Status.OK.getStatusCode()) { 20 delta = httpResponse.getEntity(); 21 } 22 23 if (delta == null) { 24 logger.warn("The server does not allow the delta revision to be applied because it is not safe. " 25 + "Hence got the full registry."); 26 getAndStoreFullRegistry(); 27 } else if (fetchRegistryGeneration.compareAndSet(currentUpdateGeneration, currentUpdateGeneration + 1)) { 28 logger.debug("Got delta update with apps hashcode {}", delta.getAppsHashCode()); 29 String reconcileHashCode = ""; 30 if (fetchRegistryUpdateLock.tryLock()) { 31 try { 32 updateDelta(delta); 33 reconcileHashCode = getReconcileHashCode(applications); 34 } finally { 35 fetchRegistryUpdateLock.unlock(); 36 } 37 } else { 38 logger.warn("Cannot acquire update lock, aborting getAndUpdateDelta"); 39 } 40 // There is a diff in number of instances for some reason 41 if (!reconcileHashCode.equals(delta.getAppsHashCode()) || clientConfig.shouldLogDeltaDiff()) { 42 reconcileAndLogDifference(delta, reconcileHashCode); // this makes a remoteCall 43 } 44 } else { 45 logger.warn("Not updating application delta as another thread is updating it already"); 46 logger.debug("Ignoring delta update with apps hashcode {}, as another thread is updating it already", delta.getAppsHashCode()); 47 } 48 }
##updateDelta
1/** 2 * Updates the delta information fetches from the eureka server into the 3 * local cache. 4 * 5 * @param delta 6 * the delta information received from eureka server in the last 7 * poll cycle. 8 */ 9 private void updateDelta(Applications delta) { 10 int deltaCount = 0; 11 for (Application app : delta.getRegisteredApplications()) { 12 for (InstanceInfo instance : app.getInstances()) { 13 Applications applications = getApplications(); 14 String instanceRegion = instanceRegionChecker.getInstanceRegion(instance); 15 if (!instanceRegionChecker.isLocalRegion(instanceRegion)) { 16 Applications remoteApps = remoteRegionVsApps.get(instanceRegion); 17 if (null == remoteApps) { 18 remoteApps = new Applications(); 19 remoteRegionVsApps.put(instanceRegion, remoteApps); 20 } 21 applications = remoteApps; 22 } 23 24 ++deltaCount; 25 if (ActionType.ADDED.equals(instance.getActionType())) { 26 Application existingApp = applications.getRegisteredApplications(instance.getAppName()); 27 if (existingApp == null) { 28 applications.addApplication(app); 29 } 30 logger.debug("Added instance {} to the existing apps in region {}", instance.getId(), instanceRegion); 31 applications.getRegisteredApplications(instance.getAppName()).addInstance(instance); 32 } else if (ActionType.MODIFIED.equals(instance.getActionType())) { 33 Application existingApp = applications.getRegisteredApplications(instance.getAppName()); 34 if (existingApp == null) { 35 applications.addApplication(app); 36 } 37 logger.debug("Modified instance {} to the existing apps ", instance.getId()); 38 39 applications.getRegisteredApplications(instance.getAppName()).addInstance(instance); 40 41 } else if (ActionType.DELETED.equals(instance.getActionType())) { 42 Application existingApp = applications.getRegisteredApplications(instance.getAppName()); 43 if (existingApp == null) { 44 applications.addApplication(app); 45 } 46 logger.debug("Deleted instance {} to the existing apps ", instance.getId()); 47 applications.getRegisteredApplications(instance.getAppName()).removeInstance(instance); 48 } 49 } 50 } 51 logger.debug("The total number of instances fetched by the delta processor : {}", deltaCount); 52 53 getApplications().setVersion(delta.getVersion()); 54 getApplications().shuffleInstances(clientConfig.shouldFilterOnlyUpInstances()); 55 56 for (Applications applications : remoteRegionVsApps.values()) { 57 applications.setVersion(delta.getVersion()); 58 applications.shuffleInstances(clientConfig.shouldFilterOnlyUpInstances()); 59 } 60 }
获取application信息，判断是否删除，如果是则删除实例
1if (ActionType.DELETED.equals(instance.getActionType())) { 2 Application existingApp = applications.getRegisteredApplications(instance.getAppName()); 3 if (existingApp == null) { 4 applications.addApplication(app); 5 } 6 logger.debug("Deleted instance {} to the existing apps ", instance.getId()); 7 applications.getRegisteredApplications(instance.getAppName()).removeInstance(instance); 8 }