注册中心 Eureka 源码解析 —— 应用实例注册发现（六）之全量获取

// DiscoveryClient.java
/**
* Applications 在本地的缓存
*/
private final AtomicReference<Applications> localRegionApps = new AtomicReference<Applications>();

DiscoveryClient(ApplicationInfoManager applicationInfoManager, EurekaClientConfig config, AbstractDiscoveryClientOptionalArgs args,
                    Provider<BackupRegistry> backupRegistryProvider) {

     // ... 省略无关代码

    // 【3.2.5】初始化应用集合在本地的缓存
    localRegionApps.set(new Applications());

    // ... 省略无关代码     

    // 【3.2.12】从 Eureka-Server 拉取注册信息
    if (clientConfig.shouldFetchRegistry() && !fetchRegistry(false)) {
        fetchRegistryFromBackup();
    }

     // ... 省略无关代码       
}

// DiscoveryClient.java
DiscoveryClient(ApplicationInfoManager applicationInfoManager, EurekaClientConfig config, AbstractDiscoveryClientOptionalArgs args,
               Provider<BackupRegistry> backupRegistryProvider) {
    // ... 省略无关代码

    // 【3.2.9】初始化线程池
    // default size of 2 - 1 each for heartbeat and cacheRefresh
    scheduler = Executors.newScheduledThreadPool(2,
         new ThreadFactoryBuilder()
                 .setNameFormat("DiscoveryClient-%d")
                 .setDaemon(true)
                 .build());

    cacheRefreshExecutor = new ThreadPoolExecutor(
         1, clientConfig.getCacheRefreshExecutorThreadPoolSize(), 0, TimeUnit.SECONDS,
         new SynchronousQueue<Runnable>(),
         new ThreadFactoryBuilder()
                 .setNameFormat("DiscoveryClient-CacheRefreshExecutor-%d")
                 .setDaemon(true)
                 .build()
     );  // use direct handoff

    // ... 省略无关代码

    // 【3.2.14】初始化定时任务
    initScheduledTasks();

    // ... 省略无关代码
}

private void initScheduledTasks() {
    // 向 Eureka-Server 心跳（续租）执行器
    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);
     }
     // ... 省略无关代码
}

