【SpringCloud】Eureka源碼解析上

Eureka是一個服務發現與注冊組件，它包含服務端和客戶端，服務端管理服務的注冊信息，客戶端簡化服務實例與服務端的交互。我們結合源碼來分析下eureka組件的實現原理，內容分為上下兩章，第一章分析eureka的服務注冊，第二章分析eureka的心跳機制，本章節是第一章。

參考源碼：<spring-cloud.version>Hoxton.SR9</spring-cloud.version>

往期系列：

【SpringBoot】SpringBoot源碼解析第一章 SpringBoot的構造方法-CSDN博客

【SpringBoot】SpringBoot源碼解析第二章 SpringBoot的run方法-CSDN博客

【SpringBoot】SpringBoot源碼解析第三章 SpringBoot的自動化配置-CSDN博客

【SpringBoot】SpringBoot源碼解析第四章 SpringBoot的bean接口-CSDN博客

【SpringBoot】SpringBoot源碼解析第五章 SpringBoot的beanDefinition收集過程-CSDN博客

【SpringBoot】SpringBoot源碼解析第六章 SpringBoot的getBean方法-CSDN博客

【SpringBoot】SpringBoot源碼解析第七章 SpringBoot的感悟-CSDN博客

1、注冊服務

1.1 服務端接收注冊信息

spring-cloud-netflix-eureka-server依賴包下有一個spring.factories文件，文件內容如下

org.springframework.boot.autoconfigure.EnableAutoConfiguration=\org.springframework.cloud.netflix.eureka.server.EurekaServerAutoConfiguration

根據springboot自動配置的原理可知，EurekaServerAutoConfiguration會被標記成了一個自動配置類。EurekaServerAutoConfiguration配置類中有一個jerseyApplication方法，這個方法會收集指定包下被Path或Provider注解標記的類的beanDefinition，這些類可以看作是Controller

// 掃描包路徑
private static final String[] EUREKA_PACKAGES = new String[]{"com.netflix.discovery", "com.netflix.eureka"};    // 收集包下指定類的beanDefinition，放入application對象
@Bean
public Application jerseyApplication(Environment environment, ResourceLoader resourceLoader) {ClassPathScanningCandidateComponentProvider provider = new ClassPathScanningCandidateComponentProvider(false, environment);// 收集的對象要求被Path或Provider注解標記provider.addIncludeFilter(new AnnotationTypeFilter(Path.class));provider.addIncludeFilter(new AnnotationTypeFilter(Provider.class));String[] var5 = EUREKA_PACKAGES;int var6 = var5.length;for(int var7 = 0; var7 < var6; ++var7) {String basePackage = var5[var7];// 掃描包路徑，收集beanDefinitionSet<BeanDefinition> beans = provider.findCandidateComponents(basePackage);Iterator var10 = beans.iterator();while(var10.hasNext()) {BeanDefinition bd = (BeanDefinition)var10.next();Class<?> cls = ClassUtils.resolveClassName(bd.getBeanClassName(), resourceLoader.getClassLoader());classes.add(cls);}}...return rc;
}// 獲取到application，將beanDefinition置入servlet容器
@Bean
public FilterRegistrationBean<?> jerseyFilterRegistration(Application eurekaJerseyApp) {FilterRegistrationBean<Filter> bean = new FilterRegistrationBean();bean.setFilter(new ServletContainer(eurekaJerseyApp));bean.setOrder(Integer.MAX_VALUE);bean.setUrlPatterns(Collections.singletonList("/eureka/*"));return bean;
}

收集的beanDefinition會通過jerseyFilterRegistration方法放入servlet容器，這樣接收請求時就能通過url映射給指定的bean來處理請求

com.netflix.eureka包下被掃描的類如下：

ApplicationResource類是Controller中的一員，它有一個addInstance方法，這個方法就是服務端響應服務注冊的方法

@POST
@Consumes({"application/json", "application/xml"})
public Response addInstance(InstanceInfo info, @HeaderParam("x-netflix-discovery-replication") String isReplication) {...// 執行注冊this.registry.register(info, "true".equals(isReplication));return Response.status(204).build();
}

調用鏈：
-> ApplicationResource.addInstance
-> InstanceRegistry.register
-> PeerAwareInstanceRegistryImpl.register
-> AbstractInstanceRegistry.register

服務端使用currentHashMap來存儲服務的信息，服務端響應注冊的過程較為簡單

// 用currentHashMap存儲服務信息 
private final ConcurrentHashMap<String, Map<String, Lease<InstanceInfo>>> registry= new ConcurrentHashMap();    public void register(InstanceInfo registrant, int leaseDuration, boolean isReplication)   
{...Map<String, Lease<InstanceInfo>> gMap = (Map)this.registry.get(registrant.getAppName());Lease<InstanceInfo> lease = new Lease(registrant, leaseDuration);if (existingLease != null) {lease.setServiceUpTimestamp(existingLease.getServiceUpTimestamp());}// 將服務信息放入map中((Map)gMap).put(registrant.getId(), lease);...
}

