spring cloud eureka client原始碼解析
阿新 • • 發佈：2018-11-08
前言
eureka是springcloud 常用的註冊中心，這裡簡單介紹下，eureka client註冊邏輯的實現。eureka-client是1.6.2版本。
eureka server api參考 https://blog.csdn.net/qq_30062125/article/details/83829357
這裡，eureka client分析主要分三個步驟：
- eureka client 自動配置類EurekaClientAutoConfiguration是怎麼找到的
- eureka客戶端邏輯是怎麼觸發的
- eureka client 如何發起註冊
1、 自動配置類是怎麼找到的
spring boot專案啟動類一般會加一個註解@SpringBootApplication，這個註解包含了@EnableAutoConfiguration。@EnableAutoConfiguration是springboot各種自動配置外掛的入口。
EnableAutoConfiguration 程式碼清單
@Target(ElementType.TYPE) @Retention(RetentionPolicy.RUNTIME) @Documented @Inherited @AutoConfigurationPackage @Import(EnableAutoConfigurationImportSelector.class) public @interface EnableAutoConfiguration { String ENABLED_OVERRIDE_PROPERTY = "spring.boot.enableautoconfiguration"; /** * Exclude specific auto-configuration classes such that they will never be applied. * @return the classes to exclude */ Class<?>[] exclude() default {}; /** * Exclude specific auto-configuration class names such that they will never be * applied. * @return the class names to exclude * @since 1.3.0 */ String[] excludeName() default {}; }
我們可以看到@Import(EnableAutoConfigurationImportSelector.class)，進入它的父類AutoConfigurationImportSelector，可以看到引入bean邏輯：
@Override public String[] selectImports(AnnotationMetadata annotationMetadata) { if (!isEnabled(annotationMetadata)) { return NO_IMPORTS; } try { AutoConfigurationMetadata autoConfigurationMetadata = AutoConfigurationMetadataLoader .loadMetadata(this.beanClassLoader); AnnotationAttributes attributes = getAttributes(annotationMetadata); // 獲取需要載入到spring容器中的bean列表 List<String> configurations = getCandidateConfigurations(annotationMetadata, attributes); configurations = removeDuplicates(configurations); configurations = sort(configurations, autoConfigurationMetadata); Set<String> exclusions = getExclusions(annotationMetadata, attributes); checkExcludedClasses(configurations, exclusions); configurations.removeAll(exclusions); configurations = filter(configurations, autoConfigurationMetadata); fireAutoConfigurationImportEvents(configurations, exclusions); return configurations.toArray(new String[configurations.size()]); } catch (IOException ex) { throw new IllegalStateException(ex); } }
getCandidateConfigurations 邏輯如下：
protected List<String> getCandidateConfigurations(AnnotationMetadata metadata,
AnnotationAttributes attributes) {
List<String> configurations = SpringFactoriesLoader.loadFactoryNames(
getSpringFactoriesLoaderFactoryClass(), getBeanClassLoader());
Assert.notEmpty(configurations,
"No auto configuration classes found in META-INF/spring.factories. If you "
+ "are using a custom packaging, make sure that file is correct.");
return configurations;
}
這裡會載入所有的META-INF/spring.factories檔案，並且會迴圈找出來org.springframework.boot.autoconfigure.EnableAutoConfiguration對應的自動註冊類，返回回去。
我們這裡關心的是Eureka client相關的，路徑在spring-cloud-netflix-eureka-client/1.3.6.RELEASE/spring-cloud-netflix-eureka-client-1.3.6.RELEASE.jar!/META-INF/spring.factories，其中內容為：
org.springframework.boot.autoconfigure.EnableAutoConfiguration=\
org.springframework.cloud.netflix.eureka.config.EurekaClientConfigServerAutoConfiguration,\
org.springframework.cloud.netflix.eureka.config.EurekaDiscoveryClientConfigServiceAutoConfiguration,\
org.springframework.cloud.netflix.eureka.EurekaClientAutoConfiguration,\
org.springframework.cloud.netflix.ribbon.eureka.RibbonEurekaAutoConfiguration
org.springframework.cloud.bootstrap.BootstrapConfiguration=\
org.springframework.cloud.netflix.eureka.config.EurekaDiscoveryClientConfigServiceBootstrapConfiguration
org.springframework.cloud.client.discovery.EnableDiscoveryClient=\
org.springframework.cloud.netflix.eureka.EurekaDiscoveryClientConfiguration
這裡載入了org.springframework.cloud.netflix.eureka.EurekaClientAutoConfiguration，這是一個比較重要的類，eureka client的很多邏輯都是從這入口的。
2、 eureka客戶端邏輯是怎麼觸發的
開啟EurekaClient需要配置@EnableDiscoveryClient或者@EnableEurekaClient，當前版本，這兩個沒啥區別。
上面我們分析瞭如何載入EurekaClientAutoConfiguration，不過如果細心看一下，我們會發現EurekaClientAutoConfiguration依舊需要一些條件才會載入。
@Configuration
@EnableConfigurationProperties
@ConditionalOnClass(EurekaClientConfig.class)
@ConditionalOnBean(EurekaDiscoveryClientConfiguration.Marker.class)
@ConditionalOnProperty(value = "eureka.client.enabled", matchIfMissing = true)
@AutoConfigureBefore({ NoopDiscoveryClientAutoConfiguration.class,
CommonsClientAutoConfiguration.class, ServiceRegistryAutoConfiguration.class })
@AutoConfigureAfter(name = "org.springframework.cloud.autoconfigure.RefreshAutoConfiguration")
public class EurekaClientAutoConfiguration {
......
}
其中關鍵的一點是 @ConditionalOnBean(EurekaDiscoveryClientConfiguration.Marker.class)
我們來看下這個bean是怎麼載入的。
這得從@EnableDiscoveryClient開始說起：
EnableDiscoveryClient程式碼：
@Target(ElementType.TYPE)
@Retention(RetentionPolicy.RUNTIME)
@Documented
@Inherited
@Import(EnableDiscoveryClientImportSelector.class)
public @interface EnableDiscoveryClient {
/**
* If true, the ServiceRegistry will automatically register the local server.
*/
boolean autoRegister() default true;
}
原理和上面類似，也是通過@Import注入bean。
我們看下EnableDiscoveryClientImportSelector的父類方法 org.springframework.cloud.commons.util.SpringFactoryImportSelector#selectImports：
@Override
public String[] selectImports(AnnotationMetadata metadata) {
if (!isEnabled()) {
return new String[0];
}
AnnotationAttributes attributes = AnnotationAttributes.fromMap(
metadata.getAnnotationAttributes(this.annotationClass.getName(), true));
Assert.notNull(attributes, "No " + getSimpleName() + " attributes found. Is "
+ metadata.getClassName() + " annotated with @" + getSimpleName() + "?");
// 獲取需要載入的bean列表
// Find all possible auto configuration classes, filtering duplicates
List<String> factories = new ArrayList<>(new LinkedHashSet<>(SpringFactoriesLoader
.loadFactoryNames(this.annotationClass, this.beanClassLoader)));
if (factories.isEmpty() && !hasDefaultFactory()) {
throw new IllegalStateException("Annotation @" + getSimpleName()
+ " found, but there are no implementations. Did you forget to include a starter?");
}
if (factories.size() > 1) {
// there should only ever be one DiscoveryClient, but there might be more than
// one factory
log.warn("More than one implementation " + "of @" + getSimpleName()
+ " (now relying on @Conditionals to pick one): " + factories);
}
return factories.toArray(new String[factories.size()]);
}
和上面自動注入邏輯類似，也是從META-INF/spring.factories中查詢當前註解對應的載入類。
這裡找的是
org.springframework.cloud.client.discovery.EnableDiscoveryClient=\
org.springframework.cloud.netflix.eureka.EurekaDiscoveryClientConfiguration
檢視EurekaDiscoveryClientConfiguration程式碼
@Configuration
@EnableConfigurationProperties
@ConditionalOnClass(EurekaClientConfig.class)
@ConditionalOnProperty(value = "eureka.client.enabled", matchIfMissing = true)
@CommonsLog
public class EurekaDiscoveryClientConfiguration {
class Marker {}
@Bean
public Marker eurekaDiscoverClientMarker() {
return new Marker();
}
......
}
最上面就是Marker bean的宣告，也就是前面@ConditionalOnBean(EurekaDiscoveryClientConfiguration.Marker.class)依賴的bean，所以只有加上@EnableDiscoveryClient，eureka客戶端邏輯才會生效。
3、 eureka client 如何發起註冊
首先看org.springframework.cloud.netflix.eureka.EurekaClientAutoConfiguration#eurekaInstanceConfigBean，eureka client的客戶端配置基本都是通過它來載入的。
@Bean
@ConditionalOnMissingBean(value = EurekaInstanceConfig.class, search = SearchStrategy.CURRENT)
public EurekaInstanceConfigBean eurekaInstanceConfigBean(InetUtils inetUtils) {
RelaxedPropertyResolver relaxedPropertyResolver = new RelaxedPropertyResolver(env, "eureka.instance.");
String hostname = relaxedPropertyResolver.getProperty("hostname");
boolean preferIpAddress = Boolean.parseBoolean(relaxedPropertyResolver.getProperty("preferIpAddress"));
EurekaInstanceConfigBean instance = new EurekaInstanceConfigBean(inetUtils);
instance.setNonSecurePort(this.nonSecurePort);
instance.setInstanceId(getDefaultInstanceId(this.env));
instance.setPreferIpAddress(preferIpAddress);
if (this.managementPort != this.nonSecurePort && this.managementPort != 0) {
if (StringUtils.hasText(hostname)) {
instance.setHostname(hostname);
}
String statusPageUrlPath = relaxedPropertyResolver.getProperty("statusPageUrlPath");
String healthCheckUrlPath = relaxedPropertyResolver.getProperty("healthCheckUrlPath");
if (StringUtils.hasText(statusPageUrlPath)) {
instance.setStatusPageUrlPath(statusPageUrlPath);
}
if (StringUtils.hasText(healthCheckUrlPath)) {
instance.setHealthCheckUrlPath(healthCheckUrlPath);
}
String scheme = instance.getSecurePortEnabled() ? "https" : "http";
instance.setStatusPageUrl(scheme + "://" + instance.getHostname() + ":"
+ this.managementPort + instance.getStatusPageUrlPath());
instance.setHealthCheckUrl(scheme + "://" + instance.getHostname() + ":"
+ this.managementPort + instance.getHealthCheckUrlPath());
}
return instance;
}
由於EurekaInstanceConfigBean類上面配置了@ConfigurationProperties，所以生成bean的過程中，在ConfigurationPropertiesBindingPostProcessor邏輯中，會注入配置檔案的配置引數。
@Data
@ConfigurationProperties("eureka.instance")
public class EurekaInstanceConfigBean implements CloudEurekaInstanceConfig, EnvironmentAware {
EurekaInstanceConfigBean這個bean為ApplicationInfoManager依賴，而ApplicationInfoManager被EurekaClient依賴。
@Bean
@ConditionalOnMissingBean(value = ApplicationInfoManager.class, search = SearchStrategy.CURRENT)
public ApplicationInfoManager eurekaApplicationInfoManager(
EurekaInstanceConfig config) {
InstanceInfo instanceInfo = new InstanceInfoFactory().create(config);
return new ApplicationInfoManager(config, instanceInfo);
}
eurekaClient bean是client的入口，注入邏輯
主要在
org.springframework.cloud.netflix.eureka.EurekaClientAutoConfiguration.EurekaClientConfiguration#eurekaClient
和
org.springframework.cloud.netflix.eureka.EurekaClientAutoConfiguration.RefreshableEurekaClientConfiguration#eurekaClient
兩個是針對不同場景的client注入。
EurekaClient bean 宣告
@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 建構函式
public CloudEurekaClient(ApplicationInfoManager applicationInfoManager,
EurekaClientConfig config,
DiscoveryClientOptionalArgs args,
ApplicationEventPublisher publisher) {
super(applicationInfoManager, config, args);
this.applicationInfoManager = applicationInfoManager;
this.publisher = publisher;
this.eurekaTransportField = ReflectionUtils.findField(DiscoveryClient.class, "eurekaTransport");
ReflectionUtils.makeAccessible(this.eurekaTransportField);
}
檢視CloudEurekaClient父類DiscoveryClient宣告。
com.netflix.discovery.DiscoveryClient#DiscoveryClient
@Inject
DiscoveryClient(ApplicationInfoManager applicationInfoManager, EurekaClientConfig config, AbstractDiscoveryClientOptionalArgs args,
Provider<BackupRegistry> backupRegistryProvider) {
......
initScheduledTasks();
.....
}
註冊主要通過initScheduledTasks實現
/**
* Initializes all scheduled tasks.
*/
private void initScheduledTasks() {
if (clientConfig.shouldFetchRegistry()) {
// registry cache refresh timer
int registryFetchIntervalSeconds = clientConfig.getRegistryFetchIntervalSeconds();
int expBackOffBound = clientConfig.getCacheRefreshExecutorExponentialBackOffBound();
scheduler.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 timer
scheduler.schedule(
new TimedSupervisorTask(
"heartbeat",
scheduler,
heartbeatExecutor,
renewalIntervalInSecs,
TimeUnit.SECONDS,
expBackOffBound,
new HeartbeatThread()
),
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");
}
}
這裡我們看下heartbeat定時任務，通過new HeartbeatThread()實現。
/**
* The heartbeat task that renews the lease in the given intervals.
*/
private class HeartbeatThread implements Runnable {
public void run() {
if (renew()) {
lastSuccessfulHeartbeatTimestamp = System.currentTimeMillis();
}
}
}
renew()邏輯，向註冊中心傳送心跳返回404，就會發起註冊。
/**
* Renew with the eureka service by making the appropriate REST call
*/
boolean renew() {
EurekaHttpResponse<InstanceInfo> httpResponse;
try {
httpResponse = eurekaTransport.registrationClient.sendHeartBeat(instanceInfo.getAppName(), instanceInfo.getId(), instanceInfo, null);
logger.debug("{} - Heartbeat status: {}", PREFIX + appPathIdentifier, httpResponse.getStatusCode());
if (httpResponse.getStatusCode() == 404) {
REREGISTER_COUNTER.increment();
logger.info("{} - Re-registering apps/{}", PREFIX + appPathIdentifier, instanceInfo.getAppName());
return register();
}
return httpResponse.getStatusCode() == 200;
} catch (Throwable e) {
logger.error("{} - was unable to send heartbeat!", PREFIX + appPathIdentifier, e);
return false;
}
}
註冊服務到註冊中心
/**
* Register with the eureka service by making the appropriate REST call.
*/
boolean register() throws Throwable {
logger.info(PREFIX + appPathIdentifier + ": registering service...");
EurekaHttpResponse<Void> httpResponse;
try {
// 實際註冊邏輯
httpResponse = eurekaTransport.registrationClient.register(instanceInfo);
} catch (Exception e) {
logger.warn("{} - registration failed {}", PREFIX + appPathIdentifier, e.getMessage(), e);
throw e;
}
if (logger.isInfoEnabled()) {
logger.info("{} - registration status: {}", PREFIX + appPathIdentifier, httpResponse.getStatusCode());
}
return httpResponse.getStatusCode() == 204;
}
最終呼叫com.netflix.discovery.shared.transport.jersey.AbstractJerseyEurekaHttpClient#register
@Override
public EurekaHttpResponse<Void> register(InstanceInfo info) {
// 這個地方發起post請求，url類似http://localhost:8761/eureka/apps/SERVICES2
// 傳遞的物件是 InstanceInfo
String urlPath = "apps/" + info.getAppName();
ClientResponse response = null;
try {
Builder resourceBuilder = jerseyClient.resource(serviceUrl).path(urlPath).getRequestBuilder();
addExtraHeaders(resourceBuilder);
response = resourceBuilder
.header("Accept-Encoding", "gzip")
.type(MediaType.APPLICATION_JSON_TYPE)
.accept(MediaType.APPLICATION_JSON)
.post(ClientResponse.class, info);
return anEurekaHttpResponse(response.getStatus()).headers(headersOf(response)).build();
} finally {
if (logger.isDebugEnabled()) {
logger.debug("Jersey HTTP POST {}/{} with instance {}; statusCode={}", serviceUrl, urlPath, info.getId(),
response == null ? "N/A" : response.getStatus());
}
if (response != null) {
response.close();
}
}
}
到這裡 服務已經註冊到註冊中心了。