SpringCloud轮询拉取注册表与服务发现流程详解

@Singleton
public class DiscoveryClient implements EurekaClient {
    ....省略n行代码......
    private final ScheduledExecutorService scheduler;
    // additional executors for supervised subtasks监督子任务的附加执行器
    private final ThreadPoolExecutor heartbeatExecutor;
    private final ThreadPoolExecutor cacheRefreshExecutor;
    private TimedSupervisorTask cacheRefreshTask;
    private TimedSupervisorTask heartbeatTask;
    ....省略n行代码......
    // Spring容器初始化时候调用
    public DiscoveryClient(ApplicationInfoManager applicationInfoManager,
                           final EurekaClientConfig config, AbstractDiscoveryClientOptionalArgs args) {
        // 调用下面重载方法
        this(applicationInfoManager, config, args, ResolverUtils::randomize);
    }
    public DiscoveryClient(ApplicationInfoManager applicationInfoManager,
                           final EurekaClientConfig config,
                           AbstractDiscoveryClientOptionalArgs args, EndpointRandomizer randomizer) {
        this(applicationInfoManager, config, args, new Provider<BackupRegistry>() {
        ....省略n行代码......
    }
    @Inject
    DiscoveryClient(ApplicationInfoManager applicationInfoManager, EurekaClientConfig config,
                    AbstractDiscoveryClientOptionalArgs args,
                    Provider<BackupRegistry> backupRegistryProvider, EndpointRandomizer endpointRandomizer) {
        try {
            // default size of 2 - 1 each for heartbeat and cacheRefresh心跳和缓存刷新的默认大小分别为2-1
            scheduler = Executors.newScheduledThreadPool(2,
                    new ThreadFactoryBuilder()
                            .setNameFormat("DiscoveryClient-%d")
                            .setDaemon(true)
                            .build());
            // 心跳执行者
            heartbeatExecutor = new ThreadPoolExecutor(
                    1, clientConfig.getHeartbeatExecutorThreadPoolSize(), 0, TimeUnit.SECONDS,
                    new SynchronousQueue<Runnable>(),
                    new ThreadFactoryBuilder()
                            .setNameFormat("DiscoveryClient-HeartbeatExecutor-%d")
                            .setDaemon(true)
                            .build()
            );  // use direct handoff
            // 缓存刷新执行者
            cacheRefreshExecutor = new ThreadPoolExecutor(
                    1, clientConfig.getCacheRefreshExecutorThreadPoolSize(), 0, TimeUnit.SECONDS,
                    new SynchronousQueue<Runnable>(),
                    new ThreadFactoryBuilder()
                            .setNameFormat("DiscoveryClient-CacheRefreshExecutor-%d")
                            .setDaemon(true)
                            .build()
            );  // use direct handoff
            // 初始化通信封装类
            eurekaTransport = new EurekaTransport();
        ....省略n行代码......
        } catch (Throwable e) {
            throw new RuntimeException("Failed to initialize DiscoveryClient!", e);
        }
        // 默认true，可更改配置不建议
        if (clientConfig.shouldFetchRegistry()) {
            try {// 初始化注册表
                boolean primaryFetchRegistryResult = fetchRegistry(false);
                // 下面主要打印失败日志，初始化时控制台可见是处理成功的
                if (!primaryFetchRegistryResult) {
                    // 从主服务器初始注册表提取失败
                    logger.info("Initial registry fetch from primary servers failed");
                }
                boolean backupFetchRegistryResult = true;
                if (!primaryFetchRegistryResult && !fetchRegistryFromBackup()) {
                    // 如果所有的eureka服务器网址都无法访问，从备份注册表中获取注册表信息也失败。
                    backupFetchRegistryResult = false;
                    // 从备份服务器初始注册表提取失败
                    logger.info("Initial registry fetch from backup servers failed");
                }
                if (!primaryFetchRegistryResult && !backupFetchRegistryResult && clientConfig.shouldEnforceFetchRegistryAtInit()) {
                    // 在启动时获取注册表错误。初始获取失败。
                    throw new IllegalStateException("Fetch registry error at startup. Initial fetch failed.");
                }
            } catch (Throwable th) {
                logger.error("Fetch registry error at startup: {}", th.getMessage());
                throw new IllegalStateException(th);
            }
        }
       ....省略n行代码......
        // 最后，初始化调度任务(例如，集群解析器、 heartbeat、 instanceInfo replicator、 fetch
        initScheduledTasks();
      ....省略n行代码......
    }

    private void initScheduledTasks() {
        // 默认true，可更改配置不建议
        if (clientConfig.shouldFetchRegistry()) {
            // registry cache refresh timer注册表缓存刷新计时器
            // 默认30
            int registryFetchIntervalSeconds = clientConfig.getRegistryFetchIntervalSeconds();
            int expBackOffBound = clientConfig.getCacheRefreshExecutorExponentialBackOffBound();
            cacheRefreshTask = new TimedSupervisorTask(
                    "cacheRefresh",
                    scheduler,
                    cacheRefreshExecutor,
                    registryFetchIntervalSeconds,
                    TimeUnit.SECONDS,
                    expBackOffBound,
                    new CacheRefreshThread()
            );
            scheduler.schedule(
                    cacheRefreshTask,
                    registryFetchIntervalSeconds, TimeUnit.SECONDS);
        }
        if (clientConfig.shouldRegisterWithEureka()) {
            /*  LeaseInfo：
                public static final int DEFAULT_LEASE_RENEWAL_INTERVAL = 30;
                // Client settings
                private int renewalIntervalInSecs = DEFAULT_LEASE_RENEWAL_INTERVAL;
             */
            // 默认30
            int renewalIntervalInSecs = instanceInfo.getLeaseInfo().getRenewalIntervalInSecs();
            int expBackOffBound = clientConfig.getHeartbeatExecutorExponentialBackOffBound();
            logger.info("Starting heartbeat executor: " + "renew interval is: {}", renewalIntervalInSecs);
            // Heartbeat timer心跳任务
            heartbeatTask = new TimedSupervisorTask(
                    "heartbeat",
                    scheduler,
                    heartbeatExecutor,
                    renewalIntervalInSecs,
                    TimeUnit.SECONDS,
                    expBackOffBound,
                    new HeartbeatThread()
            );
            // 默认的情况下会每隔30秒向注册中心 (eureka.instance.lease-renewal-interval-in-seconds)发送一次心跳来进行服务续约
            scheduler.schedule(
                    heartbeatTask,
                    renewalIntervalInSecs, TimeUnit.SECONDS);
            // InstanceInfo replicator实例信息复制任务
            instanceInfoReplicator = new InstanceInfoReplicator(
                    this,
                    instanceInfo,
                    clientConfig.getInstanceInfoReplicationIntervalSeconds(),
                    2); // burstSize
            // 状态变更监听者
            statusChangeListener = new ApplicationInfoManager.StatusChangeListener() {
                @Override
                public String getId() {
                    return "statusChangeListener";
                }
                @Override
                public void notify(StatusChangeEvent statusChangeEvent) {
                    // Saw local status change event StatusChangeEvent [timestamp=1668595102513, current=UP, previous=STARTING]
                    logger.info("Saw local status change event {}", statusChangeEvent);
                    instanceInfoReplicator.onDemandUpdate();
                }
            };
            // 初始化状态变更监听者
            if (clientConfig.shouldOnDemandUpdateStatusChange()) {
                applicationInfoManager.registerStatusChangeListener(statusChangeListener);
            }
            // 3.2 定时刷新服务实例信息和检查应用状态的变化，在服务实例信息发生改变的情况下向server重新发起注册
            instanceInfoReplicator.start(clientConfig.getInitialInstanceInfoReplicationIntervalSeconds());
        } else {
            logger.info("Not registering with Eureka server per configuration");
        }
    }

    private boolean fetchRegistry(boolean forceFullRegistryFetch) {
        Stopwatch tracer = FETCH_REGISTRY_TIMER.start();
        try {
            // If the delta is disabled or if it is the first time, get all
            // applications如果 delta 被禁用或者是第一次，那么获取所有的应用程序
            Applications applications = getApplications();
            // shouldDisableDelta默认false
            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 false: {}", clientConfig.shouldDisableDelta());
                logger.info("Single vip registry refresh property null: {}", clientConfig.getRegistryRefreshSingleVipAddress());
                logger.info("Force full registry fetch false: {}", forceFullRegistryFetch);
                logger.info("Application is null false: {}", (applications == null));
                logger.info("Registered Applications size is zero true: {}",
                        (applications.getRegisteredApplications().size() == 0));
                logger.info("Application version is -1true: {}", (applications.getVersion() == -1));
                // 全量拉取
                getAndStoreFullRegistry();
            } else {
                // 增量拉取
                getAndUpdateDelta(applications);
            }
            applications.setAppsHashCode(applications.getReconcileHashCode());
            logTotalInstances();
        } catch (Throwable e) {
            logger.info(PREFIX + "{} - was unable to refresh its cache! This periodic background refresh will be retried in {} seconds. status = {} stacktrace = {}",
                    appPathIdentifier, clientConfig.getRegistryFetchIntervalSeconds(), e.getMessage(), ExceptionUtils.getStackTrace(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;
    }

    private void getAndStoreFullRegistry() throws Throwable {
        long currentUpdateGeneration = fetchRegistryGeneration.get();
        // 从 eureka 服务器上获取所有实例注册信息
        logger.info("Getting all instance registry info from the eureka server");
        Applications apps = null;
        // RegistryRefreshSingleVipAddress默认空
        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();
        }
        // 200
        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));
            // 如：UP_1_
            logger.debug("Got full registry with apps hashcode {}", apps.getAppsHashCode());
        } else {
            logger.warn("Not updating applications as another thread is updating it already");
        }
    }

