服务注册与发现组件 Eureka 客户端实现原理解析

logging:
  level:
    org.springframework: DEBUG

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.netflix.eureka.EurekaDiscoveryClientConfiguration

org.springframework.cloud.bootstrap.BootstrapConfiguration=\
org.springframework.cloud.netflix.eureka.config.EurekaDiscoveryClientConfigServiceBootstrapConfiguration

public class EurekaInstanceConfigBean implements CloudEurekaInstanceConfig, EnvironmentAware {

// 服务实例的应用名
private String appname; 
// 服务实例的Id，通常和appname共同唯一标记一个服务实例
private String instanceId; 
// 自定义添加的元数据，由用户使用以适配扩展业务需求
private Map<String, String> metadataMap;
// 如果服务实例部署在AWS上，该类将持有服务实例部署所在的数据中心的准确信息
private DataCenterInfo dataCenterInfo;
// 服务实例的Ip地址
private String ipAddress;
// 服务实例主页地址
private String homePageUrl;
// 服务实例健康检查地址
private String healthCheckUrlPath;
// 服务实例的状态地址
private String statusPageUrlPath

.....
}

public interface DiscoveryClient {

   // 获取实现类的描述
	String description();

	// 通过服务Id获取服务实例的信息
	List<ServiceInstance> getInstances(String serviceId);

	// 获取所有的服务实例的Id
	List<String> getServices();

}

@Override
public List<ServiceInstance> getInstances(String serviceId) {
	List<InstanceInfo> infos = this.eurekaClient.getInstancesByVipAddress(serviceId,false);
	List<ServiceInstance> instances = new ArrayList<>();
	for (InstanceInfo info : infos) {
		instances.add(new EurekaServiceInstance(info));
	}
	return instances;
}

@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.ipAddress");
	boolean isSecurePortEnabled = Boolean.parseBoolean(getProperty("eureka.instance.secure-port-enabled"));

	String serverContextPath = env.getProperty("server.context-path", "/");
	int serverPort = Integer.valueOf(env.getProperty("server.port", env.getProperty("port", "8080")));

	Integer managementPort = env.getProperty("management.server.port", Integer.class);// nullable. should be wrapped into optional
	String managementContextPath = env.getProperty("management.server.context-path");// nullable. should be wrapped into optional
	Integer jmxPort = env.getProperty("com.sun.management.jmxremote.port", Integer.class);//nullable
	EurekaInstanceConfigBean instance = new EurekaInstanceConfigBean(inetUtils);
   // 设置非空属性
	instance.setNonSecurePort(serverPort);
	instance.setInstanceId(getDefaultInstanceId(env));
	instance.setPreferIpAddress(preferIpAddress);
	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);

   .....
	return instance;
}

public InstanceInfo create(EurekaInstanceConfig config) {
		LeaseInfo.Builder leaseInfoBuilder = LeaseInfo.Builder.newBuilder()
				.setRenewalIntervalInSecs(config.getLeaseRenewalIntervalInSeconds())
				.setDurationInSecs(config.getLeaseExpirationDurationInSeconds());
		// 创建服务实例的信息用来注册到eureka server上
		InstanceInfo.Builder builder = InstanceInfo.Builder.newBuilder();

		String namespace = config.getNamespace();
		if (!namespace.endsWith(".")) {
			namespace = namespace + ".";
		}
		builder.setNamespace(namespace).setAppName(config.getAppname())
				.setInstanceId(config.getInstanceId())
				.setAppGroupName(config.getAppGroupName())
				.setDataCenterInfo(config.getDataCenterInfo())
				.setIPAddr(config.getIpAddress()).setHostName(config.getHostName(false))
				.setPort(config.getNonSecurePort())
				.enablePort(InstanceInfo.PortType.UNSECURE,
						config.isNonSecurePortEnabled())
				.setSecurePort(config.getSecurePort())
				.enablePort(InstanceInfo.PortType.SECURE, config.getSecurePortEnabled())
				.setVIPAddress(config.getVirtualHostName())
				.setSecureVIPAddress(config.getSecureVirtualHostName())
				.setHomePageUrl(config.getHomePageUrlPath(), config.getHomePageUrl())
				.setStatusPageUrl(config.getStatusPageUrlPath(),
						config.getStatusPageUrl())
				.setHealthCheckUrls(config.getHealthCheckUrlPath(),
						config.getHealthCheckUrl(), config.getSecureHealthCheckUrl())
				.setASGName(config.getASGName());

       ....
		InstanceInfo instanceInfo = builder.build();
		instanceInfo.setLeaseInfo(leaseInfoBuilder.build());
		return instanceInfo;
	}