1.2 客戶端發送注冊信息

1.2.1 client客戶端

spring-cloud-netflix-eureka-client依賴包下也有一個spring.factories文件，文件內容如下

...
org.springframework.cloud.netflix.eureka.EurekaClientAutoConfiguration
...

EurekaClientAutoConfiguration被標記成自動配置類，它里面有一個創建EurekaClient類對象的bean方法，看類的名稱我們知道這是一個客戶端

@Bean(destroyMethod = "shutdown"
)
@ConditionalOnMissingBean(value = {EurekaClient.class},search = SearchStrategy.CURRENT
)
@org.springframework.cloud.context.config.annotation.RefreshScope
@Lazy
public EurekaClient eurekaClient(ApplicationInfoManager manager, EurekaClientConfig config, EurekaInstanceConfig instance, @Autowired(required = false) HealthCheckHandler healthCheckHandler) {ApplicationInfoManager appManager;if (AopUtils.isAopProxy(manager)) {appManager = (ApplicationInfoManager)ProxyUtils.getTargetObject(manager);} else {appManager = manager;}// 創建客戶端CloudEurekaClient cloudEurekaClient = new CloudEurekaClient(appManager, config, this.optionalArgs, this.context);cloudEurekaClient.registerHealthCheck(healthCheckHandler);return cloudEurekaClient;
}

調用鏈：
-> EurekaAutoServiceRegistration.eurekaClient
-> new CloudEurekaClient(appManager, config, this.optionalArgs, this.context);
-> CloudEurekaClient.super(applicationInfoManager, config, args);
-> DiscoveryClient.DiscoveryClient... 構造方法重載-> DiscoveryClient.DiscoveryClient
-> initScheduledTasks

跟蹤EurekaClient類的構造方法找到DiscoveryClient類，DiscoveryClient類的構造方法調用了initScheduledTasks方法，初始化了一個定時任務

private void initScheduledTasks() {...// 添加狀態變更監聽器this.statusChangeListener = new ApplicationInfoManager.StatusChangeListener() {public String getId() {return "statusChangeListener";}public void notify(StatusChangeEvent statusChangeEvent) {if (statusChangeEvent.getStatus() == InstanceStatus.DOWN) {DiscoveryClient.logger.error("Saw local status change event {}", statusChangeEvent);} else {DiscoveryClient.logger.info("Saw local status change event {}", statusChangeEvent);}// 監聽器被通知后調用onDemandUpdate方法DiscoveryClient.this.instanceInfoReplicator.onDemandUpdate();}};...}

定時任務內添加了一個狀態修改監聽器，監聽器調用notify方法時會回調onDemandUpdate方法，追蹤這個回調方法

調用鏈：
-> InstanceInfoReplicator.onDemandUpdate
-> InstanceInfoReplicator.this.run
-> this.discoveryClient.register
-> this.eurekaTransport.registrationClient.register(this.instanceInfo)
-> AbstractJerseyEurekaHttpClient.register

進入到AbstractJerseyEurekaHttpClient類的register方法

public EurekaHttpResponse<Void> register(InstanceInfo info) {String urlPath = "apps/" + info.getAppName();ClientResponse response = null;EurekaHttpResponse var5;try {// 向注冊中心發送http請求WebResource.Builder resourceBuilder = this.jerseyClient.resource(this.serviceUrl).path(urlPath).getRequestBuilder();this.addExtraHeaders(resourceBuilder);response = (ClientResponse)((WebResource.Builder)((WebResource.Builder)((WebResource.Builder)resourceBuilder.header("Accept-Encoding", "gzip")).type(MediaType.APPLICATION_JSON_TYPE)).accept(new String[]{"application/json"})).post(ClientResponse.class, info);var5 = EurekaHttpResponse.anEurekaHttpResponse(response.getStatus()).headers(headersOf(response)).build();} finally {if (logger.isDebugEnabled()) {logger.debug("Jersey HTTP POST {}/{} with instance {}; statusCode={}", new Object[]{this.serviceUrl, urlPath, info.getId(), response == null ? "N/A" : response.getStatus()});}if (response != null) {response.close();}}return var5;
}

注冊方法寫得很直白了：客戶端拿到注冊中心地址，然后攜帶服務元數據，發送請求完成注冊。不過還有一個問題，之前我們提到定時任務內初始化了一個監聽器，這個監聽器只有被通知了才會執行后續的注冊方法，那么監聽器是如何被通知的？它的觸發時機又在何時？

1.2.2?監聽器

EurekaClientAutoConfiguration配置類還有一個創建EurekaAutoServiceRegistration類的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);
}

