💖 Spring家族及微服务系列文章

✨【微服务】SpringCloud微服务剔除下线源码解析

✨【微服务】SpringCloud微服务续约源码解析

✨【微服务】SpringCloud微服务注册源码解析

✨【微服务】Nacos2.x服务发现？RPC调用？重试机制？

✨【微服务】Nacos通知客户端服务变更以及重试机制

✨【微服务】Nacos服务发现源码分析

✨【微服务】SpringBoot监听器机制以及在Nacos中的应用

✨【微服务】Nacos服务端完成微服务注册以及健康检查流程

✨【微服务】Nacos客户端微服务注册原理流程

✨【微服务】SpringCloud中使用Ribbon实现负载均衡的原理

✨【微服务】SpringBoot启动流程注册FeignClient

✨【微服务】SpringBoot启动流程初始化OpenFeign的入口

✨Spring Bean的生命周期

✨Spring事务原理

✨SpringBoot自动装配原理机制及过程

✨SpringBoot获取处理器流程

✨SpringBoot中处理器映射关系注册流程

✨Spring5.x中Bean初始化流程

✨Spring中Bean定义的注册流程

✨Spring的处理器映射器与适配器的架构设计

✨SpringMVC执行流程图解及源码

目录

​编辑

💖 Spring家族及微服务系列文章

一、前言

二、轮询拉取注册表

1、构造初始化

1.1、缓存刷新

2、刷新注册表

2.1、全量拉取注册表

3、缓存刷新任务

4、增量拉取注册表

4.1、增量更新到本地缓存

三、服务发现

1、客户端获取服务实例

2、从本地列表获取

一、前言

    上一篇我们讨论了关于周期性任务的一些应用等，本篇文章我们来探究一下这些内容：周期性刷新注册表？全量拉取注册表还是增量拉取注册表、更新本地缓存？服务发现的入口、获取本地服务列表？

二、轮询拉取注册表

1、构造初始化

同样是在Spring容器初始化的过程中初始化的，基于SpringBoot自动装配集成。上一节也讲了一部分，这里补充：