@Bean
@ConditionalOnMissingBean(value = ApplicationInfoManager.class, search = SearchStrategy.CURRENT)
public ApplicationInfoManager eurekaApplicationInfoManager(EurekaInstanceConfig config) {
	InstanceInfo instanceInfo = new InstanceInfoFactory().create(config);
	return new ApplicationInfoManager(config, instanceInfo);
		}

@Bean
@ConditionalOnMissingBean(value = EurekaClientConfig.class, search = SearchStrategy.CURRENT)
public EurekaClientConfigBean eurekaClientConfigBean() {
	EurekaClientConfigBean client = new EurekaClientConfigBean();
	if ("bootstrap".equals(propertyResolver.getProperty("spring.config.name"))) {
		// We don't register during bootstrap by default, but there will be another
		// chance later.
		client.setRegisterWithEureka(false);
	}
	return client;
}

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

public interface ServiceInstance {

	//获取服务实例的serviceId
	String getServiceId();

	//获取服务实例的hostname
	String getHost();

   //获取服务实例的端口号
	int getPort();

	boolean isSecure();

	//获取服务实例的uri地址
	URI getUri();

	//获取服务实例的元数据key-value对
	Map<String, String> getMetadata();
}

@Bean
@ConditionalOnBean(AutoServiceRegistrationProperties.class)
@ConditionalOnProperty(value = "spring.cloud.service-registry.auto-registration.enabled", matchIfMissing = true)
public EurekaRegistration eurekaRegistration(EurekaClient eurekaClient, CloudEurekaInstanceConfig instanceConfig, ApplicationInfoManager applicationInfoManager, ObjectProvider<HealthCheckHandler> healthCheckHandler) {
		return EurekaRegistration.builder(instanceConfig)
				.with(applicationInfoManager)
				.with(eurekaClient)
				.with(healthCheckHandler)
				.build();
	}

public interface ServiceRegistry<R extends Registration> {

	//注册服务实例，registration当中通常有关于服务实例的信息，例如hostname和port
	void register(R registration);

   //注销服务实例
	void deregister(R registration);

	//关闭ServiceRegistry，通常在服务关闭的时候被调用
	void close();

	//设置服务实例的状态
	void setStatus(R registration, String status);

	//获取服务实例的状态
	<T> T getStatus(R registration);
}

@Override
public void register(EurekaRegistration reg) {
   // 初始化EurekaRegistration中的EurekaClient，如果为null
	maybeInitializeClient(reg);

    // 修改服务的状态为UP
	reg.getApplicationInfoManager()
			.setInstanceStatus(reg.getInstanceConfig().getInitialStatus());
    // 设置健康检查
	reg.getHealthCheckHandler().ifAvailable(healthCheckHandler ->
			reg.getEurekaClient().registerHealthCheck(healthCheckHandler));
}
	
private void maybeInitializeClient(EurekaRegistration reg) {
	reg.getApplicationInfoManager().getInfo();
	reg.getEurekaClient().getApplications();
}
	
@Override
public void deregister(EurekaRegistration reg) {
	if (reg.getApplicationInfoManager().getInfo() != null) {
		// 设置服务的状态为DOWN	
		reg.getApplicationInfoManager().setInstanceStatus(InstanceInfo.InstanceStatus.DOWN);
	}
}

...
@EventListener(ServletWebServerInitializedEvent.class)
public void onApplicationEvent(ServletWebServerInitializedEvent event) {
	int localPort = event.getWebServer().getPort();
	if (this.port.get() == 0) {
		log.info("Updating port to " + localPort);
		this.port.compareAndSet(0, localPort);
		start();
	}
}

@EventListener(ContextClosedEvent.class)
public void onApplicationEvent(ContextClosedEvent event) {
	if( event.getApplicationContext() == context ) {
		stop();
	}
}

@Override
public void start() {
	// only set the port if the nonSecurePort or securePort is 0 and this.port != 0
	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) {

       // 注册服务
		this.serviceRegistry.register(this.registration);

		this.context.publishEvent(
				new InstanceRegisteredEvent<>(this, this.registration.getInstanceConfig()));
		this.running.set(true);
	}
}
@Override
public void stop() {
   // 服务下线
	this.serviceRegistry.deregister(this.registration);
	this.running.set(false);
}