// DiscoveryClient.java
  1: void refreshRegistry() {
  2:     try {
  3:         // TODO 芋艿：TODO[0009]：RemoteRegionRegistry
  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:             // 设置 注册信息的应用实例数
 33:             registrySize = localRegionApps.get().size();
 34:             // 设置 最后获取注册信息时间
 35:             lastSuccessfulRegistryFetchTimestamp = System.currentTimeMillis();
 36:         }
 37: 
 38:         // 打印日志
 39:         if (logger.isDebugEnabled()) {
 40:             StringBuilder allAppsHashCodes = new StringBuilder();
 41:             allAppsHashCodes.append("Local region apps hashcode: ");
 42:             allAppsHashCodes.append(localRegionApps.get().getAppsHashCode());
 43:             allAppsHashCodes.append(", is fetching remote regions? ");
 44:             allAppsHashCodes.append(isFetchingRemoteRegionRegistries);
 45:             for (Map.Entry<String, Applications> entry : remoteRegionVsApps.entrySet()) {
 46:                 allAppsHashCodes.append(", Remote region: ");
 47:                 allAppsHashCodes.append(entry.getKey());
 48:                 allAppsHashCodes.append(" , apps hashcode: ");
 49:                 allAppsHashCodes.append(entry.getValue().getAppsHashCode());
 50:             }
 51:             logger.debug("Completed cache refresh task for discovery. All Apps hash code is {} ",
 52:                     allAppsHashCodes.toString());
 53:         }
 54:     } catch (Throwable e) {
 55:         logger.error("Cannot fetch registry from server", e);
 56:     }        
 57: }

  1: private boolean fetchRegistry(boolean forceFullRegistryFetch) {
  2:     Stopwatch tracer = FETCH_REGISTRY_TIMER.start();
  3: 
  4:     try {
  5:         // 获取 本地缓存的注册的应用实例集合
  6:         // If the delta is disabled or if it is the first time, get all
  7:         // applications
  8:         Applications applications = getApplications();
  9: 
 10:         // 全量获取
 11:         if (clientConfig.shouldDisableDelta() // 禁用增量获取
 12:                 || (!Strings.isNullOrEmpty(clientConfig.getRegistryRefreshSingleVipAddress()))
 13:                 || forceFullRegistryFetch
 14:                 || (applications == null) // 空
 15:                 || (applications.getRegisteredApplications().size() == 0) // 空
 16:                 || (applications.getVersion() == -1)) //Client application does not have latest library supporting delta
 17:         {
 18:             logger.info("Disable delta property : {}", clientConfig.shouldDisableDelta());
 19:             logger.info("Single vip registry refresh property : {}", clientConfig.getRegistryRefreshSingleVipAddress());
 20:             logger.info("Force full registry fetch : {}", forceFullRegistryFetch);
 21:             logger.info("Application is null : {}", (applications == null));
 22:             logger.info("Registered Applications size is zero : {}",
 23:                     (applications.getRegisteredApplications().size() == 0));
 24:             logger.info("Application version is -1: {}", (applications.getVersion() == -1));
 25:             // 执行 全量获取
 26:             getAndStoreFullRegistry();
 27:         } else {
 28:             // 执行 增量获取
 29:             getAndUpdateDelta(applications);
 30:         }
 31:         // 设置 应用集合 hashcode
 32:         applications.setAppsHashCode(applications.getReconcileHashCode());
 33:         // 打印 本地缓存的注册的应用实例数量
 34:         logTotalInstances();
 35:     } catch (Throwable e) {
 36:         logger.error(PREFIX + appPathIdentifier + " - was unable to refresh its cache! status = " + e.getMessage(), e);
 37:         return false;
 38:     } finally {
 39:         if (tracer != null) {
 40:             tracer.stop();
 41:         }
 42:     }
 43: 
 44:     // Notify about cache refresh before updating the instance remote status
 45:     onCacheRefreshed();
 46: 
 47:     // Update remote status based on refreshed data held in the cache
 48:     updateInstanceRemoteStatus();
 49: 
 50:     // registry was fetched successfully, so return true
 51:     return true;
 52: }

  1: private void getAndStoreFullRegistry() throws Throwable {
  2:     long currentUpdateGeneration = fetchRegistryGeneration.get();
  3: 
  4:     logger.info("Getting all instance registry info from the eureka server");
  5: 
  6:     // 全量获取注册信息
  7:     Applications apps = null;
  8:     EurekaHttpResponse<Applications> httpResponse = clientConfig.getRegistryRefreshSingleVipAddress() == null
  9:             ? eurekaTransport.queryClient.getApplications(remoteRegionsRef.get())
 10:             : eurekaTransport.queryClient.getVip(clientConfig.getRegistryRefreshSingleVipAddress(), remoteRegionsRef.get());
 11:     if (httpResponse.getStatusCode() == Status.OK.getStatusCode()) {
 12:         apps = httpResponse.getEntity();
 13:     }
 14:     logger.info("The response status is {}", httpResponse.getStatusCode());
 15: 
 16:     // 设置到本地缓存
 17:     if (apps == null) {
 18:         logger.error("The application is null for some reason. Not storing this information");
 19:     } else if (fetchRegistryGeneration.compareAndSet(currentUpdateGeneration, currentUpdateGeneration + 1)) {
 20:         localRegionApps.set(this.filterAndShuffle(apps));
 21:         logger.debug("Got full registry with apps hashcode {}", apps.getAppsHashCode());
 22:     } else {
 23:         logger.warn("Not updating applications as another thread is updating it already");
 24:     }
 25: }

  1: @GET
  2: public Response getContainers(@PathParam("version") String version,
  3:                               @HeaderParam(HEADER_ACCEPT) String acceptHeader,
  4:                               @HeaderParam(HEADER_ACCEPT_ENCODING) String acceptEncoding,
  5:                               @HeaderParam(EurekaAccept.HTTP_X_EUREKA_ACCEPT) String eurekaAccept,
  6:                               @Context UriInfo uriInfo,
  7:                               @Nullable @QueryParam("regions") String regionsStr) {
  8:     // TODO[0009]：RemoteRegionRegistry
  9:     boolean isRemoteRegionRequested = null != regionsStr && !regionsStr.isEmpty();
 10:     String[] regions = null;
 11:     if (!isRemoteRegionRequested) {
 12:         EurekaMonitors.GET_ALL.increment();
 13:     } else {
 14:         regions = regionsStr.toLowerCase().split(",");
 15:         Arrays.sort(regions); // So we don't have different caches for same regions queried in different order.
 16:         EurekaMonitors.GET_ALL_WITH_REMOTE_REGIONS.increment();
 17:     }
 18: 
 19:     // 判断是否可以访问
 20:     // Check if the server allows the access to the registry. The server can
 21:     // restrict access if it is not
 22:     // ready to serve traffic depending on various reasons.
 23:     if (!registry.shouldAllowAccess(isRemoteRegionRequested)) {
 24:         return Response.status(Status.FORBIDDEN).build();
 25:     }
 26: 
 27:     // API 版本
 28:     CurrentRequestVersion.set(Version.toEnum(version));
 29: 
 30:     // 返回数据格式
 31:     KeyType keyType = Key.KeyType.JSON;
 32:     String returnMediaType = MediaType.APPLICATION_JSON;
 33:     if (acceptHeader == null || !acceptHeader.contains(HEADER_JSON_VALUE)) {
 34:         keyType = Key.KeyType.XML;
 35:         returnMediaType = MediaType.APPLICATION_XML;
 36:     }
 37: 
 38:     // 响应缓存键( KEY )
 39:     Key cacheKey = new Key(Key.EntityType.Application,
 40:             ResponseCacheImpl.ALL_APPS,
 41:             keyType, CurrentRequestVersion.get(), EurekaAccept.fromString(eurekaAccept), regions
 42:     );
 43: 
 44:     //
 45:     Response response;
 46:     if (acceptEncoding != null && acceptEncoding.contains(HEADER_GZIP_VALUE)) {
 47:         response = Response.ok(responseCache.getGZIP(cacheKey))
 48:                 .header(HEADER_CONTENT_ENCODING, HEADER_GZIP_VALUE)
 49:                 .header(HEADER_CONTENT_TYPE, returnMediaType)
 50:                 .build();
 51:     } else {
 52:         response = Response.ok(responseCache.get(cacheKey))
 53:                 .build();
 54:     }
 55:     return response;
 56: }

