Eureka(三)——client注册过程

发表于

服务注册是Spring Cloud的关键功能，本文着重来分析eureka-client服务注册的过程。

服务注册

要想将一个服务注册到Eureka Server非常简单：

  1. 在pom.xml文件中加入Eureka Server依赖

    1
    2
    3
    4
    <dependency>
        <groupId>org.springframework.cloud</groupId>
        <artifactId>spring-cloud-starter-netflix-eureka-client</artifactId>
    </dependency>

  2. 在启动类上添加注解@EnableDiscoveryClient或@EnableEurekaClient

    1
    2
    3
    4
    5
    6
    7
    @EnableDiscoveryClient
    @SpringBootApplication
    public class EurekaClientApplication {
        public static void main(String[] args) {
            SpringApplication.run(EurekaClientApplication.class, args);
        }
    }

  3. 配置

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    spring:
      application:
        name: eureka-client

    server:
      port: 8762

    eureka:
      instance:
        hostname: localhost
      client:
        serviceUrl:
          defaultZone: http://${eureka.instance.hostname}:8761/eureka/

从Spring Cloud Edgware开始，注解@EnableDiscoveryClient或@EnableEurekaClient可省略。只需加上相关依赖，并进行相应配置，即可将微服务注册到服务发现组件上。

Eureka-client注册服务的原理

我们知道，在初始化过程中，SpringBoot会扫描spring.factories，加载其中的配置类。Eureka-client的spring.factories如下所示：

1
2
3
4
5
6
7
8
9
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

我们重点关注两个类：EurekaDiscoveryClientConfiguration和EurekaClientAutoConfiguration。

因为在EurekaClientAutoConfiguration有如下条件：

1
2
3
4
@ConditionalOnBean(EurekaDiscoveryClientConfiguration.Marker.class)
@AutoConfigureAfter(name = {"org.springframework.cloud.autoconfigure.RefreshAutoConfiguration",
        "org.springframework.cloud.netflix.eureka.EurekaDiscoveryClientConfiguration",
        "org.springframework.cloud.client.serviceregistry.AutoServiceRegistrationAutoConfiguration"})

因此首先执行EurekaDiscoveryClientConfiguration配置类，加载其中的配置，然后再去执行EurekaClientAutoConfiguration。

EurekaClientAutoConfiguration

EurekaClientAutoConfiguration类的主要功能是配置EurekaClient。其中有个关键的内部类RefreshableEurekaClientConfiguration：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
@Configuration
@ConditionalOnRefreshScope
protected static class RefreshableEurekaClientConfiguration {

    @Autowired
    private ApplicationContext context;

    @Autowired
    private AbstractDiscoveryClientOptionalArgs<?> optionalArgs;

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

    @Bean
    @ConditionalOnMissingBean(value = ApplicationInfoManager.class, search = SearchStrategy.CURRENT)
    @org.springframework.cloud.context.config.annotation.RefreshScope
    @Lazy
    public ApplicationInfoManager eurekaApplicationInfoManager(EurekaInstanceConfig config) {
        InstanceInfo instanceInfo = new InstanceInfoFactory().create(config);
        return new ApplicationInfoManager(config, instanceInfo);
    }

}

RefreshableEurekaClientConfiguration的注释@ConditionOnRefreshScope定义如下：

1
2
3
4
5
6
7
8
@Target({ ElementType.TYPE, ElementType.METHOD })
@Retention(RetentionPolicy.RUNTIME)
@Documented
@ConditionalOnClass(RefreshScope.class)
@ConditionalOnBean(RefreshAutoConfiguration.class)
@interface ConditionalOnRefreshScope {

}

因为在spring-cloud-context包的spring.factories中有这样的配置：