EurekaAutoServiceRegistration類實現了SmartLifecycle接口。當spring容器加載完所有bean后會調用SmartLifeCycle接口實現類的start方法，start方法調用EurekaServiceRegistry類的regiser方法

public class EurekaAutoServiceRegistration implements AutoServiceRegistration, SmartLifecycle, Ordered, SmartApplicationListener 
{public void start() {if (this.port.get() != 0) {if (this.registration.getNonSecurePort() == 0) {this.registration.setNonSecurePort(this.port.get());}if (this.registration.getSecurePort() == 0 && this.registration.isSecure()) {this.registration.setSecurePort(this.port.get());}}if (!this.running.get() && this.registration.getNonSecurePort() > 0) {// 調用EurekaServiceRegistry的regiserthis.serviceRegistry.register(this.registration);this.context.publishEvent(new InstanceRegisteredEvent(this, this.registration.getInstanceConfig()));this.running.set(true);}}
}

EurekaServiceRegistry類的regiser方法會設置實例的狀態。進入ApplicationInfoManager類的setInstanceStatus方法

// 設置實例狀態        
reg.getApplicationInfoManager().setInstanceStatus(reg.getInstanceConfig().getInitialStatus());

setInstanceStatus方法觸發了一個狀態修改事件，并且通知了監聽器

public synchronized void setInstanceStatus(InstanceInfo.InstanceStatus status) {InstanceInfo.InstanceStatus next = this.instanceStatusMapper.map(status);if (next != null) {InstanceInfo.InstanceStatus prev = this.instanceInfo.setStatus(next);if (prev != null) {Iterator var4 = this.listeners.values().iterator();while(var4.hasNext()) {StatusChangeListener listener = (StatusChangeListener)var4.next();try {// 通知監聽器listener.notify(new StatusChangeEvent(prev, next));} catch (Exception var7) {logger.warn("failed to notify listener: {}", listener.getId(), var7);}}}}
}

這里的監聽器和上面提到的狀態修改監聽器其實是同一個監聽器，在調用EurekaAutoServiceRegistration對象的start方法后，監聽器會收到通知然后調用客戶端的register方法，這就是發送注冊服務請求的執行時機

2、拉取服務

2.1 初次拉取

客戶端第一次拉取服務和DiscoveryClient類的構造方法有關，詳情如下：

@Inject
DiscoveryClient(...){...// 調用fetchRegistry方法，拉取服務boolean primaryFetchRegistryResult = this.fetchRegistry(false);if (!primaryFetchRegistryResult) {logger.info("Initial registry fetch from primary servers failed");}...
}  private boolean fetchRegistry(boolean forceFullRegistryFetch) {...// 調用getAndStoreFullRegistry方法，拉取全部服務this.getAndStoreFullRegistry();...
}private void getAndStoreFullRegistry() throws Throwable {...long currentUpdateGeneration = this.fetchRegistryGeneration.get();// 啟動時會打印這行日志logger.info("Getting all instance registry info from the eureka server");Applications apps = null;// 發送http請求EurekaHttpResponse<Applications> httpResponse = this.clientConfig.getRegistryRefreshSingleVipAddress() == null ? this.eurekaTransport.queryClient.getApplications((String[])this.remoteRegionsRef.get()) : this.eurekaTransport.queryClient.getVip(this.clientConfig.getRegistryRefreshSingleVipAddress(), (String[])this.remoteRegionsRef.get());if (httpResponse.getStatusCode() == Status.OK.getStatusCode()) {apps = (Applications)httpResponse.getEntity();}...
}

2.2 定時拉取

為了保證服務信息真實可信，客戶端會定時拉取遠程注冊列表更新本地數據。提到到定時任務，自然的聯想到DiscoveryClient類的initScheduledTasks方法（1.2.1的內容）

private void initScheduledTasks() {int renewalIntervalInSecs;int expBackOffBound;if (this.clientConfig.shouldFetchRegistry()) {renewalIntervalInSecs = this.clientConfig.getRegistryFetchIntervalSeconds();expBackOffBound = this.clientConfig.getCacheRefreshExecutorExponentialBackOffBound();// 定時刷新本地服務列表任務，具體任務在CacheRefreshThread內this.cacheRefreshTask = new TimedSupervisorTask("cacheRefresh", this.scheduler, this.cacheRefreshExecutor, renewalIntervalInSecs, TimeUnit.SECONDS, expBackOffBound, new // 執行任務的線程CacheRefreshThread());this.scheduler.schedule(this.cacheRefreshTask, (long)renewalIntervalInSecs, TimeUnit.SECONDS);}
}class CacheRefreshThread implements Runnable {CacheRefreshThread() {}public void run() {// 刷新服務列表DiscoveryClient.this.refreshRegistry();}
}@VisibleForTesting
void refreshRegistry() {...// 獲取服務列表boolean success = this.fetchRegistry(remoteRegionsModified);...
}