public interface ResponseCache {

    String get(Key key);

    byte[] getGZIP(Key key);

    void invalidate(String appName, @Nullable String vipAddress, @Nullable String secureVipAddress);

    AtomicLong getVersionDelta();

    AtomicLong getVersionDeltaWithRegions();

}

public class Key {

    public enum KeyType {
        JSON, XML
    }

    /**
     * An enum to define the entity that is stored in this cache for this key.
     */
    public enum EntityType {
        Application, VIP, SVIP
    }

    /**
     * 实体名
     */
    private final String entityName;
    /**
     * TODO[0009]：RemoteRegionRegistry
     */
    private final String[] regions;
    /**
     * 请求参数类型
     */
    private final KeyType requestType;
    /**
     * 请求 API 版本号
     */
    private final Version requestVersion;
    /**
     * hashKey
     */
    private final String hashKey;
    /**
     * 实体类型
     *
     * {@link EntityType}
     */
    private final EntityType entityType;
    /**
     * {@link EurekaAccept}
     */
    private final EurekaAccept eurekaAccept;

    public Key(EntityType entityType, String entityName, KeyType type, Version v, EurekaAccept eurekaAccept, @Nullable String[] regions) {
        this.regions = regions;
        this.entityType = entityType;
        this.entityName = entityName;
        this.requestType = type;
        this.requestVersion = v;
        this.eurekaAccept = eurekaAccept;
        hashKey = this.entityType + this.entityName + (null != this.regions ? Arrays.toString(this.regions) : "")
                + requestType.name() + requestVersion.name() + this.eurekaAccept.name();
    }

    public Key(EntityType entityType, String entityName, KeyType type, Version v, EurekaAccept eurekaAccept, @Nullable String[] regions) {
        this.regions = regions;
        this.entityType = entityType;
        this.entityName = entityName;
        this.requestType = type;
        this.requestVersion = v;
        this.eurekaAccept = eurekaAccept;
        hashKey = this.entityType + this.entityName + (null != this.regions ? Arrays.toString(this.regions) : "")
                + requestType.name() + requestVersion.name() + this.eurekaAccept.name();
    }

    @Override
    public int hashCode() {
        String hashKey = getHashKey();
        return hashKey.hashCode();
    }

    @Override
    public boolean equals(Object other) {
        if (other instanceof Key) {
            return getHashKey().equals(((Key) other).getHashKey());
        } else {
            return false;
        }
    }

}

  1: private final ConcurrentMap<Key, Value> readOnlyCacheMap = new ConcurrentHashMap<Key, Value>();
  2: 
  3: private final LoadingCache<Key, Value> readWriteCacheMap;
  4: 
  5: public String get(final Key key) {
  6:     return get(key, shouldUseReadOnlyResponseCache);
  7: }
  8: 
  9: String get(final Key key, boolean useReadOnlyCache) {
 10:     Value payload = getValue(key, useReadOnlyCache);
 11:     if (payload == null || payload.getPayload().equals(EMPTY_PAYLOAD)) {
 12:         return null;
 13:     } else {
 14:         return payload.getPayload();
 15:     }
 16: }
 17: 
 18: Value getValue(final Key key, boolean useReadOnlyCache) {
 19:     Value payload = null;
 20:     try {
 21:         if (useReadOnlyCache) {
 22:             final Value currentPayload = readOnlyCacheMap.get(key);
 23:             if (currentPayload != null) {
 24:                 payload = currentPayload;
 25:             } else {
 26:                 payload = readWriteCacheMap.get(key);
 27:                 readOnlyCacheMap.put(key, payload);
 28:             }
 29:         } else {
 30:             payload = readWriteCacheMap.get(key);
 31:         }
 32:     } catch (Throwable t) {
 33:         logger.error("Cannot get value for key :" + key, t);
 34:     }
 35:     return payload;
 36: }

// ... 省略代码，超过微信文章长度

// ... 省略代码，超过微信文章长度

// ... 省略代码，超过微信文章长度

expireAfterWrite(serverConfig.getResponseCacheAutoExpirationInSeconds())

// ... 省略代码，超过微信文章长度