SpringCloud源码剖析-Eureka Client服务发现

/**初始化定时调度任务* Initializes all scheduled tasks.*/privatevoidinitScheduledTasks() {
//判断是否开启服务发现功能if (clientConfig.shouldFetchRegistry()) {
// registry cache refresh timer//刷新注册表定时任务时间间隔intregistryFetchIntervalSeconds=clientConfig.getRegistryFetchIntervalSeconds();
intexpBackOffBound=clientConfig.getCacheRefreshExecutorExponentialBackOffBound();
//定时任务调度器scheduler.schedule(
newTimedSupervisorTask(
"cacheRefresh",
scheduler,
cacheRefreshExecutor,
registryFetchIntervalSeconds,
TimeUnit.SECONDS,
expBackOffBound,
//属性注册表任务newCacheRefreshThread()
            ),
registryFetchIntervalSeconds, TimeUnit.SECONDS);
    }
    ....省略代码....

classCacheRefreshThreadimplementsRunnable {
publicvoidrun() {
//调用刷新注册表方法refreshRegistry();
    }
}
@VisibleForTestingvoidrefreshRegistry() {
try {
//这里在获取Regions 不为空，默认为空booleanisFetchingRemoteRegionRegistries=isFetchingRemoteRegionRegistries();
booleanremoteRegionsModified=false;
// This makes sure that a dynamic change to remote regions to fetch is honored.//这里在获取eureka:client:fetch-remote-regions-registry配置，即远程regions，默认为空StringlatestRemoteRegions=clientConfig.fetchRegistryForRemoteRegions();
if (null!=latestRemoteRegions) {
            ....省略.....
            }
//开始获取注册表fetchRegistry(false);booleansuccess=fetchRegistry(remoteRegionsModified);
if (success) {
//获取注册表大小registrySize=localRegionApps.get().size();
lastSuccessfulRegistryFetchTimestamp=System.currentTimeMillis();
        }
if (logger.isDebugEnabled()) {
            ...省略...
            }
    } catch (Throwablee) {
logger.error("Cannot fetch registry from server", e);
    }
}
...省略代码...

privatebooleanfetchRegistry(booleanforceFullRegistryFetch) {
//计算器开始Stopwatchtracer=FETCH_REGISTRY_TIMER.start();
try {
// If the delta is disabled or if it is the first time, get all// applications//这里是获取本地所有的注册的服务应用，里面包含了注册的所有的服务Applicationsapplications=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 {
//从eureka服务器获取注册表信息，差别获取，并在本地更新getAndUpdateDelta(applications);
        }
applications.setAppsHashCode(applicat                                     ...省略代码...                  

privatevoidgetAndStoreFullRegistry() throwsThrowable {
longcurrentUpdateGeneration=fetchRegistryGeneration.get();
logger.info("Getting all instance registry info from the eureka server");
Applicationsapps=null;
//通过eurekaTransport.queryClient得到一个EurekaHttpClientEurekaHttpResponse<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");
    } elseif (fetchRegistryGeneration.compareAndSet(currentUpdateGeneration, currentUpdateGeneration+1)) {
//存储注册表到Applications中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");
    }
}

@OverridepublicEurekaHttpResponse<Applications>getApplications(finalString... regions) {
returnexecute(newRequestExecutor<Applications>() {
@OverridepublicEurekaHttpResponse<Applications>execute(EurekaHttpClientdelegate) {
//最终会调用AbstractJerseyEurekaHttpClient获取注册表returndelegate.getApplications(regions);
        }
@OverridepublicRequestTypegetRequestType() {
returnRequestType.GetApplications;
        }
    });
}

publicabstractclassAbstractJerseyEurekaHttpClientimplementsEurekaHttpClient {
@OverridepublicEurekaHttpResponse<Applications>getApplications(String... regions) {
returngetApplicationsInternal("apps/", regions);
    }
    ...省略...
privateEurekaHttpResponse<Applications>getApplicationsInternal(StringurlPath, String[] regions) {
ClientResponseresponse=null;
StringregionsParamValue=null;
try {
//得到一个web请求，serviceUrl是服务器的地址WebResourcewebResource=jerseyClient.resource(serviceUrl).path(urlPath);
if (regions!=null&&regions.length>0) {
regionsParamValue=StringUtil.join(regions);
webResource=webResource.queryParam("regions", regionsParamValue);
            }
BuilderrequestBuilder=webResource.getRequestBuilder();
addExtraHeaders(requestBuilder);
//发送get请求response=requestBuilder.accept(MediaType.APPLICATION_JSON_TYPE).get(ClientResponse.class);
Applicationsapplications=null;
if (response.getStatus() ==Status.OK.getStatusCode() &&response.hasEntity()) {
//获取返回的服务注册表applications=response.getEntity(Applications.class);
            }
//创建一个相应对象EurekaHttpResponseBuilderreturnanEurekaHttpResponse(response.getStatus(), Applications.class)
                .headers(headersOf(response))
                .entity(applications)
                .build();
        } finally {
if (logger.isDebugEnabled()) {
logger.debug("Jersey HTTP GET {}/{}?{}; statusCode={}",
serviceUrl, urlPath,
regionsParamValue==null?"" : "regions="+regionsParamValue,
response==null?"N/A" : response.getStatus()
                            );
            }
if (response!=null) {
response.close();
            }
        }
    }

privatevoidgetAndUpdateDelta(Applicationsapplications) throwsThrowable {
longcurrentUpdateGeneration=fetchRegistryGeneration.get();
Applicationsdelta=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();
    } elseif (fetchRegistryGeneration.compareAndSet(currentUpdateGeneration, currentUpdateGeneration+1)) {
logger.debug("Got delta update with apps hashcode {}", delta.getAppsHashCode());
StringreconcileHashCode="";
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 reasonif (!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());
    }
}