Eureka Client启动流程分析
@EnableDiscoveryClient注解作用
- autoRegister()方法返回true则注册到注册中心,如果你配置为false,那么就不会自动注册
- 导入EnableDiscoveryClientImportSelector类
自动装载核心配置类
根据自动装载原则可以在spring-cloud-netflix-eureka-client-2.2.5.RELEASE.jar下的META-INF目录下找到 spring.factories文件
org.springframework.boot.autoconfigure.EnableAutoConfiguration=\ org.springframework.cloud.netflix.eureka.config.EurekaClientConfigServerAutoConfiguration,\ org.springframework.cloud.netflix.eureka.config.DiscoveryClientOptionalArgsConfiguration,\ org.springframework.cloud.netflix.eureka.EurekaClientAutoConfiguration,\ org.springframework.cloud.netflix.ribbon.eureka.RibbonEurekaAutoConfiguration,\ org.springframework.cloud.netflix.eureka.EurekaDiscoveryClientConfiguration,\ org.springframework.cloud.netflix.eureka.reactive.EurekaReactiveDiscoveryClientConfiguration,\ org.springframework.cloud.netflix.eureka.loadbalancer.LoadBalancerEurekaAutoConfiguration org.springframework.cloud.bootstrap.BootstrapConfiguration=\ org.springframework.cloud.netflix.eureka.config.EurekaConfigServerBootstrapConfigurationEurekaDiscoveryClientConfigServiceBootstrapConfiguration
@ConditionalOnClass(ConfigServicePropertySourceLocator.class) @ConditionalOnProperty(value = "spring.cloud.config.discovery.enabled", matchIfMissing = false) @Configuration(proxyBeanMethods = false) @EnableConfigurationProperties public class EurekaConfigServerBootstrapConfiguration { }上方两个注解则是这个配置类是否能够开启的条件,这里就不再展开,直接看它引入的配置类吧
EurekaDiscoveryClientConfiguration
- 1、细心的读者可能会发现这里又注册了一个Marker类,可以猜测也是某个地方的开关。
- 2、EurekaClientConfigurationRefresher这个类看名字就知道这是当配置被动态刷新时的一个处理器,这里也不再展开了。
- 3、EurekaHealthCheckHandlerConfiguration这里面注册了一个Eureka健康检查的处理类。
EurekaClientAutoConfiguration
这个类里面全是重点,也是我们本文的核心
@Configuration(proxyBeanMethods = false) @EnableConfigurationProperties @ConditionalOnClass(EurekaClientConfig.class) @ConditionalOnProperty(value = "eureka.client.enabled", matchIfMissing = true) @ConditionalOnDiscoveryEnabled @AutoConfigureBefore({ NoopDiscoveryClientAutoConfiguration.class, CommonsClientAutoConfiguration.class, ServiceRegistryAutoConfiguration.class }) @AutoConfigureAfter(name = { "org.springframework.cloud.netflix.eureka.config.DiscoveryClientOptionalArgsConfiguration", "org.springframework.cloud.autoconfigure.RefreshAutoConfiguration", "org.springframework.cloud.netflix.eureka.EurekaDiscoveryClientConfiguration", "org.springframework.cloud.client.serviceregistry.AutoServiceRegistrationAutoConfiguration" }) public class EurekaClientAutoConfiguration { }首先可以看到这个类一共包含这些注解,我们来一一解析比较重要的几个注解吧
-
@Import(DiscoveryClientOptionalArgsConfiguration.class)引入了两个bean,RestTemplateDiscoveryClientOptionalArgs和MutableDiscoveryClientOptionalArgs ,这两个类的作用暂且不说。
-
@ConditionalOnBean(EurekaDiscoveryClientConfiguration.Marker.class)刚才说的Marker类的作用出来了。
-
@AutoConfigureBefore 既然必须在这三个类完成自动装配之后才能进行装配,那就代表着这三个类肯定大有用途,研究一下吧
-
NoopDiscoveryClientAutoConfiguration 故名思意,负责服务发现的类,咱们重点关注一下其中的几个方法
-
1、init()方法
这里构造了一个DefaultServiceInstance对象,这个对象包含了当前项目的ip+端口+项目名称。
- 2、注入beanNoopDiscoveryClient
这个类包含了获取当前实例以及当前服务的方法,其类图如下,在使用时肯定使用的是EurekaDiscoveryClient
EurekaClientAutoConfiguration作为自动配置类,看看它主要配置了哪些东西
配置当前实例信息
配置实例信息包含很多,不过核心的无非就是名称、唯一标识、IP地址、端口等等
@Bean @ConditionalOnMissingBean(value = EurekaInstanceConfig.class, search = SearchStrategy.CURRENT) public EurekaInstanceConfigBean eurekaInstanceConfigBean(InetUtils inetUtils, ManagementMetadataProvider managementMetadataProvider) { String hostname = getProperty("eureka.instance.hostname"); boolean preferIpAddress = Boolean .parseBoolean(getProperty("eureka.instance.prefer-ip-address")); String ipAddress = getProperty("eureka.instance.ip-address"); boolean isSecurePortEnabled = Boolean .parseBoolean(getProperty("eureka.instance.secure-port-enabled")); String serverContextPath = env.getProperty("server.servlet.context-path", "/"); int serverPort = Integer.parseInt( env.getProperty("server.port", env.getProperty("port", "8080"))); Integer managementPort = env.getProperty("management.server.port", Integer.class); String managementContextPath = env .getProperty("management.server.servlet.context-path"); Integer jmxPort = env.getProperty("com.sun.management.jmxremote.port", Integer.class); EurekaInstanceConfigBean instance = new EurekaInstanceConfigBean(inetUtils); instance.setNonSecurePort(serverPort); instance.setInstanceId(getDefaultInstanceId(env)); instance.setPreferIpAddress(preferIpAddress); instance.setSecurePortEnabled(isSecurePortEnabled); if (StringUtils.hasText(ipAddress)) { instance.setIpAddress(ipAddress); } if (isSecurePortEnabled) { instance.setSecurePort(serverPort); } if (StringUtils.hasText(hostname)) { instance.setHostname(hostname); } String statusPageUrlPath = getProperty("eureka.instance.status-page-url-path"); String healthCheckUrlPath = getProperty("eureka.instance.health-check-url-path"); if (StringUtils.hasText(statusPageUrlPath)) { instance.setStatusPageUrlPath(statusPageUrlPath); } if (StringUtils.hasText(healthCheckUrlPath)) { instance.setHealthCheckUrlPath(healthCheckUrlPath); } ManagementMetadata metadata = managementMetadataProvider.get(instance, serverPort, serverContextPath, managementContextPath, managementPort); if (metadata != null) { instance.setStatusPageUrl(metadata.getStatusPageUrl()); instance.setHealthCheckUrl(metadata.getHealthCheckUrl()); if (instance.isSecurePortEnabled()) { instance.setSecureHealthCheckUrl(metadata.getSecureHealthCheckUrl()); } Map<String, String> metadataMap = instance.getMetadataMap(); metadataMap.computeIfAbsent("management.port", k -> String.valueOf(metadata.getManagementPort())); } else { // without the metadata the status and health check URLs will not be set // and the status page and health check url paths will not include the // context path so set them here if (StringUtils.hasText(managementContextPath)) { instance.setHealthCheckUrlPath( managementContextPath + instance.getHealthCheckUrlPath()); instance.setStatusPageUrlPath( managementContextPath + instance.getStatusPageUrlPath()); } } setupJmxPort(instance, jmxPort); return instance; }负责注册的Bean
@Bean public EurekaServiceRegistry eurekaServiceRegistry() { return new EurekaServiceRegistry(); }自动注册调用的Bean
@Bean @ConditionalOnBean(AutoServiceRegistrationProperties.class) @ConditionalOnProperty( value = "spring.cloud.service-registry.auto-registration.enabled", matchIfMissing = true) public EurekaAutoServiceRegistration eurekaAutoServiceRegistration( ApplicationContext context, EurekaServiceRegistry registry, EurekaRegistration registration) { return new EurekaAutoServiceRegistration(context, registry, registration); }Eureka待注册的对象
这个对象会包含上面的eurekaInstanceIConfigBean
@Bean @ConditionalOnBean(AutoServiceRegistrationProperties.class) @ConditionalOnProperty( value = "spring.cloud.service-registry.auto-registration.enabled", matchIfMissing = true) public EurekaRegistration eurekaRegistration(EurekaClient eurekaClient, CloudEurekaInstanceConfig instanceConfig, ApplicationInfoManager applicationInfoManager, @Autowired( required = false) ObjectProvider<HealthCheckHandler> healthCheckHandler) { return EurekaRegistration.builder(instanceConfig).with(applicationInfoManager) .with(eurekaClient).with(healthCheckHandler).build(); }Eureka client配置
@Bean @ConditionalOnMissingBean(value = EurekaClientConfig.class, search = SearchStrategy.CURRENT) public EurekaClientConfigBean eurekaClientConfigBean(ConfigurableEnvironment env) { return new EurekaClientConfigBean(); }EurekaClient
@Bean(destroyMethod = "shutdown") @ConditionalOnMissingBean(value = EurekaClient.class, search = SearchStrategy.CURRENT) public EurekaClient eurekaClient(ApplicationInfoManager manager, EurekaClientConfig config) { return new CloudEurekaClient(manager, config, this.optionalArgs, this.context); }其中CloudEurekaClient是DiscoveryClient的子类,而DiscoveryClient则是EurekaClient的核心类。
new CloudEurekaClient会调用父类DiscoveryClient的构造方法
public class CloudEurekaClient extends DiscoveryClient { public CloudEurekaClient(ApplicationInfoManager applicationInfoManager, EurekaClientConfig config, AbstractDiscoveryClientOptionalArgs<?> args, ApplicationEventPublisher publisher) { //这里会调用父类DiscoveryClient的构造方法 super(applicationInfoManager, config, args); this.applicationInfoManager = applicationInfoManager; this.publisher = publisher; this.eurekaTransportField = ReflectionUtils.findField(DiscoveryClient.class, "eurekaTransport"); ReflectionUtils.makeAccessible(this.eurekaTransportField); } }在DiscoveryClient中最终会调用到@Inject注解修饰的DiscoveryClient构造方法
@Singleton public class DiscoveryClient implements EurekaClient { @Inject DiscoveryClient(ApplicationInfoManager applicationInfoManager, EurekaClientConfig config, AbstractDiscoveryClientOptionalArgs args, Provider<BackupRegistry> backupRegistryProvider, EndpointRandomizer endpointRandomizer) { if (args != null) { this.healthCheckHandlerProvider = args.healthCheckHandlerProvider; this.healthCheckCallbackProvider = args.healthCheckCallbackProvider; this.eventListeners.addAll(args.getEventListeners()); this.preRegistrationHandler = args.preRegistrationHandler; } else { this.healthCheckCallbackProvider = null; this.healthCheckHandlerProvider = null; this.preRegistrationHandler = null; } this.applicationInfoManager = applicationInfoManager; InstanceInfo myInfo = applicationInfoManager.getInfo(); clientConfig = config; staticClientConfig = clientConfig; transportConfig = config.getTransportConfig(); instanceInfo = myInfo; if (myInfo != null) { appPathIdentifier = instanceInfo.getAppName() + "/" + instanceInfo.getId(); } else { logger.warn("Setting instanceInfo to a passed in null value"); } this.backupRegistryProvider = backupRegistryProvider; this.endpointRandomizer = endpointRandomizer; this.urlRandomizer = new EndpointUtils.InstanceInfoBasedUrlRandomizer(instanceInfo); localRegionApps.set(new Applications()); fetchRegistryGeneration = new AtomicLong(0); remoteRegionsToFetch = new AtomicReference<String>(clientConfig.fetchRegistryForRemoteRegions()); remoteRegionsRef = new AtomicReference<>(remoteRegionsToFetch.get() == null ? null : remoteRegionsToFetch.get().split(",")); if (config.shouldFetchRegistry()) { this.registryStalenessMonitor = new ThresholdLevelsMetric(this, METRIC_REGISTRY_PREFIX + "lastUpdateSec_", new long[]{15L, 30L, 60L, 120L, 240L, 480L}); } else { this.registryStalenessMonitor = ThresholdLevelsMetric.NO_OP_METRIC; } if (config.shouldRegisterWithEureka()) { this.heartbeatStalenessMonitor = new ThresholdLevelsMetric(this, METRIC_REGISTRATION_PREFIX + "lastHeartbeatSec_", new long[]{15L, 30L, 60L, 120L, 240L, 480L}); } else { this.heartbeatStalenessMonitor = ThresholdLevelsMetric.NO_OP_METRIC; } logger.info("Initializing Eureka in region {}", clientConfig.getRegion()); if (!config.shouldRegisterWithEureka() && !config.shouldFetchRegistry()) { logger.info("Client configured to neither register nor query for data."); scheduler = null; heartbeatExecutor = null; cacheRefreshExecutor = null; eurekaTransport = null; instanceRegionChecker = new InstanceRegionChecker(new PropertyBasedAzToRegionMapper(config), clientConfig.getRegion()); // This is a bit of hack to allow for existing code using DiscoveryManager.getInstance() // to work with DI'd DiscoveryClient DiscoveryManager.getInstance().setDiscoveryClient(this); DiscoveryManager.getInstance().setEurekaClientConfig(config); initTimestampMs = System.currentTimeMillis(); initRegistrySize = this.getApplications().size(); registrySize = initRegistrySize; logger.info("Discovery Client initialized at timestamp {} with initial instances count: {}", initTimestampMs, initRegistrySize); return; // no need to setup up an network tasks and we are done } try { // default size of 2 - 1 each for heartbeat and cacheRefresh 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(); scheduleServerEndpointTask(eurekaTransport, args); AzToRegionMapper azToRegionMapper; if (clientConfig.shouldUseDnsForFetchingServiceUrls()) { azToRegionMapper = new DNSBasedAzToRegionMapper(clientConfig); } else { azToRegionMapper = new PropertyBasedAzToRegionMapper(clientConfig); } if (null != remoteRegionsToFetch.get()) { azToRegionMapper.setRegionsToFetch(remoteRegionsToFetch.get().split(",")); } instanceRegionChecker = new InstanceRegionChecker(azToRegionMapper, clientConfig.getRegion()); } catch (Throwable e) { throw new RuntimeException("Failed to initialize DiscoveryClient!", e); } 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()) { 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); } } // call and execute the pre registration handler before all background tasks (inc registration) is started if (this.preRegistrationHandler != null) { this.preRegistrationHandler.beforeRegistration(); } if (clientConfig.shouldRegisterWithEureka() && clientConfig.shouldEnforceRegistrationAtInit()) { try { if (!register() ) { throw new IllegalStateException("Registration error at startup. Invalid server response."); } } catch (Throwable th) { logger.error("Registration error at startup: {}", th.getMessage()); throw new IllegalStateException(th); } } // finally, init the schedule tasks (e.g. cluster resolvers, heartbeat, instanceInfo replicator, fetch //初始化调度任务(例如群集解析程序、心跳、服务实例同步、获取注册信息) initScheduledTasks(); try { Monitors.registerObject(this); } catch (Throwable e) { logger.warn("Cannot register timers", e); } // This is a bit of hack to allow for existing code using DiscoveryManager.getInstance() // to work with DI'd DiscoveryClient DiscoveryManager.getInstance().setDiscoveryClient(this); DiscoveryManager.getInstance().setEurekaClientConfig(config); initTimestampMs = System.currentTimeMillis(); initRegistrySize = this.getApplications().size(); registrySize = initRegistrySize; logger.info("Discovery Client initialized at timestamp {} with initial instances count: {}", initTimestampMs, initRegistrySize); } }initScheduledTasks()
初始化调度任务(例如群集解析程序、心跳、服务实例同步、获取注册信息)
@Singleton public class DiscoveryClient implements EurekaClient { /** * 初始化所有计划的任务 */ private void initScheduledTasks() { //获取注册信息的定时任务 if (clientConfig.shouldFetchRegistry()) { // registry cache refresh timer 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()) { 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() ); 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) { 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(); } }; if (clientConfig.shouldOnDemandUpdateStatusChange()) { applicationInfoManager.registerStatusChangeListener(statusChangeListener); } //初始化定时服务注册任务 instanceInfoReplicator.start(clientConfig.getInitialInstanceInfoReplicationIntervalSeconds()); } else { logger.info("Not registering with Eureka server per configuration"); } } }CacheRefreshThread——定时更新服务注册列表信息
@Singleton public class DiscoveryClient implements EurekaClient { class CacheRefreshThread implements Runnable { public void run() { //缓存刷新(refreshRegistry) refreshRegistry(); } } }缓存刷新(refreshRegistry)
- 系统默认是每隔30秒刷新本地存储的注册表
获取注册表——fetchRegistry
@Singleton public class DiscoveryClient implements EurekaClient { private boolean fetchRegistry(boolean forceFullRegistryFetch) { //用Stopwatch做耗时分析 Stopwatch tracer = FETCH_REGISTRY_TIMER.start(); try { // 取出本地缓存的,之前获取的服务列表信息 Applications applications = getApplications(); //判断多个条件,确定是否触发全量更新,如下任一个满足都会全量更新: //1. 是否禁用增量更新; //2. 是否对某个region特别关注; //3. 外部调用时是否通过入参指定全量更新; //4. 本地还未缓存有效的服务列表信息; 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)); //全量更新 getAndStoreFullRegistry(); } else { //增量更新 getAndUpdateDelta(applications); } //重新计算和设置一致性hash码 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(); } } //将本地缓存更新的事件广播给所有已注册的监听器,注意该方法已被CloudEurekaClient类重写 onCacheRefreshed(); //检查刚刚更新的缓存中,有来自Eureka server的服务列表,其中包含了当前应用的状态, //当前实例的成员变量lastRemoteInstanceStatus,记录的是最后一次更新的当前应用状态, //上述两种状态在updateInstanceRemoteStatus方法中作比较 ,如果不一致,就更新lastRemoteInstanceStatus,并且广播对应的事件 updateInstanceRemoteStatus(); // registry was fetched successfully, so return true return true; } }全量更新本地缓存的服务列表
- getAndStoreFullRegistry方法负责全量更新,代码如下所示,非常简单的逻辑:
getAndStoreFullRegistry方法中并无复杂逻辑,只有eurekaTransport.queryClient.getApplications(remoteRegionsRef.get())这段需要展开细看,和Eureka server交互的逻辑都在这里面,方法getApplications的具体实现是在EurekaHttpClientDecorator类:
public abstract class EurekaHttpClientDecorator implements EurekaHttpClient { @Override public EurekaHttpResponse<Applications> getApplications(final String... regions) { return execute(new RequestExecutor<Applications>() { @Override public EurekaHttpResponse<Applications> execute(EurekaHttpClient delegate) { return delegate.getApplications(regions); } @Override public RequestType getRequestType() { //本次向Eureka server请求的类型:获取服务列表 return RequestType.GetApplications; } }); } }EurekaHttpClientDecorator类从名字看是个装饰者模式的实现,看它的其他代码,发现各类远程服务都在此被封装成API了,例如注册的:
@Override public EurekaHttpResponse<Void> register(final InstanceInfo info) { return execute(new RequestExecutor<Void>() { @Override public EurekaHttpResponse<Void> execute(EurekaHttpClient delegate) { return delegate.register(info); } @Override public RequestType getRequestType() { return RequestType.Register; } }); }还有续租的:
@Override public EurekaHttpResponse<InstanceInfo> sendHeartBeat(final String appName, final String id, final InstanceInfo info, final InstanceStatus overriddenStatus) { return execute(new RequestExecutor<InstanceInfo>() { @Override public EurekaHttpResponse<InstanceInfo> execute(EurekaHttpClient delegate) { return delegate.sendHeartBeat(appName, id, info, overriddenStatus); } @Override public RequestType getRequestType() { return RequestType.SendHeartBeat; } }); }再继续追踪 delegate.register(info),进入了AbstractJerseyEurekaHttpClient类,这里面是各种网络请求的具体实现,EurekaHttpClientDecorator类中的getApplications、register、sendHeartBeat等方法对应的网络请求响应逻辑在AbstractJerseyEurekaHttpClient中都有具体实现,篇幅所限我们只关注getApplications:
public abstract class AbstractJerseyEurekaHttpClient implements EurekaHttpClient { @Override public EurekaHttpResponse<Applications> getApplications(String... regions) { //取全量数据的path是""apps" return getApplicationsInternal("apps/", regions); } @Override public EurekaHttpResponse<Applications> getDelta(String... regions) { //取增量数据的path是""apps/delta" return getApplicationsInternal("apps/delta", regions); } //具体的请求响应处理都在此方法中 private EurekaHttpResponse<Applications> getApplicationsInternal(String urlPath, String[] regions) { ClientResponse response = null; String regionsParamValue = null; try { //jersey、resource这些关键词都预示着这是个restful请求 WebResource webResource = jerseyClient.resource(serviceUrl).path(urlPath); if (regions != null && regions.length > 0) { regionsParamValue = StringUtil.join(regions); webResource = webResource.queryParam("regions", regionsParamValue); } Builder requestBuilder = webResource.getRequestBuilder(); addExtraHeaders(requestBuilder); //发起网络请求,将响应封装成ClientResponse实例 response = requestBuilder.accept(MediaType.APPLICATION_JSON_TYPE).get(ClientResponse.class); Applications applications = null; if (response.getStatus() == Status.OK.getStatusCode() && response.hasEntity()) { //取得全部应用信息 applications = response.getEntity(Applications.class); } return anEurekaHttpResponse(response.getStatus(), Applications.class) .headers(headersOf(response)) .entity(applications) .build(); } finally { if (logger.isDebugEnabled()) { logger.debug("Jersey HTTP GET {}/{}?{}; statusCode={}", serviceUrl, urlPath, regionsParamValue == null ? "" : "regions=" + regionsParamValue, response == null ? "N/A" : response.getStatus() ); } if (response != null) { response.close(); } } } }上述代码中,利用jersey-client库的API向Eureka server发起restful请求,并将响应数据封装到EurekaHttpResponse实例中返回;
小结:获取全量数据,是通过jersey-client库的API向Eureka server发起restful请求实现的,并将响应的服务列表数据放在一个成员变量中作为本地缓存;
获取服务列表信息的增量更新
获取服务列表信息的增量更新是通过getAndUpdateDelta方法完成的,具体分析请看下面的中文注释:
@Singleton public class DiscoveryClient implements EurekaClient { private void getAndUpdateDelta(Applications applications) throws Throwable { long currentUpdateGeneration = fetchRegistryGeneration.get(); Applications delta = null; //增量信息是通过eurekaTransport.queryClient.getDelta方法完成的 EurekaHttpResponse<Applications> httpResponse = eurekaTransport.queryClient.getDelta(remoteRegionsRef.get()); if (httpResponse.getStatusCode() == Status.OK.getStatusCode()) { //delta中保存了Eureka server返回的增量更新 delta = httpResponse.getEntity(); } if (delta == null) { logger.warn("The server does not allow the delta revision to be applied because it is not safe. " + "Hence got the full registry."); //如果增量信息为空,就直接发起一次全量更新 getAndStoreFullRegistry(); } //考虑到多线程同步问题,这里通过CAS来确保请求发起到现在是线程安全的, //如果这期间fetchRegistryGeneration变了,就表示其他线程也做了类似操作,因此放弃本次响应的数据 else if (fetchRegistryGeneration.compareAndSet(currentUpdateGeneration, currentUpdateGeneration + 1)) { logger.debug("Got delta update with apps hashcode {}", delta.getAppsHashCode()); String reconcileHashCode = ""; if (fetchRegistryUpdateLock.tryLock()) { try { //用Eureka返回的增量数据和本地数据做合并操作,这个方法稍后会细说 updateDelta(delta); //用合并了增量数据之后的本地数据来生成一致性哈希码 reconcileHashCode = getReconcileHashCode(applications); } finally { fetchRegistryUpdateLock.unlock(); } } else { logger.warn("Cannot acquire update lock, aborting getAndUpdateDelta"); } //Eureka server在返回增量更新数据时,也会返回服务端的一致性哈希码, //理论上每次本地缓存数据经历了多次增量更新后,计算出的一致性哈希码应该是和服务端一致的, //如果发现不一致,就证明本地缓存的服务列表信息和Eureka server不一致了,需要做一次全量更新 if (!reconcileHashCode.equals(delta.getAppsHashCode()) || clientConfig.shouldLogDeltaDiff()) { //一致性哈希码不同,就在reconcileAndLogDifference方法中做全量更新 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()); } } }上述代码中有几处需要注意:
- a、获取增量更新数据使用的方法是:eurekaTransport.queryClient.getDelta(remoteRegionsRef.get());
- b、将增量更新的数据和本地缓存合并的方法是: updateDelta(delta);
- c、通过检查一致性哈希码可以确定历经每一次增量更新后,本地的服务列表信息和Eureka server上的是否还保持一致,若不一致就要做一次全量更新,通过调用reconcileAndLogDifference方法来完成;
上述a、b、c三点,接下来依次展开:
- 1、向Eureka server发起网络请求的逻辑和前面全量更新的差不多,也是EurekaHttpClientDecorator和AbstractJerseyEurekaHttpClient这两个类合作实现的,先看EurekaHttpClientDecorator部分:
- 2、再看AbstractJerseyEurekaHttpClient类中的getDelta方法,居然和全量获取服务列表数据调用了相同的方法getApplicationsInternal,只是ur参数不一样而已;
由上述代码可见,从Eureka server的获取增量更新,和一些常见的方式略有区别:
-
a、一般的增量更新是在请求中增加一个时间戳或者上次更新的tag号等参数,由服务端根据参数来判断哪些数据是客户端没有的;
-
b、而这里的Eureka client却没有这类参数,联想到前面官方文档中提到的“Eureka会把更新数据保留三分钟”,就可以理解了:Eureka把最近的变更数据保留三分钟,这三分钟内每个Eureka client来请求增量更新时,server都返回同样的缓存数据,只要client能保证三分钟之内有一次请求,就能保证自己的数据和Eureka server端的保持一致;
-
c、那么如果client有问题,导致超过三分钟才来获取增量更新数据,那就有可能client和server数据不一致了,此时就要有一种方式来判断是否不一致,如果不一致,client就会做一次全量更新,这种判断就是一致性哈希码;
-
3、Eureka client获取到增量更新后,通过updateDelta方法将增量更新数据和本地数据做合并:
上述代码有几点需要注意:
- a、检查每个服务的region,如果跨region的,就合并到另一个专门存放跨region服务的缓存中;
- b、增量数据中,对每个应用下实例的变动,分为新增、修改、删除三种,合并的过程就是对这三种数据在本地缓存中做不同的处理;
- c、合并过程中还会对缓存数据做整理,这样后续每次使用时,获取的多个实例其顺序是一样的;
前面曾经提到,如果Eureka client不及时做增量更新,那么有可能会错过Eureka server上的数据变化,导致两边的服务列表信息不一致,这个问题会通过一致性哈希码对比发现,发现后如何处理呢?先看增量更新的getAndUpdateDelta方法中的一个注释,如下图红框所示,个人觉得这个注释写得很好,内容既简洁又重要:
上图红框中提醒:此处会发生一次远程调用,这说明发现Eureka server和Eureka client保存的服务列表数据不一致时会向Eureka server发起一次请求,打开reconcileAndLogDifference方法看详情:
@Singleton public class DiscoveryClient implements EurekaClient { private void reconcileAndLogDifference(Applications delta, String reconcileHashCode) throws Throwable { logger.debug("The Reconcile hashcodes do not match, client : {}, server : {}. Getting the full registry", reconcileHashCode, delta.getAppsHashCode()); RECONCILE_HASH_CODES_MISMATCH.increment(); long currentUpdateGeneration = fetchRegistryGeneration.get(); //从Eureka server获取全量数据 EurekaHttpResponse<Applications> httpResponse = clientConfig.getRegistryRefreshSingleVipAddress() == null ? eurekaTransport.queryClient.getApplications(remoteRegionsRef.get()) : eurekaTransport.queryClient.getVip(clientConfig.getRegistryRefreshSingleVipAddress(), remoteRegionsRef.get()); Applications serverApps = httpResponse.getEntity(); if (serverApps == null) { logger.warn("Cannot fetch full registry from the server; reconciliation failure"); return; } //CAS成功就把全量数据更新到本地缓存中 if (fetchRegistryGeneration.compareAndSet(currentUpdateGeneration, currentUpdateGeneration + 1)) { localRegionApps.set(this.filterAndShuffle(serverApps)); getApplications().setVersion(delta.getVersion()); logger.debug( "The Reconcile hashcodes after complete sync up, client : {}, server : {}.", getApplications().getReconcileHashCode(), delta.getAppsHashCode()); } else { logger.warn("Not setting the applications map as another thread has advanced the update generation"); } } }上述代码较简单:从Eureka server获取全量数据,再尝试CAS,如果成功就更新本地缓存数据;
至此,全量和增量更新的源码都看过了,接下来看看更新完数据后的两次广播:更新缓存和状态变化(有变化才广播);
广播:更新缓存
更新缓存的广播是在onCacheRefreshed方法中执行的,该方法在扩展类CloudEurekaClient中被覆盖:
public class CloudEurekaClient extends DiscoveryClient { @Override protected void onCacheRefreshed() { super.onCacheRefreshed(); if (this.cacheRefreshedCount != null) { // might be called during construction and // will be null long newCount = this.cacheRefreshedCount.incrementAndGet(); log.trace("onCacheRefreshed called with count: " + newCount); //spring容器内的广播 this.publisher.publishEvent(new HeartbeatEvent(this, newCount)); } } }上述代码显示,这是个spring容器内的广播,this.publisher的类型是ApplicationEventPublisher。
广播:本地状态变化
从Eureka server中取得的服务列表,自然也包括当前应用自己的信息,这个信息会保存在成员变量lastRemoteInstanceStatus中,每次更新了缓存后,都会用缓存中的信息和lastRemoteInstanceStatus对比,如果不一致,就表示在Eureka server端记录的当前应用状态发生了变化,此时就广播一次;
@Singleton public class DiscoveryClient implements EurekaClient { private synchronized void updateInstanceRemoteStatus() { // Determine this instance's status for this app and set to UNKNOWN if not found InstanceInfo.InstanceStatus currentRemoteInstanceStatus = null; if (instanceInfo.getAppName() != null) { Application app = getApplication(instanceInfo.getAppName()); if (app != null) { InstanceInfo remoteInstanceInfo = app.getByInstanceId(instanceInfo.getId()); if (remoteInstanceInfo != null) { currentRemoteInstanceStatus = remoteInstanceInfo.getStatus(); } } } if (currentRemoteInstanceStatus == null) { currentRemoteInstanceStatus = InstanceInfo.InstanceStatus.UNKNOWN; } // Notify if status changed if (lastRemoteInstanceStatus != currentRemoteInstanceStatus) { //这里发起广播 onRemoteStatusChanged(lastRemoteInstanceStatus, currentRemoteInstanceStatus); lastRemoteInstanceStatus = currentRemoteInstanceStatus; } } }最准确的说明信息来自Netflix的官方文档,地址:https://github.com/Netflix/eureka/wiki/Understanding-eureka-client-server-communication#fetch-registry
小结:
- 1、官方文档对整个过程做了准确的总结,围绕着这些总结去看代码,能够事半功倍,重要是整个过程都保持的正确的方向,不会由于细节的干扰而偏离主线;
- 2、Eureka的注册中心设计,尽管多个client轮询请求会增加服务器压力,但使用增量更新再加上Server自身缓存3分钟数据的方式,可以有效的减少数据量和相关的计算,再加上一致性哈希码来弥补增量更新的弊端,在性能和完整性方面都有了保证,另外增量更新不需要client的时间戳,这样既节省性能又简化了实现逻辑,这种设计方式值得我们学习;
HeartbeatThread——服务定时续约线程
HeartbeatThread类中,通过调用renew方法实现续租,如下代码所示,方法注释已说明是Restfult请求来实现的,对应Eureka server的返回信息httpResponse,除了检查返回码是否等于200就没有任何作用了,想想也是如此,30秒一次的心跳,不论是请求还是响应都应该尽量简洁,降低服务器和网络的压力:
@Singleton public class DiscoveryClient implements EurekaClient { private class HeartbeatThread implements Runnable { public void run() { if (renew()) { lastSuccessfulHeartbeatTimestamp = System.currentTimeMillis(); } } } //服务续约 boolean renew() { EurekaHttpResponse<InstanceInfo> httpResponse; try { //发Restful请求,即心跳 httpResponse = eurekaTransport.registrationClient.sendHeartBeat(instanceInfo.getAppName(), instanceInfo.getId(), instanceInfo, null); logger.debug(PREFIX + "{} - Heartbeat status: {}", appPathIdentifier, httpResponse.getStatusCode()); //404错误会触发注册逻辑 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; } //返回码200表示心跳成功 return httpResponse.getStatusCode() == Status.OK.getStatusCode(); } catch (Throwable e) { logger.error(PREFIX + "{} - was unable to send heartbeat!", appPathIdentifier, e); return false; } } }继续展开上面代码段中的 eurekaTransport.registrationClient.sendHeartBeat方法,源码在EurekaHttpClientDecorator类中:
@Override public EurekaHttpResponse<InstanceInfo> sendHeartBeat(final String appName, final String id, final InstanceInfo info, final InstanceStatus overriddenStatus) { return execute(new RequestExecutor<InstanceInfo>() { @Override public EurekaHttpResponse<InstanceInfo> execute(EurekaHttpClient delegate) { //网络处理委托给代理类完成 return delegate.sendHeartBeat(appName, id, info, overriddenStatus); } @Override public RequestType getRequestType() { //请求类型为心跳 return RequestType.SendHeartBeat; } }); }继续展开delegate.sendHeartBeat,多层调用一路展开,最终由JerseyApplicationClient类来完成操作,对应源码在父类AbstractJerseyEurekaHttpClient中,如下所示,主要工作是利用jersey库的Restful Api将自身的信息PUT到Eureka server,注意:这里不是POST,也不是GET,而是PUT:
@Override public EurekaHttpResponse<InstanceInfo> sendHeartBeat(String appName, String id, InstanceInfo info, InstanceStatus overriddenStatus) { String urlPath = "apps/" + appName + '/' + id; ClientResponse response = null; try { //请求参数有两个:Eureka client自身状态、自身关键信息(状态、元数据等)最后一次变化的时间 WebResource webResource = jerseyClient.resource(serviceUrl) .path(urlPath) .queryParam("status", info.getStatus().toString()) .queryParam("lastDirtyTimestamp", info.getLastDirtyTimestamp().toString()); if (overriddenStatus != null) { webResource = webResource.queryParam("overriddenstatus", overriddenStatus.name()); } Builder requestBuilder = webResource.getRequestBuilder(); addExtraHeaders(requestBuilder); //注意:这里不是POST,也不是GET,而是PUT response = requestBuilder.put(ClientResponse.class); EurekaHttpResponseBuilder<InstanceInfo> eurekaResponseBuilder = anEurekaHttpResponse(response.getStatus(), InstanceInfo.class).headers(headersOf(response)); if (response.hasEntity() && !HTML.equals(response.getType().getSubtype())) { //don't try and deserialize random html errors from the server eurekaResponseBuilder.entity(response.getEntity(InstanceInfo.class)); } return eurekaResponseBuilder.build(); } finally { if (logger.isDebugEnabled()) { logger.debug("Jersey HTTP PUT {}/{}; statusCode={}", serviceUrl, urlPath, response == null ? "N/A" : response.getStatus()); } if (response != null) { response.close(); } } }至此,Eureka client向服务续租的源码就分析完毕了,过程相对简单,DiscoveryClient、TimedSupervisorTask、JerseyApplicationClient等实例各司其职,定时发送PUT请求到Eureka server。
最准确的说明信息来自Netflix的官方文档,地址:https://github.com/Netflix/eureka/wiki/Understanding-eureka-client-server-communication#renew
参考: https://www.cnblogs.com/zhixiang-org-cn/p/11689212.html
https://blog.csdn.net/u010647035/article/details/83245433
https://xinchen.blog.csdn.net/article/details/82915355
https://blog.csdn.net/boling_cavalry/article/details/82813180