    class CacheRefreshThread implements Runnable {
        @Override
        public void run() {
            refreshRegistry();
        }
    }
    @VisibleForTesting
    void refreshRegistry() {
        try {
            boolean isFetchingRemoteRegionRegistries = isFetchingRemoteRegionRegistries();
            boolean remoteRegionsModified = false;
            // This makes sure that a dynamic change to remote regions to fetch is honored.
            // 这可以确保对要获取的远程区域的动态更改得到遵守。
            // 默认null
            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
                    // RemoteRegionsToFetch 和 AzToRegionMapper.regionsToFetch 都需要同步
                    synchronized (instanceRegionChecker.getAzToRegionMapper()) {
                        // CAS
                        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
                    // 只需刷新映射以反映任何 DNS/属性更改
                    instanceRegionChecker.getAzToRegionMapper().refreshMapping();
                }
            }
            // 刷新注册表
            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);
        }
    }

    private void getAndUpdateDelta(Applications applications) throws Throwable {
        long currentUpdateGeneration = fetchRegistryGeneration.get();
        Applications delta = null;
        // 增量查询获取
        EurekaHttpResponse<Applications> httpResponse = eurekaTransport.queryClient.getDelta(remoteRegionsRef.get());
        // 响应成功
        if (httpResponse.getStatusCode() == Status.OK.getStatusCode()) {
            delta = httpResponse.getEntity();
        }
        if (delta == null) {
            // 服务器不允许应用delta修订，因为它不安全。因此得到了完整的登记表，即转换为全量拉取
            logger.warn("The server does not allow the delta revision to be applied because it is not safe. "
                    + "Hence got the full registry.");
            getAndStoreFullRegistry();
        } else if (fetchRegistryGeneration.compareAndSet(currentUpdateGeneration, currentUpdateGeneration + 1)) {
            // CAS成功
            logger.debug("Got delta update with apps hashcode {}", delta.getAppsHashCode());
            String reconcileHashCode = "";
            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 reason出于某种原因，数量有所不同
            if (!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());
        }
    }

    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);
                    }
                    // 将实例{}添加到区域{}中的现有应用程序
                    // ceam-config:8888,ceam-auth:8005,region null
                    logger.debug("Added instance {} to the existing apps in region {}", instance.getId(), instanceRegion);
                    // 即添加到Application的instancesMap
                    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) {
                        // 删除现有应用程序的实例{}
                        logger.debug("Deleted instance {} to the existing apps ", instance.getId());
                        existingApp.removeInstance(instance);
                        /*
                         * We find all instance list from application(The status of instance status is not only the status is UP but also other status)
                         * if instance list is empty, we remove the application.
                         * 我们从应用程序中找到所有的实例列表(实例状态的状态不仅是状态是 UP，
                         * 还有其他状态)如果实例列表为空，我们删除应用程序。
                         */
                        if (existingApp.getInstancesAsIsFromEureka().isEmpty()) {
                            applications.removeApplication(existingApp);
                        }
                    }
                }
            }
        }
        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());
        }
    }

	@Override
	public List<ServiceInstance> getInstances(String serviceId) {
		// 委托eurekaClient处理
		List<InstanceInfo> infos = this.eurekaClient.getInstancesByVipAddress(serviceId,
				false);
		List<ServiceInstance> instances = new ArrayList<>();
		for (InstanceInfo info : infos) {
			instances.add(new EurekaServiceInstance(info));
		}
		return instances;
	}

    @Override
    public List<InstanceInfo> getInstancesByVipAddress(String vipAddress, boolean secure) {
        return getInstancesByVipAddress(vipAddress, secure, instanceRegionChecker.getLocalRegion());
    }
    @Override
    public List<InstanceInfo> getInstancesByVipAddress(String vipAddress, boolean secure,
                                                       @Nullable String region) {
        if (vipAddress == null) {
            throw new IllegalArgumentException(
                    "Supplied VIP Address cannot be null");
        }
        Applications applications;
        // 如果eureka:region:没有指定，则使用默认值且非空，即默认使用默认localRegion
        if (instanceRegionChecker.isLocalRegion(region)) {
            // 本地获取
            applications = this.localRegionApps.get();
        } else {
            applications = remoteRegionVsApps.get(region);
            if (null == applications) {
                logger.debug("No applications are defined for region {}, so returning an empty instance list for vip "
                        + "address {}.", region, vipAddress);
                return Collections.emptyList();
            }
        }
        // secure默认false
        if (!secure) {
            return applications.getInstancesByVirtualHostName(vipAddress);
        } else {
            return applications.getInstancesBySecureVirtualHostName(vipAddress);
        }
    }