@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() {....省略n行代码......}@InjectDiscoveryClient(ApplicationInfoManager applicationInfoManager, EurekaClientConfig config,AbstractDiscoveryClientOptionalArgs args,Provider backupRegistryProvider, EndpointRandomizer endpointRandomizer) {try {// default size of 2 - 1 each for heartbeat and cacheRefresh心跳和缓存刷新的默认大小分别为2-1scheduler = Executors.newScheduledThreadPool(2,new ThreadFactoryBuilder().setNameFormat("DiscoveryClient-%d").setDaemon(true).build());// 心跳执行者heartbeatExecutor = new ThreadPoolExecutor(1, clientConfig.getHeartbeatExecutorThreadPoolSize(), 0, TimeUnit.SECONDS,new SynchronousQueue(),new ThreadFactoryBuilder().setNameFormat("DiscoveryClient-HeartbeatExecutor-%d").setDaemon(true).build());  // use direct handoff// 缓存刷新执行者cacheRefreshExecutor = new ThreadPoolExecutor(1, clientConfig.getCacheRefreshExecutorThreadPoolSize(), 0, TimeUnit.SECONDS,new SynchronousQueue(),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、 fetchinitScheduledTasks();....省略n行代码......}

主要逻辑：

1. 初始化缓存刷新执行器，用于周期性执行任务，下面继续分析
2. 默认需要刷新注册表，默认不使用全量拉取，但是初始化时使用下面2分析：会调用注册中心完成注册表初始化，返回是否刷新成功；如果从主服务器初始注册表提取失败打印日志；如果所有的eureka服务器网址都无法访问，从备份注册表中获取注册表信息也失败打印日志；在启动时获取注册表错误，抛异常。
3. 可见在2中没有特殊原因的话，一般是使用全量拉取注册表初始化成功了，否则的话抛异常

1.1、缓存刷新

大部分逻辑在前面的章节已经分析，TimedSupervisorTask跟发送心跳服务续约逻辑是一样的，这里补充刷新本地服务列表任务。

    private void initScheduledTasks() {// 默认true，可更改配置不建议if (clientConfig.shouldFetchRegistry()) {// registry cache refresh timer注册表缓存刷新计时器// 默认30int 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 settingsprivate int renewalIntervalInSecs = DEFAULT_LEASE_RENEWAL_INTERVAL;*/// 默认30int 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() {@Overridepublic String getId() {return "statusChangeListener";}@Overridepublic 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");}}

主要逻辑：

1. 默认需要刷新注册表，要达到服务高可用。1）clientConfig.getRegistryFetchIntervalSeconds()获取注册表刷新时间，默认30秒，可在配置文件更改；2）初始化cacheRefreshTask为TimedSupervisorTask类型，跟心跳任务一样处理逻辑，我们这节就只分析CacheRefreshThread刷新缓存逻辑，见3

2、刷新注册表

    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默认falseif (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 truereturn true;}

主要逻辑：

1. 由上面的1.1可见状态变更监听者还没有初始化，从前面的文章也知道它的作用完成服务注册，故这里从本地获取应用就为空。所以先打印下日志，调用全量拉取注册表方法，下面分析。轨迹跟踪对象非空，关闭。
2. 在更新实例远程状态之前通知缓存刷新
3. 根据缓存中保存的刷新数据更新远程状态

2.1、全量拉取注册表

    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 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();}// 200logger.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");}}

主要逻辑：

1. RegistryRefreshSingleVipAddress默认空，故调用eurekaTransport.queryClient.getApplications(remoteRegionsRef.get())请求注册中心
2. 响应成功获取应用
3. 如果应用apps空则打印下日志；一般CAS成功，在筛选仅具有 UP 状态的实例的应用程序并对它们进行洗牌之后获取应用程序，缓存到本地localRegionApps(AtomicReference类型)；否则打印下日志

3、缓存刷新任务

    class CacheRefreshThread implements Runnable {@Overridepublic void run() {refreshRegistry();}}@VisibleForTestingvoid refreshRegistry() {try {boolean isFetchingRemoteRegionRegistries = isFetchingRemoteRegionRegistries();boolean remoteRegionsModified = false;// This makes sure that a dynamic change to remote regions to fetch is honored.// 这可以确保对要获取的远程区域的动态更改得到遵守。// 默认nullString 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()) {// CASif (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 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);}}

CacheRefreshThread 实现了Runnable接口，但是run()中任务逻辑封装了出去，在refreshRegistry()中处理。在2中分析了初始化时已经使用全量拉取注册表并缓存应用到本地localRegionApps，那么这里使用延迟任务处理的话就会执行增量拉取逻辑了，在下面4分析

4、增量拉取注册表

    private void getAndUpdateDelta(Applications applications) throws Throwable {long currentUpdateGeneration = fetchRegistryGeneration.get();Applications delta = null;// 增量查询获取EurekaHttpResponse 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());}}

主要逻辑：

1. 增量查询获取，响应成功获取数据
2. CAS成功并且加锁成功，将响应结果更新到本地，然后释放锁

4.1、增量更新到本地缓存

    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 nulllogger.debug("Added instance {} to the existing apps in region {}", instance.getId(), instanceRegion);// 即添加到Application的instancesMapapplications.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());}}

主要逻辑：

1. 遍历增量的应用数据，遍历应用中的实例
2. 从本地获取应用数据，以便更新处理
3. 如果是ADDED，则将实例添加到区域中的现有应用程序；如果是MODIFIED，则修改现有应用程序的实例；如果是DELETED，则删除现有应用程序的实例，并且从应用程序中找到所有的实例列表(实例状态的状态不仅是状态是 UP，还有其他状态)如果实例列表为空，删除应用程序。
4. 对提供的实例进行洗牌，以便它们不总是以相同的顺序返回。

三、服务发现

1、客户端获取服务实例

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

跟Nacos的入口是类似的，需要实现spring-cloud-commons的DiscoveryClient接口。这里EurekaDiscoveryClient会委托Eureka项目里面的EurekaClient处理，见下面2分析。然后将Instances列表转换为spring-cloud-commons里面的ServiceInstance类型列表。

2、从本地列表获取

    @Overridepublic List getInstancesByVipAddress(String vipAddress, boolean secure) {return getInstancesByVipAddress(vipAddress, secure, instanceRegionChecker.getLocalRegion());}@Overridepublic List getInstancesByVipAddress(String vipAddress, boolean secure,@Nullable String region) {if (vipAddress == null) {throw new IllegalArgumentException("Supplied VIP Address cannot be null");}Applications applications;// 如果eureka:region:没有指定，则使用默认值且非空，即默认使用默认localRegionif (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默认falseif (!secure) {return applications.getInstancesByVirtualHostName(vipAddress);} else {return applications.getInstancesBySecureVirtualHostName(vipAddress);}}

主要逻辑：

1. 调用重载方法，vipAddress即serviceId，secure为false
2. 如果eureka:region:没有指定，则使用默认值且非空，即默认使用默认localRegion。从本地应用获取列表。
3. secure为false，获取与虚拟主机名关联的instance列表