本文主要内容：

1. 客户端如何向Eureka注册信息
2. 客户端如何通过心跳续约服务
3. 客户端如何从Eureka获取服务列表
4. 学习借鉴Eurek客户端如何使用定时线程

spring-cloud-netflix-eureka-client不同版本之间，代码会不一样，本文分析的是spring-cloud-netflix-eureka-client-2.2.2.RELEASE.jar版

一、Eureka客户端注册

1、SmartLifeCycle知识

在学习Eureka客户端如何注册之前，先了解一个关于SmartLifeCycle的知识点，即：在Springboot在执行FinshRefresh()方法时，会逐一调用容器中所有实现了Lifecycle接口的Bean的star()方法，而SmartLifeCycle继承Lifecycle接口，也就说会调用SmartLifeCycle实现类的star()方法

源码方法调用链路：

• SpringApplication.run()方法
  • refreshContext()方法
  • refresh()方法
• AbstractApplicationContext.refresh()方法
  • finishRefresh()方法
  • getLifecycleProcessor().onRefresh();方法
• DefaultLifecycleProcessor.onRefresh()方法
  • startBeans()：方法

DefaultLifecycleProcessor#startBeans()的源码：

private void startBeans(boolean autoStartupOnly) {// 获取Lifecycle实现类beanMap<String, Lifecycle> lifecycleBeans = getLifecycleBeans();Map<Integer, LifecycleGroup> phases = new HashMap<Integer, LifecycleGroup>();for (Map.Entry<String, ? extends Lifecycle> entry : lifecycleBeans.entrySet()) {Lifecycle bean = entry.getValue();// SmartLifecycle实现类beanif (!autoStartupOnly || (bean instanceof SmartLifecycle && ((SmartLifecycle) bean).isAutoStartup())) {int phase = getPhase(bean);LifecycleGroup group = phases.get(phase);if (group == null) {group = new LifecycleGroup(phase, this.timeoutPerShutdownPhase, lifecycleBeans, autoStartupOnly);phases.put(phase, group);}group.add(entry.getKey(), bean);}}// 调用SmartLifecycle实现类的start()方法if (!phases.isEmpty()) {List<Integer> keys = new ArrayList<Integer>(phases.keySet());Collections.sort(keys);for (Integer key : keys) {phases.get(key).start();}}}

可以自定义一个TestSmartLifeCycle类验证此过程：

@Component
public class TestSmartLifeCycle implements SmartLifecycle {@Overridepublic void start() {System.out.println("TestSmartLifeCycle start");}@Overridepublic void stop() {System.out.println("TestSmartLifeCycle stop");}@Overridepublic boolean isRunning() {return false;}
}

Eureka客户端在进行服务注册时，就使用到了SmartLifeCycle相关功能，下面来看下具体注册流程

2、Eureka客户端注册流程及示意图

以下是Eureka客户端注册实例信息的流程流、和简易流程图，可以先看分析在回来看这部分介绍

Eureka Client会在服务启动时，调用EurekaAutoServiceRegistration#start()方法，此方法会进行以下操作：

1. EurekaClientAutoConfiguration注册Bean：EurekaAutoServiceRegistration、EurekaRegistration、ApplicationInfoMangager、EurekaClient(实现类：CloudEurekaClient)
2. 在Springboot.onFresh()方法中调用EurekaAutoServiceRegistration#star()方法
3. EurekaAutoServiceRegistration调用EurekaServiceRegistry#register()方法
4. EurekaServiceRegistry调用以下对象方法：
   • ApplicationInfoManager.setInstanceStatus()：设置ApplicationInfoManager的实例状态
   • CloudEurekaClient.registerHealthCheck()：注册一个心跳检测
5. ApplicationInfoManager最终调用**监听器listener.notify(new StatusChangeEvent(prev, status));**方法

以下是Eureka Client注册流程示意图：

3、Eureka客户端原理分析

3.1、EurekaClientAutoConfiguration配置类源码分析

当我们启用了Eureka注册中心功能后，Springboot会自动加载Eureka定义的自动装载配置类：EurekaClientAutoConfiguration.class，同时初始化以下几个重要的Bean：

• EurekaAutoServiceRegistration
• EurekaRegistration
• ApplicationInfoMangager
• EurekaClient(实现类：CloudEurekaClient)

@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);
}@Configuration(proxyBeanMethods = false)
@ConditionalOnMissingRefreshScope
protected static class EurekaClientConfiguration {@Autowiredprivate ApplicationContext context;@Autowiredprivate AbstractDiscoveryClientOptionalArgs<?> optionalArgs;@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);}@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@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();}}

3.2、EurekaAutoServiceRegistration服务注册源码分析

