根据前文我们对Eureka Server 有了一定的了解，本节我们主要学习Eureka Client 与 Eureka Server 如何通讯的及相关通信机制是什么，本文会弄清楚一下几个问题：
当我们使用服务发现的注解时，却发现了两种注解，经过尝试发现两者的使用效果是一样的，那么他们的区别是什么呢,通过查询官方文档是这样解释的：
Spring Cloud Commons provides the @EnableDiscoveryClient annotation.
This looks for implementations of the DiscoveryClient interface with
META-INF/spring.factories. Implementations of the Discovery Client add
a configuration class to spring.factories under the
org.springframework.cloud.client.discovery.EnableDiscoveryClient key.
Examples of DiscoveryClient implementations include Spring Cloud
Netflix Eureka, Spring Cloud Consul Discovery, and Spring Cloud
Zookeeper Discovery.
Spring Cloud Commons 提供 @EnableDiscoveryClient 注解。这将使用META-INF / spring.factories查找DiscoveryClient接口的实现。Discovery Client的实现将配置类添加到org.springframework.cloud.client.discovery.EnableDiscoveryClient项下的spring.factories。比如 DiscoveryClient 实现包括Spring Cloud Netflix Eureka，Spring Cloud Consul Discovery和Spring Cloud Zookeeper Discovery。
By default, implementations of DiscoveryClient auto-register the local
Spring Boot server with the remote discovery server. This behavior can
be disabled by setting autoRegister=false in @EnableDiscoveryClient.
默认情况下，DiscoveryClient 的实现会使用远程发现服务器自动注册本地Spring Boot服务器。可以通过在@EnableDiscoveryClient中设置autoRegister = false来禁用此行为。
简而言之 spring cloud 中服务发现有多种实现（eureka、consul、zookeeper等等），@EnableDiscoveryClient 基于 spring-cloud-commons, @EnableEurekaClient 基于 spring-cloud-netflix。
如果选用的注册中心是eureka，那么就推荐@EnableEurekaClient，如果是其他的注册中心，那么推荐使用@EnableDiscoveryClient。
我们看一下这两个注解的源码：
@Target(ElementType.TYPE)
@Retention(RetentionPolicy.RUNTIME)
@Documented
@Inherited
@Import(EnableDiscoveryClientImportSelector.class)
public @interface EnableDiscoveryClient {
/**
* 如果为true，本服务奖自动注册到指定的服务中心，默认是启用的，
可以通过设置为 false 来禁用自动注册
*/
boolean autoRegister() default true;
}
@Target(ElementType.TYPE)
@Retention(RetentionPolicy.RUNTIME)
@Documented
@Inherited
public @interface EnableEurekaClient {
}我们根据注解 @EnableEurekaClient 找到该注解所在工程 spring-cloud-netflix-eureka-client-2.0.1.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.netflix.eureka.EurekaDiscoveryClientConfiguration
org.springframework.cloud.bootstrap.BootstrapConfiguration=\
org.springframework.cloud.netflix.eureka.config.EurekaDiscoveryClientConfigServiceBootstrapConfiguration本文主要看 org.springframework.cloud.netflix.eureka.EurekaClientAutoConfiguration，其它的主要是客户端查找配置中心及客户端负载均衡（Ribbon）的引导服务，在后续文章的学习总结中再做介绍。
下面我们就从 EurekaClientAutoConfiguration 的源码着手，看看都有哪些操作,先看一下该配置类的注解有哪些
@Configuration
# 加载配置文件解析
@EnableConfigurationProperties
# 加载client配置项
@ConditionalOnClass(EurekaClientConfig.class)
# 加载 DiscoveryClientOptionalArgsConfiguration 到容器
@Import(DiscoveryClientOptionalArgsConfiguration.class)
# 客户端启用引导标记
@ConditionalOnBean(EurekaDiscoveryClientConfiguration.Marker.class)
# 启用eureka client，默认启用
@ConditionalOnProperty(value = "eureka.client.enabled", matchIfMissing = true)
# 当前配置类EurekaClientAutoConfiguration 加载完毕后的后续加载操作
@AutoConfigureBefore({ NoopDiscoveryClientAutoConfiguration.class,
CommonsClientAutoConfiguration.class, ServiceRegistryAutoConfiguration.class })
@AutoConfigureAfter(name = {"org.springframework.cloud.autoconfigure.RefreshAutoConfiguration",
"org.springframework.cloud.netflix.eureka.EurekaDiscoveryClientConfiguration",
"org.springframework.cloud.client.serviceregistry.AutoServiceRegistrationAutoConfiguration"})
public class EurekaClientAutoConfiguration {
}@ImplementedBy(DefaultEurekaClientConfig.class)
public interface EurekaClientConfig {}查看默认实现 DefaultEurekaClientConfig，eureka client 默认值加载
@Singleton
@ProvidedBy(DefaultEurekaClientConfigProvider.class)
public class DefaultEurekaClientConfig implements EurekaClientConfig {
/** @deprecated */
@Deprecated
public static final String DEFAULT_NAMESPACE = "eureka.";
public static final String DEFAULT_ZONE = "defaultZone";
private final String namespace;
private final DynamicPropertyFactory configInstance;
private final EurekaTransportConfig transportConfig;
public DefaultEurekaClientConfig() {
this("eureka");
}
public DefaultEurekaClientConfig(String namespace) {
this.namespace = namespace.endsWith(".") ? namespace : namespace + ".";
this.configInstance = Archaius1Utils.initConfig("eureka-client");
this.transportConfig = new DefaultEurekaTransportConfig(namespace, this.configInstance);
}
public int getRegistryFetchIntervalSeconds() {
return this.configInstance.getIntProperty(this.namespace + "client.refresh.interval", 30).get();
}
public int getInstanceInfoReplicationIntervalSeconds() {
return this.configInstance.getIntProperty(this.namespace + "appinfo.replicate.interval", 30).get();
}
public int getInitialInstanceInfoReplicationIntervalSeconds() {
return this.configInstance.getIntProperty(this.namespace + "appinfo.initial.replicate.time", 40).get();
}
public int getEurekaServiceUrlPollIntervalSeconds() {
return this.configInstance.getIntProperty(this.namespace + "serviceUrlPollIntervalMs", 300000).get() / 1000;
}
public String getProxyHost() {
return this.configInstance.getStringProperty(this.namespace + "eurekaServer.proxyHost", (String)null).get();
}
........................eureka client 默认值加载............................
}@Configuration
@EnableConfigurationProperties
@ConditionalOnClass(EurekaClientConfig.class)
@ConditionalOnProperty(value = "eureka.client.enabled", matchIfMissing = true)
public class EurekaDiscoveryClientConfiguration {
class Marker {}
@Bean
public Marker eurekaDiscoverClientMarker() {
return new Marker();
}
# 具体配置刷新事件的监听器
@Configuration
@ConditionalOnClass(RefreshScopeRefreshedEvent.class)
protected static class EurekaClientConfigurationRefresher {
@Autowired(required = false)
private EurekaClient eurekaClient;
@Autowired(required = false)
private EurekaAutoServiceRegistration autoRegistration;
@EventListener(RefreshScopeRefreshedEvent.class)
public void onApplicationEvent(RefreshScopeRefreshedEvent event) {
//This will force the creation of the EurkaClient bean if not already created
//to make sure the client will be reregistered after a refresh event
if(eurekaClient != null) {
eurekaClient.getApplications();
}
if (autoRegistration != null) {
// register in case meta data changed
this.autoRegistration.stop();
this.autoRegistration.start();
}
}
}
# 健康检查配置
@Configuration
@ConditionalOnProperty(value = "eureka.client.healthcheck.enabled", matchIfMissing = false)
protected static class EurekaHealthCheckHandlerConfiguration {
@Autowired(required = false)
private HealthAggregator healthAggregator = new OrderedHealthAggregator();
@Bean
@ConditionalOnMissingBean(HealthCheckHandler.class)
public EurekaHealthCheckHandler eurekaHealthCheckHandler() {
return new EurekaHealthCheckHandler(this.healthAggregator);
}
}
}该配置文件主要做了两件事情，一个是监听RefreshScopeRefreshedEvent事件，配置文件动态刷新时触发。另一个是配置健康检查处理程序。
根据源码可以看到这里创建了EurekaClientConfigBean、EurekaInstanceConfigBean两个基本配置，以及EurekaServiceRegistry
@Bean
@ConditionalOnMissingBean(value = ApplicationInfoManager.class, search = SearchStrategy.CURRENT)
public ApplicationInfoManager eurekaApplicationInfoManager(
EurekaInstanceConfig config) {
InstanceInfo instanceInfo = new InstanceInfoFactory().create(config);
return new ApplicationInfoManager(config, instanceInfo);
}使用EurekaInstanceConfig，通过new InstanceInfoFactory().create(config)创建
使用InstanceInfo以及EurekaInstanceConfig创建：new ApplicationInfoManager(config, instanceInfo)
@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);
}使用ApplicationInfoManager、EurekaClientConfig创建：new CloudEurekaClient(manager, config, this.optionalArgs,this.context)
@Bean
public DiscoveryClient discoveryClient(EurekaInstanceConfig config, EurekaClient client) {
return new EurekaDiscoveryClient(config, client);
}通过EurekaInstanceConfig、EurekaClient创建：new EurekaDiscoveryClient(config, client)
@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();
}通过EurekaClient、CloudEurekaInstanceConfig、ApplicationInfoManager来创建
@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);
}通过EurekaServiceRegistry、EurekaRegistration来创建
该类实现了接口 SmartLifecycle 的方法。SmartLifecycle 是一个接口。当Spring容器加载所有bean并完成初始化之后，会接着回调实现该接口的类中对应的方法。
public class EurekaAutoServiceRegistration implements AutoServiceRegistration, SmartLifecycle, Ordered {
private static final Log log = LogFactory.getLog(EurekaAutoServiceRegistration.class);
private AtomicBoolean running = new AtomicBoolean(false);
private int order = 0;
private AtomicInteger port = new AtomicInteger(0);
private ApplicationContext context;
private EurekaServiceRegistry serviceRegistry;
private EurekaRegistration registration;
public EurekaAutoServiceRegistration(ApplicationContext context, EurekaServiceRegistry serviceRegistry, EurekaRegistration registration) {
this.context = context;
this.serviceRegistry = serviceRegistry;
this.registration = registration;
}
@Override
public void start() {
..................
}
@Override
public void stop() {
this.serviceRegistry.deregister(this.registration);
this.running.set(false);
}
@Override
public boolean isRunning() {
return this.running.get();
}
@Override
public int getPhase() {
return 0;
}
@Override
public boolean isAutoStartup() {
return true;
}
@Override
public void stop(Runnable callback) {
stop();
callback.run();
}
@Override
public int getOrder() {
return this.order;
}
@EventListener(WebServerInitializedEvent.class)
public void onApplicationEvent(WebServerInitializedEvent event) {
..................
}
@EventListener(ContextClosedEvent.class)
public void onApplicationEvent(ContextClosedEvent event) {
if( event.getApplicationContext() == context ) {
stop();
}
}
}@Override
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) {
//注册自身实例服务到注册中心
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);
}调用this.serviceRegistry.deregister(this.registration)方法，告知eureka server自身服务要下线
根据源码可知 EurekaAutoServiceRegistration 中的 start()、stop() 方法分别调用了 EurekaServiceRegistry 的 register()、deregister() 方法实现了服务注册和服务主动下线。下面我们就看一下这两个方法的实现。
调用ApplicationInfoManager的setInstanceStatus方法来更改状态
@Override
public void register(EurekaRegistration reg) {
maybeInitializeClient(reg);
if (log.isInfoEnabled()) {
log.info("Registering application " + reg.getInstanceConfig().getAppname()
+ " with eureka with status "
+ reg.getInstanceConfig().getInitialStatus());
}
reg.getApplicationInfoManager()
.setInstanceStatus(reg.getInstanceConfig().getInitialStatus());
reg.getHealthCheckHandler().ifAvailable(healthCheckHandler ->
reg.getEurekaClient().registerHealthCheck(healthCheckHandler));
}
private void maybeInitializeClient(EurekaRegistration reg) {
// force initialization of possibly scoped proxies
reg.getApplicationInfoManager().getInfo();
reg.getEurekaClient().getApplications();
}调用ApplicationInfoManager的setInstanceStatus方法来将状态设置为InstanceInfo.InstanceStatus.DOWN
@Override
public void deregister(EurekaRegistration reg) {
if (reg.getApplicationInfoManager().getInfo() != null) {
if (log.isInfoEnabled()) {
log.info("Unregistering application " + reg.getInstanceConfig().getAppname()
+ " with eureka with status DOWN");
}
reg.getApplicationInfoManager().setInstanceStatus(InstanceInfo.InstanceStatus.DOWN);
//shutdown of eureka client should happen with EurekaRegistration.close()
//auto registration will create a bean which will be properly disposed
//manual registrations will need to call close()
}
}设置服务实例状态并发布StatusChangeEvent事件，通知所有该事件监听者
/**
* 设置此实例的状态。 应用程序可以使用它来标识它是否已准备好接收流量。 在此处设置状态还会通知所有已注册的监听器状态更改事件。
*/
public synchronized void setInstanceStatus(InstanceStatus status) {
//当前服务实例状态
InstanceStatus next = instanceStatusMapper.map(status);
if (next == null) {
return;
}
//当前服务实例之前的状态
InstanceStatus prev = instanceInfo.setStatus(next);
if (prev != null) {
for (StatusChangeListener listener : listeners.values()) {
try {
//状态变化事件通知所有监听者
listener.notify(new StatusChangeEvent(prev, next));
} catch (Exception e) {
logger.warn("failed to notify listener: {}", listener.getId(), e);
}
}
}
}这里注册了StatusChangeListener，之后触发instanceInfoReplicator.onDemandUpdate()
/**
* Initializes all scheduled tasks.
*/
private void initScheduledTasks() {
if (clientConfig.shouldFetchRegistry()) {
// 缓存刷新定时器
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);
// 心跳定时器
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");
}
}这里的onDemandUpdate()方法主要是执行InstanceInfoReplicator.this.run() 而这个run方法主要是判断是否dirty，如果是则调用discoveryClient.register()
public boolean onDemandUpdate() {
if (rateLimiter.acquire(burstSize, allowedRatePerMinute)) {
if (!scheduler.isShutdown()) {
scheduler.submit(new Runnable() {
@Override
public void run() {
logger.debug("Executing on-demand update of local InstanceInfo");
Future latestPeriodic = scheduledPeriodicRef.get();
if (latestPeriodic != null && !latestPeriodic.isDone()) {
logger.debug("Canceling the latest scheduled update, it will be rescheduled at the end of on demand update");
latestPeriodic.cancel(false);
}
InstanceInfoReplicator.this.run();
}
});
return true;
} else {
logger.warn("Ignoring onDemand update due to stopped scheduler");
return false;
}
} else {
logger.warn("Ignoring onDemand update due to rate limiter");
return false;
}
}
public void run() {
try {
discoveryClient.refreshInstanceInfo();
Long dirtyTimestamp = instanceInfo.isDirtyWithTime();
if (dirtyTimestamp != null) {
discoveryClient.register();
instanceInfo.unsetIsDirty(dirtyTimestamp);
}
} catch (Throwable t) {
logger.warn("There was a problem with the instance info replicator", t);
} finally {
Future next = scheduler.schedule(this, replicationIntervalSeconds, TimeUnit.SECONDS);
scheduledPeriodicRef.set(next);
}
}最后终于找到了服务注册的具体真正实现，register() 才是真正去与远程的Eureka Server交互，注册服务的操作
/**
* 通过REST调用来注册eureka服务。
*/
boolean register() throws Throwable {
logger.info(PREFIX + "{}: registering service...", appPathIdentifier);
EurekaHttpResponse<Void> httpResponse;
try {
httpResponse = eurekaTransport.registrationClient.register(instanceInfo);
} catch (Exception e) {
logger.warn(PREFIX + "{} - registration failed {}", appPathIdentifier, e.getMessage(), e);
throw e;
}
if (logger.isInfoEnabled()) {
logger.info(PREFIX + "{} - registration status: {}", appPathIdentifier, httpResponse.getStatusCode());
}
return httpResponse.getStatusCode() == 204;
}最后总结一下：
EurekaClientAutoConfiguration 构造了 EurekaClientConfigBean、EurekaInstanceConfigBean 以及 EurekaServiceRegistry，接着在这几个对象的基础上进一步构建了 ApplicationInfoManager、CloudEurekaClient等。
其中ApplicationInfoManager 负责变更实例状态并发布 StatusChangeEvent事件，而CloudEurekaClient继承了com.netflix.discovery.DiscoveryClient 包含了statusChangeListener 用于响应S tatusChangeEvent，最后触发的是DiscoveryClient.register方法，与远程的Eureka Server通信，同步实例状态。
版权说明 : 本文为转载文章, 版权归原作者所有 版权申明
原文链接 : https://blog.csdn.net/u010647035/article/details/83242887
内容来源于网络，如有侵权，请联系作者删除！