1
2
3
4
5
6
org.springframework.boot.autoconfigure.EnableAutoConfiguration=\
org.springframework.cloud.autoconfigure.ConfigurationPropertiesRebinderAutoConfiguration,\
org.springframework.cloud.autoconfigure.LifecycleMvcEndpointAutoConfiguration,\
org.springframework.cloud.autoconfigure.RefreshAutoConfiguration,\
org.springframework.cloud.autoconfigure.RefreshEndpointAutoConfiguration,\
org.springframework.cloud.autoconfigure.WritableEnvironmentEndpointAutoConfiguration

其中包含了RefreshAutoConfiguration，所以@ConditionalOnRefreshScope注释上的@ConditionalOnBean(RefreshAutoConfiguration.class)是生效的。因此Spring会创建RefreshableEurekaClientConfiguration，包括其中的EurekaClient，其实例为CloudEurekaClient。

CloudEurekaClient

CloudEurekaClient的继承关系如下：

屏幕快照 2018-06-27 下午2.13.26

它继承了com.netflix.discovery.DiscoveryClient。在构造方法中调用了initScheduledTasks()方法来初始化各种定时任务：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
public class DiscoveryClient implements EurekaClient {
    ...
    /**
     * 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是一个Runnable接口实现类，服务注册的逻辑在run方法中。它的功能是检查服务状态，向Server中注册服务
            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
            instanceInfoReplicator.start(clientConfig.getInitialInstanceInfoReplicationIntervalSeconds());
        } else {
            logger.info("Not registering with Eureka server per configuration");
        }
    }
    ...
}

InstanceInfoReplicator服务注册的定时任务

InstanceInfoReplicator类的功能是更新本地的服务实例信息，并将本地的服务实例信息复制到注册服务中。其定时周期由EurekaClientConfig.getInstanceInfoReplicationIntervalSeconds()控制，默认为30秒。

DiscoveryClient.initScheduledTasks()方法中调用InstanceInfoReplicator.start()方法启动InstanceInfoReplicator：

1
2
3
4
5
6
7
public void start(int initialDelayMs) {
    if (started.compareAndSet(false, true)) {
        instanceInfo.setIsDirty();  // for initial register
        Future next = scheduler.schedule(this, initialDelayMs, TimeUnit.SECONDS);
        scheduledPeriodicRef.set(next);
    }
}

流程如下：

  1. 调用compareAndSet确保只有InstanceInfoReplicator只启动一个定时任务。
  2. 调用setIsDirty将本地的实例信息设置为脏数据，这是为了向注册中心注册服务。
  3. 然后调用schedule启动延时任务

延时任务的执行逻辑在run方法中：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
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);
    }
}

流程如下：

  1. 调用DiscoveryClient.refreshInstanceInfo刷新当前本地的实例信息。如果发生了改变，实例信息中的isDirty标记会被设置为true
  2. 检查实例信息是否发生了改变，如果发生改变则调用DiscoveryClient.register方法将实例信息注册到注册中心
  3. 再次调用schedule启用延时任务，相当于周期性的执行InstanceInfoReplicator任务
刷新实例信息

刷新实例信息的任务在DiscoveryClient.refreshInstanceInfo方法中，主要流程如下：

  1. 调用ApplicationInfoManager.refreshDataCenterInfoIfRequired检查Server的hostname是否发生了修改
  2. 调用ApplicationInfoManager.refreshLeaseInfoIfRequired检查lease.duration和lease.renewalInterval两个续约配置是否发生了修改
  3. 调用HealthCheckCallbackToHandlerBridge.getStatus获取服务实例的状态，并设置服务实例的状态
注册服务实例信息

注册服务实例信息的任务在DiscoveryClient.register方法中，其最终调用的方法是AbstractJerseyEurekaHttpClient.register(InstanceInfo info)。该方法向Eureka Server发送POST请求，请求的urlPath是apps/EUREKA-CLIENT，EUREKA-CLIENT是服务实例的名称，在本例中完整的请求地址是http://localhost:8761/eureka/apps/EUREKA-CLIENT。将服务实例信息InstanceInfo通过POST请求发送给Eureka Server完成注册。

刷新服务列表缓存的定时任务

刷新服务列表缓存的定时任务由以下代码启动:

1
2
3
4
5
6
7
8
9
10
11
scheduler.schedule(
        new TimedSupervisorTask(
                "cacheRefresh",
                scheduler,
                cacheRefreshExecutor,
                registryFetchIntervalSeconds,
                TimeUnit.SECONDS,
                expBackOffBound,
                new CacheRefreshThread()
        ),
        registryFetchIntervalSeconds, TimeUnit.SECONDS);

其定时周期由EurekaClientConfig.getRegistryFetchIntervalSeconds()控制，默认为30秒。

在TimedSupervisorTask中也是通过循环调用schedule的方式形成一个周期任务，以定时执行CacheRefreshThread线程。

CacheRefreshThread的任务就是调用refreshRegistry方法：

1
2
3
4
5
class CacheRefreshThread implements Runnable {
    public void run() {
        refreshRegistry();
    }
}
refreshRegistry

refreshRegistry的任务是获取注册的所有服务，主要的流程如下：

  1. 调用EurekaClientConfig.fetchRegistryForRemoteRegions重新获取注册中心的地址，因为这些配置有可能发生动态修改。
  2. 调用DiscoveryClient.fetchRegistry获取注册信息
fetchRegistry

fetchRegistry的任务是获取注册信息，主要流程如下：

  1. 调用DiscoveryClient.getApplications方法获取本地保存的所有注册信息
  2. 判断是否需要获取全量注册信息，分别调用DiscoveryClient.getAndStoreFullRegistry和DiscoveryClient.getAndUpdateDelta方法
  3. 对注册信息设置HashCode
  4. 调用onCacheRefreshed广播CacheRefreshedEvent事件通知服务发生改变。例如Ribbon收到事件后会更新它保存的服务信息
  5. 调用updateInstanceRemoteStatus更新实例的状态
getAndStoreFullRegistry

DiscoveryClient.getAndStoreFullRegistry的任务是从Eureka Server中获取所有的注册信息，然后保存在本地。主要流程如下：

  1. 调用EurekaHttpClientDecorator.getApplications方法，返回EurekaHttpResponse

    最终调用的是AbstractJerseyEurekaHttpClient.getApplicationsInternal，该方法向Eureka Server发送请求，请求的urlPath是apps/，在本例中完整的请求地址是http://localhost:8761/eureka/apps/。根据返回的数据生成Applications实例。

  2. 如果返回的Applications不为null，将其保存到DiscoveryClient.localRegionApps变量中

getAndUpdateDelta

DiscoveryClient.getAndUpdateDelta的任务是从Eureka Server中获取增量注册信息，更新本地保存的信息。主要流程如下：

  1. 调用EurekaHttpClientDecorator.getDelta方法，

    最终调用的是AbstractJerseyEurekaHttpClient.getApplicationsInternal，该方法向Eureka Server发送请求，请求的urlPath是apps/delta，在本例中完整的请求地址是http://localhost:8761/eureka/apps/delta。根据返回的数据生成Applications实例。

  2. 如果返回的delta为null，则调用getAndStoreFullRegistry获取全量的数据

  3. 调用Discovery.updateDelta更新增量服务信息

    遍历增量服务信息，根据服务信息的操作类型（新增、修改、删除）进行相应的处理

服务续约的定时任务

服务续约的定时任务由以下代码启动:

1
2
3
4
5
6
7
8
9
10
11
scheduler.schedule(
        new TimedSupervisorTask(
                "heartbeat",
                scheduler,
                heartbeatExecutor,
                renewalIntervalInSecs,
                TimeUnit.SECONDS,
                expBackOffBound,
                new HeartbeatThread()
        ),
        renewalIntervalInSecs, TimeUnit.SECONDS);

在TimedSupervisorTask中也是通过循环调用schedule的方式形成一个周期任务，以定时执行HeartbeatThread线程。其定时周期由LeaseInfo.renewalIntervalInSecs变量控制，默认为DEFAULT_LEASE_RENEWAL_INTERVAL，即30秒。

HeartbeatThread的任务就是调用renew方法：

1
2
3
4
5
6
7
private class HeartbeatThread implements Runnable {
    public void run() {
        if (renew()) {
            lastSuccessfulHeartbeatTimestamp = System.currentTimeMillis();
        }
    }
}
renew

DiscoveryClient.renew方法的任务是向Eureka Server发送服务续约请求，流程如下：

  1. 调用EurekaHttpClientDecorator.sendHeartBeat发送心跳请求

    最终调用的是AbstractJerseyEurekaHttpClient.sendHeartBeat，该方法向Eureka Server发送心跳请求，请求的urlPath是apps/EUREKA-CLIENT/wangqideimac.lan:eureka-client:8762，在本例中完整的请求地址是http://localhost:8761/eureka/apps/EUREKA-CLIENT/wangqideimac.lan:eureka-client:8762?status=UP&lastDirtyTimestamp=1530096210220。其中EUREEKA-CLIENT表示服务名称，wangqideimac.lan:eureka-client:8762表示服务id，status表示服务的状态，lastDirtyTimestamp表示实例更新的时间。

  2. 如果返回的状态是404，表示当前服务还没有注册过，于是调用register方法向Eureka Server注册服务。

服务下线

服务下线一般在服务关闭(shut down)的时候调用，用来把自身的服务从Eureka Server中删除，以防客户端调用不存在的服务。

服务下线的任务在Discovery.unregister方法中，其最终调用的方法是AbstractJerseyEurekaHttpClient.cancel(String appName, String id)。该方法向Eureka Server发送DELETE请求，请求的urlPath是apps/EUREKA-CLIENT/wangqideimac.lan:eureka-client:8762，其中EUREEKA-CLIENT表示服务名称，wangqideimac.lan:eureka-client:8762表示服务id。在本例中完整的请求地址是http://localhost:8761/eureka/apps/EUREKA-CLIENT/wangqideimac.lan:eureka-client:8762。

总结

经过前文的总结，我们可以看到，服务注册的过程是围绕CloudEurekaClient来进行的，它的父类DiscoveryClient在初始化的过程中会调用initScheduledTasks方法，其中会创建三个定时任务：

  • 服务注册。

    其定时周期由EurekaClientConfig.getInstanceInfoReplicationIntervalSeconds()控制，默认为30秒。

    调用DiscoveryClient.register方法。向Eureka Server发送请求，请求的urlPath是”apps/{app_name}”，app_name是服务实例的名称。将服务实例信息InstanceInfo通过POST请求发送给Eureka Server完成注册。

    本示例完整的请求实例：”http://localhost:8761/eureka/apps/EUREKA-CLIENT/"

  • 服务续约。

    其定时周期由LeaseInfo.renewalIntervalInSecs变量控制，默认为DEFAULT_LEASE_RENEWAL_INTERVAL，即30秒。

    调用DiscoveryClient.renew方法。向Eureka Server发送请求，请求的urlPath是”apps/{app_name}/{id}:8762?status={status}&lastDirtyTimestamp={timestamp}”，其中app_name表示服务名称，id表示服务id，status表示服务的状态，timestamp表示实例更新的时间。

    本示例完整的请求实例：”http://localhost:8761/eureka/apps/EUREKA-CLIENT/wangqideimac.lan:eureka-client:8762?status=UP&lastDirtyTimestamp=1530096210220"

  • 刷新服务列表缓存。调用DiscoveryClient.fetchRegistry方法，

    其定时周期由EurekaClientConfig.getRegistryFetchIntervalSeconds()控制，默认为30秒。

    向Eureka Server发送请求，全量请求的urlPath是”apps/“，增量请求的urlPath是”apps/delta”

http://blog.didispace.com/spring-cloud-eureka-register-detail/
https://my.oschina.net/u/3039671/blog/1546168
http://www.itmuch.com/spring-cloud/edgware-new-optional-enable-discovery-client/
https://blog.csdn.net/Mr_rain/article/details/78790292