上面我们知道，Springboot在启动时会调用SmartLifeCycle实现类的start(）方法，而EurekaAutoServiceRegistration实现了SmartLifeCycle接口，所以在Springboot初始化容器后，会调用EurekaAutoServiceRegistration.star()方法：

Eureka不同版本对SmartLifeCycle接口的实现类不一样：

• 2.2.2版是：EurekaAutoServiceRegistration
• 1.2.5版是：EurekaDiscoveryClientConfiguration

public class EurekaAutoServiceRegistration implements AutoServiceRegistration, SmartLifecycle, Ordered, SmartApplicationListener {......@Overridepublic void start() {// only set the port if the nonSecurePort or securePort is 0 and this.port != 0if (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());}}// only initialize if nonSecurePort is greater than 0 and it isn't already running// because of containerPortInitializer belowif (!this.running.get() && this.registration.getNonSecurePort() > 0) {// 调用服务注册器注册服务:EurekaServiceRegistry.register()this.serviceRegistry.register(this.registration);this.context.publishEvent(new InstanceRegisteredEvent<>(this,this.registration.getInstanceConfig()));this.running.set(true);}}......
}

3.3、EurekaServiceRegistry服务注册器源码分析

进入EurekaServiceRegistry # register ()方法，又会分别调用以下两个方法：

• ApplicationInfoManager.setInstanceStatus()：设置ApplicationInfoManager的实例状态，
• CloudEurekaClient.registerHealthCheck()：注册一个心跳检测

@Override
public void register(EurekaRegistration reg) {maybeInitializeClient(reg);if (log.isInfoEnabled()) {log.info("Registering application "	+ reg.getApplicationInfoManager().getInfo().getAppName()+ " with eureka with status "+ reg.getInstanceConfig().getInitialStatus());}// 设置当前实例的状态，实例的状态发生变化，只要状态不是DOWN，会被监听器监听并且执行服务注册。reg.getApplicationInfoManager().setInstanceStatus(reg.getInstanceConfig().getInitialStatus());// 设置健康检查处理reg.getHealthCheckHandler().ifAvailable(healthCheckHandler -> reg.getEurekaClient().registerHealthCheck(healthCheckHandler));
}

3.4、ApplicationInfoManager.setInstanceStatus()：设置实例状态

此方法：设置当前你实例服务状态信息，并发布了一个StatusChangeEvent事件给监听器

 public synchronized void setInstanceStatus(InstanceStatus status) {InstanceStatus prev = instanceInfo.setStatus(status);if (prev != null) {for (StatusChangeListener listener : listeners.values()) {try {listener.notify(new StatusChangeEvent(prev, status));} catch (Exception e) {logger.warn("failed to notify listener: {}", listener.getId(), e);}}}}

看到这里我们发现，ApplicationInfoManager发布了一个StatusChangeEvent状态变更事件，并交由监听器处理，但是当我们查看Map<String, StatusChangeListener> listeners时发现找不到listener实现类？

之前说到，EurekaClientAutoConfiguration初始化了四个Bean，其中有一个Bean：EurekaClient(实现类：CloudEurekaClient)还没有提到过，而问题关键就在这个类里面

3.5、CloudEurekaClient 客户端源码分析

看CloudEurekaClient 源码，关键要看它的父类DiscoveryClient，其核心就在DiscoveryClient的构造方法上，在DiscoveryClient构造方法的中会调用内部initScheduledTasks()方法，完成以下操作：

源码很长，简要说几个核心功能，后面也会有部分代码，但不是全部，有兴趣可以去看一下DiscoveryClient构造方法代码

1. 如果需要获取服务列表：
   • 启动一个定时任务线程，定时执行refreshRegistry()方法拉取服务列表
2. 如果需要注册服务信息到服务端：
   • 启动一个定时心跳任务线程，定时执行renew()方法续约服务
   • 初始化一个StatusChangeListener监听器，监听StatusChange状态变更事件。并根据事件类型上报服务实例信息。

找到了Listener，我们接着3.4步骤继续调用：ApplicationInfoManager.setInstanceStatus()调用StatusChangeListener.notify()方法。此方法通过以下步骤最终进入AbstractJerseyEurekaHttpClient.register(InstanceInfo info)方法，向Eureka服务端的/apps/info.getAppName()注册自身信息：

1. statusChangeListener.notify(StatusChangeEvent statusChangeEvent)
2. instanceInfoReplicator.onDemandUpdate();
3. InstanceInfoReplicator.this.run();
4. discoveryClient.refreshInstanceInfo();
5. discoveryClient.register();
6. eurekaTransport.registrationClient.register(instanceInfo);
7. AbstractJerseyEurekaHttpClient.register(InstanceInfo info)

AbstractJerseyEurekaHttpClient.register(InstanceInfo info)源码

@Override
public EurekaHttpResponse<Void> register(InstanceInfo info) {String urlPath = "apps/" + info.getAppName();ClientResponse response = null;try {// 构建请求clientBuilder resourceBuilder = jerseyClient.resource(serviceUrl).path(urlPath).getRequestBuilder();addExtraHeaders(resourceBuilder);// 发送rest请求到Eureka服务端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();}}
}

总结

Eureka Cdlient向服务端注册实例信息有两个触发点：1、定时线程，2、容器初始化时候

1. Spring Boot启动时，自动装配机制将CloudEurekaClient注入到了容器中，在其构造方法中，会启动一个定时任务线程，每40s判断一次实例信息是否发生变化，如果有变化则发起服务注册流程
2. SpringBoot.finishRefresh方法，会调用SmartLifeCycle实现类的Bean的start(）方法，然后调用StatusChangeListener.notify进行服务状态变更，从而发起服务注册。

二、Eureka客户端续约

上面说到DiscoveryClient在初始化的时候，会调用initScheduledTasks()方法启动一个心跳线程，定时执行renew()方法上报实例信息进行续约

// Heartbeat timer 
// 心跳任务：HeartbeatThread，时间间隔：renewalIntervalInSecs，默认30秒
heartbeatTask = new TimedSupervisorTask("heartbeat",scheduler,heartbeatExecutor,renewalIntervalInSecs,TimeUnit.SECONDS,expBackOffBound,new HeartbeatThread()
);
// 线程执行心跳任务
scheduler.schedule(heartbeatTask,renewalIntervalInSecs, TimeUnit.SECONDS);// InstanceInfo replicator
instanceInfoReplicator = new InstanceInfoReplicator(this,instanceInfo,clientConfig.getInstanceInfoReplicationIntervalSeconds(),2); // burstSize

这个定时任务会，定时执行renew()方法

 private class HeartbeatThread implements Runnable {public void run() {if (renew()) {lastSuccessfulHeartbeatTimestamp = System.currentTimeMillis();}}}

renew()方法源码

 boolean renew() {EurekaHttpResponse<InstanceInfo> httpResponse;try {// AbstractJerseyEurekaHttpClient客户端:发送rest请求到`urlPath = "apps/" + appName + '/' + id;`进行续约信息httpResponse = eurekaTransport.registrationClient.sendHeartBeat(instanceInfo.getAppName(), instanceInfo.getId(), instanceInfo, null);logger.debug(PREFIX + "{} - Heartbeat status: {}", appPathIdentifier, httpResponse.getStatusCode());if (httpResponse.getStatusCode() == Status.NOT_FOUND.getStatusCode()) {REREGISTER_COUNTER.increment();logger.info(PREFIX + "{} - Re-registering apps/{}", appPathIdentifier, instanceInfo.getAppName());long timestamp = instanceInfo.setIsDirtyWithTime();boolean success = register();if (success) {instanceInfo.unsetIsDirty(timestamp);}return success;}return httpResponse.getStatusCode() == Status.OK.getStatusCode();} catch (Throwable e) {logger.error(PREFIX + "{} - was unable to send heartbeat!", appPathIdentifier, e);return false;}}

在EurekaInstanceConfigBean中，默认30s进行一次服务续约

private int leaseRenewalIntervalInSeconds = 30;

三、Eureka客户端服务发现

3.1、拉取方法：DiscoveryClient#fetchRegistry()

DiscoveryClient#fetchRegistry()方法，就是客户端拉取服务的方法，它在两个地方被分别调用：

• DiscoveryClient初始化的时候调用
• 定时任务线程定时调用

3.1.1、DiscoveryClient初始化调用

DiscoveryClient初始化的时候，会判断是否需要拉取服务列表，需要则主动调用：fetchRegistry(false)方法

// 根据配置，去服务端拉取服务
if (clientConfig.shouldFetchRegistry() && !fetchRegistry(false)) {fetchRegistryFromBackup();
}

3.1.2、CacheRefreshThread定时任务线程调用

此处调用同Heartbeat服务续约，同样是在DiscoveryClient初始化时，创建了一个定时任务线程，默认每30s调用一次refreshRegistry()方法

if (clientConfig.shouldFetchRegistry()) {// registry cache refresh timerint registryFetchIntervalSeconds = clientConfig.getRegistryFetchIntervalSeconds();int expBackOffBound = clientConfig.getCacheRefreshExecutorExponentialBackOffBound();// 创建了一个定时任务线程，默认每30s执行一次本地服务缓存刷新，其中会调用服务拉取方法cacheRefreshTask = new TimedSupervisorTask("cacheRefresh",scheduler,cacheRefreshExecutor,registryFetchIntervalSeconds,TimeUnit.SECONDS,expBackOffBound,new CacheRefreshThread());scheduler.schedule(cacheRefreshTask,registryFetchIntervalSeconds, TimeUnit.SECONDS);
}

fetchRegistry()源码分析

上面那是两处调用fetchRegistry()方法的地方，下面看下它的源码

fetchRegistry()方法分析：根据不同条件判断，缓存更新分为全量更新和增量更新

private boolean fetchRegistry(boolean forceFullRegistryFetch) {Stopwatch tracer = FETCH_REGISTRY_TIMER.start();try {// If the delta is disabled or if it is the first time, get all applications// 读取本地缓存服务列表信息Applications applications = getApplications();// 满足以下其中一项就进行全量拉取// 1. 禁用增量更新；// 2. 特别关注region为空； // 3. 强制指定全量更新； // 4. 本地缓存版本是-1，即：还未缓存过有效的服务列表if (clientConfig.shouldDisableDelta()|| (!Strings.isNullOrEmpty(clientConfig.getRegistryRefreshSingleVipAddress()))|| forceFullRegistryFetch|| (applications == null)|| (applications.getRegisteredApplications().size() == 0)|| (applications.getVersion() == -1)) //Client application does not have latest library supporting delta{logger.info("Disable delta property : {}", clientConfig.shouldDisableDelta());logger.info("Single vip registry refresh property : {}", clientConfig.getRegistryRefreshSingleVipAddress());logger.info("Force full registry fetch : {}", forceFullRegistryFetch);logger.info("Application is null : {}", (applications == null));logger.info("Registered Applications size is zero : {}",(applications.getRegisteredApplications().size() == 0));logger.info("Application version is -1: {}", (applications.getVersion() == -1));//全量更新服务列表：存储到localRegionApps中getAndStoreFullRegistry();} else {//增量更新服务列表：存储到localRegionApps中getAndUpdateDelta(applications);}applications.setAppsHashCode(applications.getReconcileHashCode());logTotalInstances();} catch (Throwable e) {logger.error(PREFIX + "{} - was unable to refresh its cache! status = {}", appPathIdentifier, e.getMessage(), e);return false;} finally {if (tracer != null) {tracer.stop();}}// Notify about cache refresh before updating the instance remote statusonCacheRefreshed();// Update remote status based on refreshed data held in the cacheupdateInstanceRemoteStatus();// registry was fetched successfully, so return truereturn true;
}

在EurekaClientConfigBean中，默认30s进行一次服务缓存更新

private int registryFetchIntervalSeconds = 30;

四、定时任务

在DiscoveryClient中，使用了定时会任务去完成：心跳续约、服务拉取等业务，我们可以通过这个案例学习如何使用定时任务执行任务。

DiscoveryClient的定时任务是自定义的TimedSupervisorTask类，其中有些思想值得学习借鉴：
1、如何固定时间间隔执行任务
2、当服务调用timeout时，如何让下次任务调用时间间隔翻倍

定时任务线程池：ScheduledExecutorService

// default size of 2 - 1 each for heartbeat and cacheRefresh
scheduler = Executors.newScheduledThreadPool(2,new ThreadFactoryBuilder().setNameFormat("DiscoveryClient-%d").setDaemon(true).build());

定时任务线程：TimedSupervisorTask

@Overridepublic void run() {Future<?> future = null;try {future = executor.submit(task);threadPoolLevelGauge.set((long) executor.getActiveCount());// 根据间隔时间读取数据future.get(timeoutMillis, TimeUnit.MILLISECONDS);  // block until done or timeout// 设置下次任务执行间隔时间delay.set(timeoutMillis);threadPoolLevelGauge.set((long) executor.getActiveCount());successCounter.increment();} catch (TimeoutException e) {logger.warn("task supervisor timed out", e);timeoutCounter.increment();// 获取当前任务延迟时间long currentDelay = delay.get();// 读取超时，下次执行时间是，延迟累计时间翻倍 or 指定的最大间隔时间 两个中的最小值long newDelay = Math.min(maxDelay, currentDelay * 2);delay.compareAndSet(currentDelay, newDelay);} catch (RejectedExecutionException e) {if (executor.isShutdown() || scheduler.isShutdown()) {logger.warn("task supervisor shutting down, reject the task", e);} else {logger.warn("task supervisor rejected the task", e);}rejectedCounter.increment();} catch (Throwable e) {if (executor.isShutdown() || scheduler.isShutdown()) {logger.warn("task supervisor shutting down, can't accept the task");} else {logger.warn("task supervisor threw an exception", e);}throwableCounter.increment();} finally {if (future != null) {future.cancel(true);}// 设置下次任务执行时间if (!scheduler.isShutdown()) {scheduler.schedule(this, delay.get(), TimeUnit.MILLISECONDS);}}}