spring cloud服務發現和註冊源碼分析

時間 2019-12-05

原文原文鏈接

概述

Spring Cloud中默認的服務發現採用的netflix的eureka，本篇文章就是閱讀Spring cloud中經過eureka作服務發現時的筆記。spring

順藤摸瓜

讀取Spring各類框架的時候，不少時候不知道從什麼地方開始，由於Spring中不少模塊的開啓就是經過一行註解，例如@EnableXXX。而Spring Cloud中的配置服務則是經過@EnableDiscoveryClient，(其實@EnableEurekaClient就是@EnableDiscoveryClient)緩存

/**
 * Annotation to enable a DiscoveryClient implementation.
 * @author Spencer Gibb
 */
@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(EnableDiscoveryClientImportSelector.class),而後看一下EnableDiscoveryClientImportSelector這個類服務器

public class EnableDiscoveryClientImportSelector
		extends SpringFactoryImportSelector<EnableDiscoveryClient> {

	@Override
	public String[] selectImports(AnnotationMetadata metadata) {
	...
		boolean autoRegister = attributes.getBoolean("autoRegister");

		if (autoRegister) {
			List<String> importsList = new ArrayList<>(Arrays.asList(imports));
			importsList.add("org.springframework.cloud.client.serviceregistry.AutoServiceRegistrationConfiguration");
			imports = importsList.toArray(new String[0]);
		}

		return imports;
	}


}

一看代碼第一反應就是引入了org.springframework.cloud.client.serviceregistry.AutoServiceRegistrationConfiguration這個類，再看一下這個類app

@Configuration
@EnableConfigurationProperties(AutoServiceRegistrationProperties.class)
public class AutoServiceRegistrationConfiguration {
}

什麼都沒有，怎麼回事。其實機關就是在SpringFactoryImportSelector<EnableDiscoveryClient>這個類。其實他會把類全命名做爲自動配置類的key，瞭解Spring Boot自動加載過程的就應該知道在某個jar包下META-INFO/spring.factories。框架

org.springframework.cloud.client.discovery.EnableDiscoveryClient=\
org.springframework.cloud.netflix.eureka.EurekaDiscoveryClientConfiguration

這樣，Spring Cloud啓動的時候初始化Eureka就是在EurekaDiscoveryClientConfiguration這個類中。仔細一看，也不對啊，這個類只註冊了一個Marker類，並且這個類是個空類。再全局搜索一下哪裏引用了這個Marker類，終於發現真正初始化Eureka的類：EurekaClientAutoConfigurationide

@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

追根溯源

既然找到了EurekaClientAutoConfiguration這個配置類，那麼確定有初始化和NetFlix相關類，仔細一看，就是函數

@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) {
	manager.getInfo(); // force initialization
	return new CloudEurekaClient(manager, config, this.optionalArgs,
			this.context);
}

終於發現和NetFlix的連接的地方，CloudEurekaClient繼承的DiscoveryClient其實就是NetFlix的服務發現類，這樣，就能夠好好分析是怎麼初始化的。oop

直擊要害

CloudEurekaClient在初始化的時候主要是調用父類DiscoveryClient的構造函數，會作不少事，其中註冊服務和發現服務是經過調度任務來完成，調度任務的初始化是在initScheduledTasks這個而方法中，其中服務發現的代碼this

ate 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);
    }

首先判斷是否須要進行服務發現，而後經過一個定時任務去刷新緩存信息。TimedSupervisorTask是支持timeout的調度任務，刷新緩存邏輯實際是在CacheRefreshThread()中，日後看就會看到經過Http請求和Erureka服務器交互。服務註冊的代碼緊接着在服務發現後面url

if (clientConfig.shouldRegisterWithEureka()) {
       ...
        // 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

       ...
        if (clientConfig.shouldOnDemandUpdateStatusChange()) {
            applicationInfoManager.registerStatusChangeListener(statusChangeListener);
        }

        instanceInfoReplicator.start(clientConfig.getInitialInstanceInfoReplicationIntervalSeconds());
    }

instanceInfoReplicator對當前服務信息（ instanceInfo）進行註冊。

public void run() {
    ...
    discoveryClient.refreshInstanceInfo();

    Long dirtyTimestamp = instanceInfo.isDirtyWithTime();
    if (dirtyTimestamp != null) {
        discoveryClient.register();
        instanceInfo.unsetIsDirty(dirtyTimestamp);
    }
    ...
}

另一個心跳檢測的定時任務則是對服務進行續約

boolean renew() {
        EurekaHttpResponse<InstanceInfo> httpResponse;
        try {
            httpResponse = eurekaTransport.registrationClient.sendHeartBeat(instanceInfo.getAppName(), instanceInfo.getId(), instanceInfo, null);
            if (httpResponse.getStatusCode() == 404) {
                ...
                return register();
            }
            return httpResponse.getStatusCode() == 200;
        } catch (Throwable e) {
            logger.error("{} - was unable to send heartbeat!", PREFIX + appPathIdentifier, e);
            return false;
        }
    }

能夠看出若是續約失敗則從新發起一次註冊。那麼服務註冊的具體實現就在register()中

boolean register() throws Throwable {
    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;
}

實現十分簡單，就是發起一次http調用，把當前instanceInfo傳遞過去。，那麼InstanceInfo都有什麼呢，其實就是一些appName ,url,port等等，經過這些信息咱們就能發起一次基於http的rpc調用。