Spring Cloud全解析：注册中心之Eureka服务获取和服务续约

服务获取和服务续约

eureka客户端通过定时任务的方式进行服务获取和服务续约，在com.netflix.discovery.DiscoveryClient类中，启动了两个定时任务来进行处理

private void initScheduledTasks() {// 是否需要拉取if (clientConfig.shouldFetchRegistry()) {// registry cache refresh timerint registryFetchIntervalSeconds = clientConfig.getRegistryFetchIntervalSeconds();int expBackOffBound = clientConfig.getCacheRefreshExecutorExponentialBackOffBound();// 拉取的定时任务时间间隔是registryFetchIntervalSecondsscheduler.schedule(new TimedSupervisorTask("cacheRefresh",scheduler,cacheRefreshExecutor,registryFetchIntervalSeconds,TimeUnit.SECONDS,expBackOffBound,new CacheRefreshThread()),registryFetchIntervalSeconds, TimeUnit.SECONDS);}
// 是否需要注册if (clientConfig.shouldRegisterWithEureka()) {int renewalIntervalInSecs = instanceInfo.getLeaseInfo().getRenewalIntervalInSecs();int expBackOffBound = clientConfig.getHeartbeatExecutorExponentialBackOffBound();logger.info("Starting heartbeat executor: " + "renew interval is: {}", renewalIntervalInSecs);// Heartbeat timerscheduler.schedule(new TimedSupervisorTask("heartbeat",scheduler,heartbeatExecutor,renewalIntervalInSecs,TimeUnit.SECONDS,expBackOffBound,new HeartbeatThread()),renewalIntervalInSecs, TimeUnit.SECONDS);// InstanceInfo replicatorinstanceInfoReplicator = new InstanceInfoReplicator(this,instanceInfo,clientConfig.getInstanceInfoReplicationIntervalSeconds(),2); // burstSizestatusChangeListener = new ApplicationInfoManager.StatusChangeListener() {@Overridepublic String getId() {return "statusChangeListener";}@Overridepublic void notify(StatusChangeEvent statusChangeEvent) {if (InstanceStatus.DOWN == statusChangeEvent.getStatus() ||InstanceStatus.DOWN == statusChangeEvent.getPreviousStatus()) {// log at warn level if DOWN was involvedlogger.warn("Saw local status change event {}", statusChangeEvent);} else {logger.info("Saw local status change event {}", statusChangeEvent);}instanceInfoReplicator.onDemandUpdate();}};if (clientConfig.shouldOnDemandUpdateStatusChange()) {applicationInfoManager.registerStatusChangeListener(statusChangeListener);}instanceInfoReplicator.start(clientConfig.getInitialInstanceInfoReplicationIntervalSeconds());} else {logger.info("Not registering with Eureka server per configuration");}
}

来分别看一下两个定时逻辑

服务获取

线程是CacheRefreshThread，看一下run方法

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// applicationsApplications 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{getAndStoreFullRegistry();} else {getAndUpdateDelta(applications);}applications.setAppsHashCode(applications.getReconcileHashCode());logTotalInstances();} catch (Throwable e) {logger.error(PREFIX + "{} - was unable to refresh its cache! status = {}", appPathIdentifier, 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 cacheupdateInstanceRemoteStatus();// registry was fetched successfully, so return truereturn true;}

缓存刷新逻辑

if (event instanceof CacheRefreshedEvent) {if (!updateQueued.compareAndSet(false, true)) {  // if an update is already queuedlogger.info("an update action is already queued, returning as no-op");return;}if (!refreshExecutor.isShutdown()) {try {refreshExecutor.submit(new Runnable() {@Overridepublic void run() {try {// 更新负载均衡器中的mapupdateAction.doUpdate();lastUpdated.set(System.currentTimeMillis());} catch (Exception e) {logger.warn("Failed to update serverList", e);} finally {updateQueued.set(false);}}});  // fire and forget} catch (Exception e) {logger.warn("Error submitting update task to executor, skipping one round of updates", e);updateQueued.set(false);  // if submit fails, need to reset updateQueued to false}}else {logger.debug("stopping EurekaNotificationServerListUpdater, as refreshExecutor has been shut down");stop();}}

服务续约

线程是HeartbeatThread，看一下run方法

boolean renew() {EurekaHttpResponse<InstanceInfo> httpResponse;try {// 直接发送请求进行续约，很简单的逻辑httpResponse = eurekaTransport.registrationClient.sendHeartBeat(instanceInfo.getAppName(), instanceInfo.getId(), instanceInfo, null);logger.debug(PREFIX + "{} - Heartbeat status: {}", appPathIdentifier, httpResponse.getStatusCode());if (httpResponse.getStatusCode() == Status.NOT_FOUND.getStatusCode()) {REREGISTER_COUNTER.increment();logger.info(PREFIX + "{} - Re-registering apps/{}", appPathIdentifier, instanceInfo.getAppName());long timestamp = instanceInfo.setIsDirtyWithTime();boolean success = register();if (success) {instanceInfo.unsetIsDirty(timestamp);}return success;}return httpResponse.getStatusCode() == Status.OK.getStatusCode();} catch (Throwable e) {logger.error(PREFIX + "{} - was unable to send heartbeat!", appPathIdentifier, e);return false;}
}

https://zhhll.icu/2023/框架/微服务/springcloud/注册中心/Eureka/源码分析/1.服务获取和服务续约/