@EventListener(RefreshScopeRefreshedEvent.class)
public void onApplicationEvent(RefreshScopeRefreshedEvent event) {
	// 保证了一个刷新事件发生后client的重新注册
	if(eurekaClient != null) {
		eurekaClient.getApplications();
	}
	if (autoRegistration != null) {
		// 重新注册防止本地信息与注册表中的信息不一致
		this.autoRegistration.stop();
		this.autoRegistration.start();
	}
}

public interface LookupService<T> {

	//根据服务实例注册的appName来获取，获取一个封装有相同appName的服务实例信息的容器
   Application getApplication(String appName);
	//返回当前注册的所有的服务实例信息
   Applications getApplications();
   	//根据服务实例的id获取
   	List<InstanceInfo> getInstancesById(String id);
   //获取下一个可能的Eureka Server来处理当前对注册表信息的处理，一般是通过循环的方式来获取下一个Server
   InstanceInfo getNextServerFromEureka(String virtualHostname, boolean secure);
}

public class Application {
    
    private static Random shuffleRandom = new Random();

    //服务名
    private String name;

    @XStreamOmitField
    private volatile boolean isDirty = false;

    @XStreamImplicit
    private final Set<InstanceInfo> instances;

    private final AtomicReference<List<InstanceInfo>> shuffledInstances;

    private final Map<String, InstanceInfo> instancesMap;
    
    .....
}

public void addInstance(InstanceInfo i) {
	instancesMap.put(i.getId(), i);
	synchronized (instances) {
	instances.remove(i);
	instances.add(i);
	isDirty = true;
	}
}

// 为Eureka Client注册健康检查处理器
// 一旦注册，客户端将通过调用新注册的健康检查处理器来对注册中instanceInfo
// 进行一个按需更新，随后按照eurekaclientconfig.getinstanceinforeplicationintervalseconds()
// 中配置的指定时间调用HealthCheckHandler
public void registerHealthCheck(HealthCheckHandler healthCheckHandler);

// 为eureka client注册一个EurekaEventListener(事件监听器)
// 一旦注册，当eureka client的内部状态发生改变的时候，将会调用EurekaEventListener.onEvent()
// 触发一定的事件。可以通过这种方式监听client的更新而非通过轮询的方式询问client
public void registerEventListener(EurekaEventListener eventListener);

public interface HealthCheckHandler {
    InstanceInfo.InstanceStatus getStatus(InstanceInfo.InstanceStatus currentStatus);
}

public enum InstanceStatus {
	UP, // 可以接受服务请求
	DOWN, // 无法发送流量-健康检查失败
	STARTING, // 正在启动，无法发送流量
	OUT_OF_SERVICE, // 服务关闭，不接受流量
	UNKNOWN; // 未知状态
    }

DiscoveryClient(ApplicationInfoManager applicationInfoManager, EurekaClientConfig config, 
AbstractDiscoveryClientOptionalArgs args, Provider<BackupRegistry> backupRegistryProvider)

@ImplementedBy(NotImplementedRegistryImpl.class)
public interface BackupRegistry {

    Applications fetchRegistry();

    Applications fetchRegistry(String[] includeRemoteRegions);
}

public abstract class AbstractDiscoveryClientOptionalArgs<T> {
	// 生成健康检查回调的工厂类，HealthCheckCallback已废弃
   	Provider<HealthCheckCallback> healthCheckCallbackProvider;
   // 生成健康处理器的工厂类
   Provider<HealthCheckHandler> healthCheckHandlerProvider;
   // 向Eureka Server注册之前的预处理器
   PreRegistrationHandler preRegistrationHandler;
   // Jersey过滤器集合，Jersey1和Jersey2均可使用
   Collection<T> additionalFilters;
   // Jersey客户端，主要用于client与server之间的HTTP交互
   EurekaJerseyClient eurekaJerseyClient;
   // 生成Jersey客户端的工厂
   TransportClientFactory transportClientFactory;
   // 生成Jersey客户端的工厂的工厂
   TransportClientFactories transportClientFactories;
   // Eureka事件的监听器
   private Set<EurekaEventListener> eventListeners;
....
}

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

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

private static final class EurekaTransport {
   // Server endPoint解析器
	private ClosableResolver bootstrapResolver;
	// Jersy客户端生成工厂
	private TransportClientFactory transportClientFactory;
	// 注册客户端
	private EurekaHttpClient registrationClient;
	// 注册客户端生成工厂
	private EurekaHttpClientFactory registrationClientFactory;
	// 发现服务客户端
	private EurekaHttpClient queryClient;
	// 发现服务客户端生成工厂
 	private EurekaHttpClientFactory queryClientFactory;
 	
 	....

}