這次我們說(shuō)一下eureka的全量獲取。EurekaClient 啟動(dòng)時(shí),首先執(zhí)行一次全量獲取進(jìn)行本地緩存注冊(cè)信息。
localRegionApps.set(new Applications());
if (clientConfig.shouldFetchRegistry() && !fetchRegistry(false)) {
fetchRegistryFromBackup();
}
配置 eureka.shouldFetchRegistry = true,開(kāi)啟從 EurekaServer獲取注冊(cè)信息泳梆。默認(rèn)值:true。
調(diào)用 fetchRegistry(false)方法榜掌,從 Eureka-Server 全量獲取注冊(cè)信息优妙。
EurekaClient在初始化過(guò)程中,創(chuàng)建獲取注冊(cè)信息線程憎账,固定間隔向 Eureka-Server 發(fā)起獲取注冊(cè)信息( fetch )套硼,刷新本地注冊(cè)信息緩存
private void initScheduledTasks() {
// 從 Eureka-Server 拉取注冊(cè)信息執(zhí)行器
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);
}
// 向 Eureka-Server 心跳(續(xù)租)執(zhí)行器
if (clientConfig.shouldRegisterWithEureka()) {
int renewalIntervalInSecs = instanceInfo.getLeaseInfo().getRenewalIntervalInSecs(); // 續(xù)租頻率
int expBackOffBound = clientConfig.getHeartbeatExecutorExponentialBackOffBound(); // 系數(shù)
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);
// 創(chuàng)建 應(yīng)用實(shí)例信息復(fù)制器
// InstanceInfo replicator
instanceInfoReplicator = new InstanceInfoReplicator(
this,
instanceInfo,
clientConfig.getInstanceInfoReplicationIntervalSeconds(),
2); // burstSize
// 創(chuàng)建 應(yīng)用實(shí)例狀態(tài)變更監(jiān)聽(tīng)器
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();
}
};
// 注冊(cè) 應(yīng)用實(shí)例狀態(tài)變更監(jiān)聽(tīng)器
if (clientConfig.shouldOnDemandUpdateStatusChange()) {
applicationInfoManager.registerStatusChangeListener(statusChangeListener);
}
// 開(kāi)啟 應(yīng)用實(shí)例信息復(fù)制器
instanceInfoReplicator.start(clientConfig.getInitialInstanceInfoReplicationIntervalSeconds());
} else {
logger.info("Not registering with Eureka server per configuration");
}
}
調(diào)用 refreshRegistry(false)方法,刷新注冊(cè)信息緩存
void refreshRegistry() {
try {
boolean isFetchingRemoteRegionRegistries = isFetchingRemoteRegionRegistries();
boolean remoteRegionsModified = false;
// This makes sure that a dynamic change to remote regions to fetch is honored.
String latestRemoteRegions = clientConfig.fetchRegistryForRemoteRegions();
if (null != latestRemoteRegions) {
String currentRemoteRegions = remoteRegionsToFetch.get();
if (!latestRemoteRegions.equals(currentRemoteRegions)) {
// Both remoteRegionsToFetch and AzToRegionMapper.regionsToFetch need to be in sync
synchronized (instanceRegionChecker.getAzToRegionMapper()) {
if (remoteRegionsToFetch.compareAndSet(currentRemoteRegions, latestRemoteRegions)) {
String[] remoteRegions = latestRemoteRegions.split(",");
remoteRegionsRef.set(remoteRegions);
instanceRegionChecker.getAzToRegionMapper().setRegionsToFetch(remoteRegions);
remoteRegionsModified = true;
} else {
logger.info("Remote regions to fetch modified concurrently," +
" ignoring change from {} to {}", currentRemoteRegions, latestRemoteRegions);
}
}
} else {
// Just refresh mapping to reflect any DNS/Property change
instanceRegionChecker.getAzToRegionMapper().refreshMapping();
}
}
boolean success = fetchRegistry(remoteRegionsModified);
if (success) {
// 設(shè)置 注冊(cè)信息的應(yīng)用實(shí)例數(shù)
registrySize = localRegionApps.get().size();
// 設(shè)置 最后獲取注冊(cè)信息時(shí)間
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.toString());
}
} catch (Throwable e) {
logger.error("Cannot fetch registry from server", e);
}
}
調(diào)用 fetchRegistry方法胞皱,從EurekaServer獲取注冊(cè)信息
private boolean fetchRegistry(boolean forceFullRegistryFetch) {
Stopwatch tracer = FETCH_REGISTRY_TIMER.start();
try {
// 獲取 本地緩存的注冊(cè)的應(yīng)用實(shí)例集合
// 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));
// 執(zhí)行 全量獲取
getAndStoreFullRegistry();
} else {
// 執(zhí)行 增量獲取
getAndUpdateDelta(applications);
}
// 設(shè)置 應(yīng)用集合 hashcode
applications.setAppsHashCode(applications.getReconcileHashCode());
// 打印 本地緩存的注冊(cè)的應(yīng)用實(shí)例數(shù)量
logTotalInstances();
} catch (Throwable e) {
logger.error(PREFIX + appPathIdentifier + " - was unable to refresh its cache! status = " + 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;
}
調(diào)用 getAndStoreFullRegistry方法邪意,全量獲取注冊(cè)信息,并設(shè)置到本地緩存
private void getAndStoreFullRegistry() throws Throwable {
long currentUpdateGeneration = fetchRegistryGeneration.get();
logger.info("Getting all instance registry info from the eureka server");
// 全量獲取注冊(cè)信息
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());
// 設(shè)置到本地緩存
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");
}
}
ApplicationsResource反砌,處理所有應(yīng)用的請(qǐng)求操作的 Resource ( Controller )抄罕。
接收全量獲取請(qǐng)求,映射 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) {
// TODO[0009]:RemoteRegionRegistry
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();
}
// 判斷是否可以訪問(wèn)
// 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();
}
// API 版本
CurrentRequestVersion.set(Version.toEnum(version));
// 返回?cái)?shù)據(jù)格式
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;
}
// 響應(yīng)緩存鍵( KEY )
Key cacheKey = new Key(Key.EntityType.Application,
ResponseCacheImpl.ALL_APPS,
keyType, CurrentRequestVersion.get(), EurekaAccept.fromString(eurekaAccept), regions
);
// 響應(yīng)緩存結(jié)果
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;
}
ResponseCacheImpl于颖,響應(yīng)緩存實(shí)現(xiàn)類。
在 ResponseCacheImpl 里嚷兔,將緩存拆分成兩層 :
只讀緩存( readOnlyCacheMap )
固定過(guò)期 + 固定大小的讀寫緩存( readWriteCacheMap )
緩存過(guò)期策略如下:
應(yīng)用實(shí)例注冊(cè)森渐、下線做入、過(guò)期時(shí),只過(guò)期 readWriteCacheMap 同衣。
readWriteCacheMap 寫入一段時(shí)間( 可配置 )后自動(dòng)過(guò)期竟块。
定時(shí)任務(wù)對(duì)比 readWriteCacheMap 和 readOnlyCacheMap 的緩存值,若不一致耐齐,以前者為主浪秘。通過(guò)這樣的方式,實(shí)現(xiàn)了 readOnlyCacheMap 的定時(shí)過(guò)期埠况。
調(diào)用ResponseCacheImpl的get方法耸携,獲取緩存
String get(final Key key, boolean useReadOnlyCache) {
Value payload = getValue(key, useReadOnlyCache);
if (payload == null || payload.getPayload().equals(EMPTY_PAYLOAD)) {
return null;
} else {
return payload.getPayload();
}
}
調(diào)用 AbstractInstanceRegistry的getApplications方法,獲取注冊(cè)的應(yīng)用集合
public Applications getApplications() {
boolean disableTransparentFallback = serverConfig.disableTransparentFallbackToOtherRegion();
if (disableTransparentFallback) { // TODO[0009]:RemoteRegionRegistry
return getApplicationsFromLocalRegionOnly();
} else {
return getApplicationsFromAllRemoteRegions(); // Behavior of falling back to remote region can be disabled.
}
}
定時(shí)任務(wù)對(duì)比 readWriteCacheMap 和 readOnlyCacheMap 的緩存值辕翰,若不一致夺衍,以前者為主。通過(guò)這樣的方式喜命,實(shí)現(xiàn)了 readOnlyCacheMap 的定時(shí)過(guò)期沟沙。
private TimerTask getCacheUpdateTask() {
return new TimerTask() {
@Override
public void run() {
logger.debug("Updating the client cache from response cache");
for (Key key : readOnlyCacheMap.keySet()) { // 循環(huán) readOnlyCacheMap 的緩存鍵
if (logger.isDebugEnabled()) {
Object[] args = {key.getEntityType(), key.getName(), key.getVersion(), key.getType()};
logger.debug("Updating the client cache from response cache for key : {} {} {} {}", args);
}
try {
CurrentRequestVersion.set(key.getVersion());
Value cacheValue = readWriteCacheMap.get(key);
Value currentCacheValue = readOnlyCacheMap.get(key);
if (cacheValue != currentCacheValue) { // 不一致時(shí),進(jìn)行替換
readOnlyCacheMap.put(key, cacheValue);
}
} catch (Throwable th) {
logger.error("Error while updating the client cache from response cache for key {}", key.toStringCompact(), th);
}
}
}
};
}
eureka的全量獲取就完成了壁榕。