SpringCloud 源码系列（1）—— 注册中心 Eureka（上）
2024-08-26 05:33:22
Eureka 是 Netflix 公司开源的一个服务注册与发现的组件，和其他 Netflix 公司的服务组件（例如负载均衡、熔断器、网关等）一起，被 Spring Cloud 整合为 Spring Cloud Netflix 模块。不过 Eureka 2.0 开始闭源了，但 1.x 还在继续维护中，可以继续使用。这篇文章就来深入学习下 Eureka 注册中心，便于我们更好的使用和调优注册中心。
关于版本：本文章使用的 Spring cloud 版本为 Hoxton.SR8，Spring boot 版本为 2.3.3.RELEASE，依赖的 eureka 版本则为 1.9.25。
一、Eureka 初体验
Eureka 分为 Eureka Server 和 Eureka Client，Eureka Server 为 Eureka 注册中心，Eureka Client 为 Eureka 客户端。这节先通过demo把注册中心的架子搭起来，看看注册中心的基础架构。
1、Eureka Server
① 创建注册中心服务：sunny-register
首先创建一个 maven 工程，服务名称为 sunny-register，并在 pom.xml 中引入注册中心服务端的依赖。
1 <dependencies>
2 <dependency>
3 <groupId>org.springframework.cloud</groupId>
4 <artifactId>spring-cloud-starter-netflix-eureka-server</artifactId>
5 </dependency>
6 </dependencies>
② 添加配置文件
在 resources 下添加 application.yml 配置文件，并添加注册中心相关配置。
1 server:
2 port: 8000
3 spring:
4 application:
5 name: sunny-register
6
7 eureka:
8 instance:
9 hostname: dev.lyyzoo.com
10 client:
11 # 是否向注册中心注册自己
12 register-with-eureka: false
13 # 是否检索服务
14 fetch-registry: false
15 service-url:
16 defaultZone: http://${eureka.instance.hostname}:${server.port}/eureka/
③ 添加启动类
添加启动类，并在启动类上加上 @EnableEurekaServer 注解，启用注册中心。
1 package com.lyyzoo.sunny.register;
2
3 import org.springframework.boot.SpringApplication;
4 import org.springframework.boot.autoconfigure.SpringBootApplication;
5 import org.springframework.cloud.netflix.eureka.server.EnableEurekaServer;
6
7 @EnableEurekaServer
8 @SpringBootApplication
9 public class RegisterApplication {
10
11 public static void main(String[] args) {
12 SpringApplication.run(RegisterApplication.class, args);
13 }
14 }
④ 启动注册中心
启动注册中心后，访问 http://dev.lyyzoo.com:8000/，就可以看到注册中心的页面了，现在还没有实例注册上来。（dev.lyyzoo.com 在本地 hosts 文件中映射到 127.0.0.1）
2、Eureka Client
创建两个 demo 服务，demo-producer 服务作为生产者提供一个接口，demo-consumer 服务作为消费者去调用 demo-producer 的接口。
① 创建客户端服务：demo-producer
创建maven工程，服务名称为 demo-producer，在 pom.xml 中引入注册中心客户端的依赖，并添加了 web 的依赖。
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-starter-netflix-eureka-client</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
② 添加配置文件
在 resouces 下添加 application.yml 配置文件，添加注册中心客户端相关的配置。
1 server:
2 port: 8010
3 spring:
4 application:
5 name: demo-producer
6
7 eureka:
8 client:
9 serviceUrl:
10 defaultZone: ${EUREKA_DEFAULT_ZONE:http://dev.lyyzoo.com:8000/eureka}
③ 添加启动类
添加启动类，并在启动类上加上 @EnableEurekaClient 注解，启用客户端。
1 @EnableEurekaClient
2 @SpringBootApplication
3 public class ProducerApplication {
4
5 public static void main(String[] args) {
6 SpringApplication.run(ProducerApplication.class, args);
7 }
8 }
④ 添加一个 rest 接口
添加一个接口用于测试调用：
1 @RestController
2 public class DemoController {
3
4 private final Logger logger = LoggerFactory.getLogger(getClass());
5
6 @GetMapping("/v1/uuid")
7 public ResponseEntity<String> getUUID() {
8 String uuid = UUID.randomUUID().toString();
9 logger.info("generate uuid: {}", uuid);
10 return ResponseEntity.ok(uuid);
11 }
12 }
⑤ 创建客户端服务：demo-consumer
类似的方式，再创建消费者服务：demo-producer，这个服务中添加一个消费者接口，通过 RestTemplate 负载均衡的方式来调用 demo-producer 的接口。
因此需要先配置一个带有负载均衡的 RestTemplate：
1 @EnableEurekaClient
2 @SpringBootApplication
3 public class ConsumerApplication {
4
5 @Bean
6 @LoadBalanced
7 public RestTemplate restTemplate() {
8 return new RestTemplate();
9 }
10
11 public static void main(String[] args) {
12 SpringApplication.run(ConsumerApplication.class, args);
13 }
14 }
添加消费者接口，注意这里 url 是写的服务名称，并不是具体的 ip 地址或端口，在微服务场景下，服务间调用也不可能写死某个具体的地址。
1 @RestController
2 public class DemoController {
3
4 private final Logger logger = LoggerFactory.getLogger(getClass());
5
6 @Autowired
7 private RestTemplate restTemplate;
8
9 @GetMapping("/v1/id")
10 public ResponseEntity<String> getId() {
11 ResponseEntity<String> result = restTemplate.getForEntity("http://demo-producer/v1/uuid", String.class);
12 String uuid = result.getBody();
13 logger.info("request id: {}", uuid);
14 return ResponseEntity.ok(uuid);
15 }
16 }
⑥ 启动注册中心客户端
以两个不同的端口启动 demo-producer，可以通过环境变量的方式制定端口。然后再启动 demo-consumer。
启动完成之后，就可以在注册中心看到注册上来的两个 demo-producer 实例和一个 demo-consumer 实例，并且状态都为 UP。
⑦ 测试接口
调用消费者服务的接口，多次访问 http://dev.lyyzoo.com:8020/v1/id 接口，会发现生产者服务 demo-consumer 两个实例的控制台会交替的输出日志信息。这就说明消费者客户端通过服务名称访问到生产者了。
3、Eureka 基础架构
通过前面的体验，可以发现，服务间调用只需知道某个服务的名称就可以调用这个服务的api了，而不需要指定具体的ip地址和端口，那这是怎么做到的呢？
不难看出，Eureka 的基础架构包含三种角色：
- 服务注册中心：Eureka Server，提供服务注册和发现的功能
- 服务提供者：Eureka Client，提供服务（本身也可以作为消费者）
- 服务消费者：Eureka Client，消费服务（本身也可以作为提供者）
首先需要一个服务注册中心，客户端则向注册中心注册，将自己的信息（比如服务名、服务的 IP 地址和端口信息等）提交给注册中心。客户端向注册中心获取一份服务注册列表的信息，该列表包含了所有向注册中心注册的服务信息。获取服务注册列表信息之后，客户端服务就可以根据服务名找到服务的所有实例，然后通过负载均衡选择其中一个实例，根据其 IP 地址和端口信息，就可以调用服务的API接口了。
这就是注册中心最基础的架构和功能了，提供服务注册和发现，为各个客户端提供服务注册列表信息。但为了完成这些工作，Eureka 有很多的机制来实现以及保证其高可用，如服务注册、服务续约、获取服务注册列表、服务下线、服务剔除等等。Eureka 也提供了很多参数让我们可以根据实际的场景来优化它的一些功能和配置，比如维持心跳的时间、拉取注册表的间隔时间、自我保护机制等等。下面我们就从 eureka 的源码层面来分析下 eureka 的这些功能以及参数，理解其原理，学习它的一些设计。
二、Eureka 源码准备
虽然我们在 pom.xml 中依赖的是 spring-cloud-starter-netflix-eureka-server 和 spring-cloud-starter-netflix-eureka-client，但 spring-cloud-starter-netflix 只是对 eureka 做了封装，使得其可以通过 springboot 的方式来启动和初始化，其底层其实是 netflix 的 eureka-core、eureka-client 等。所以我们先分析 netflix eureka 的源码，最后再看看 spring-cloud-starter-netflix 的源码。
1、源码环境准备
① 下载源码
Netflix Eureka：https://github.com/Netflix/eureka
Spring Cloud Netflix：https://github.com/spring-cloud/spring-cloud-netflix
克隆 eureka 的源码到本地：
$ git clone https://github.com/Netflix/eureka.git
由于我们依赖的是 1.9.25 版本，将代码克隆到本地后，将其切换到 1.9.25：
$ git checkout -b 1.9.25
然后到 eureka 根目录下执行构建的命令：
$ ./gradlew clean build -x test
② IDEA 打开源码
由于 eureka 使用 gradle 管理依赖，所以本地需要先安装 gradle，之后 IDEA 中也需要安装 gradle 的插件，跟 maven 都是类似的，安装教程可自行百度。
2、Eureka 工程结构
Eureka 主要包含如下模块：
- eureka-client：eureka 客户端
- eureka-core：eureka 服务端，注册中心的核心功能
- eureka-resources：基于jsp的eureka控制台，可以查看注册了哪些服务实例
- eureka-server：注册中心，集成了 eureka-client、eureka-core、eureka-resources，因为依赖了 eureka-client，因此 eureka-server 也是一个客户端，在 eureka server 集群模式下，eureka-server 也会作为客户端注册到其它注册中心上
- eureka-examples：eureka 例子
- eureka-test-utils：eureka 单元测试工具
- eureka-core|client-jersey2：对 jersey 框架的封装，jersey 类似于 spring mvc，支持 http restful 请求，eureka-client 和 eureka-server 之间的通信就是基于 jersey 框架来的
三、Eureka Server 启动初始化
首先要看的是 eureka-server，注册中心启起来之后，客户端才能来注册服务和发现服务。
1、eureka-server 模块
① eureka-server 目录
- resources 目录中主要是 eureka client 和 server 的配置文件
- webapp 下有一个 web.xml 配置文件，这里面就配置了启动初始化的入口，从这也可以看出，eureka-server 会被打包成 war 包来运行
- test 下有个单元测试类 EurekaClientServerRestIntegrationTest，这里面就包含了服务注册、续约、下线等单元测试，我们就可以运行这些单元测试来调试代码
② web.xml
web.xml 的内容：
1 <?xml version="1.0" encoding="UTF-8"?>
2 <web-app version="2.5"
3 xmlns="http://java.sun.com/xml/ns/javaee"
4 xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
5 xsi:schemaLocation="http://java.sun.com/xml/ns/javaee
6 http://java.sun.com/xml/ns/javaee/web-app_2_5.xsd">
7 <!-- eureka 启动初始化类 -->
8 <listener>
9 <listener-class>com.netflix.eureka.EurekaBootStrap</listener-class>
10 </listener>
11
12 <!-- 状态过滤器 -->
13 <filter>
14 <filter-name>statusFilter</filter-name>
15 <filter-class>com.netflix.eureka.StatusFilter</filter-class>
16 </filter>
17
18 <!-- 认证过滤器 -->
19 <filter>
20 <filter-name>requestAuthFilter</filter-name>
21 <filter-class>com.netflix.eureka.ServerRequestAuthFilter</filter-class>
22 </filter>
23
24 <!-- 限流过滤器 -->
25 <filter>
26 <filter-name>rateLimitingFilter</filter-name>
27 <filter-class>com.netflix.eureka.RateLimitingFilter</filter-class>
28 </filter>
29 <filter>
30 <filter-name>gzipEncodingEnforcingFilter</filter-name>
31 <filter-class>com.netflix.eureka.GzipEncodingEnforcingFilter</filter-class>
32 </filter>
33
34 <!-- jersey 容器 -->
35 <filter>
36 <filter-name>jersey</filter-name>
37 <filter-class>com.sun.jersey.spi.container.servlet.ServletContainer</filter-class>
38 <init-param>
39 <param-name>com.sun.jersey.config.property.WebPageContentRegex</param-name>
40 <param-value>/(flex|images|js|css|jsp)/.*</param-value>
41 </init-param>
42 <init-param>
43 <param-name>com.sun.jersey.config.property.packages</param-name>
44 <param-value>com.sun.jersey;com.netflix</param-value>
45 </init-param>
46
47 <!-- GZIP content encoding/decoding -->
48 <init-param>
49 <param-name>com.sun.jersey.spi.container.ContainerRequestFilters</param-name>
50 <param-value>com.sun.jersey.api.container.filter.GZIPContentEncodingFilter</param-value>
51 </init-param>
52 <init-param>
53 <param-name>com.sun.jersey.spi.container.ContainerResponseFilters</param-name>
54 <param-value>com.sun.jersey.api.container.filter.GZIPContentEncodingFilter</param-value>
55 </init-param>
56 </filter>
57
58 <filter-mapping>
59 <filter-name>statusFilter</filter-name>
60 <url-pattern>/*</url-pattern>
61 </filter-mapping>
62
63 <filter-mapping>
64 <filter-name>requestAuthFilter</filter-name>
65 <url-pattern>/*</url-pattern>
66 </filter-mapping>
67
68 <!-- Uncomment this to enable rate limiter filter.
69 <filter-mapping>
70 <filter-name>rateLimitingFilter</filter-name>
71 <url-pattern>/v2/apps</url-pattern>
72 <url-pattern>/v2/apps/*</url-pattern>
73 </filter-mapping>
74 -->
75
76 <filter-mapping>
77 <filter-name>gzipEncodingEnforcingFilter</filter-name>
78 <url-pattern>/v2/apps</url-pattern>
79 <url-pattern>/v2/apps/*</url-pattern>
80 </filter-mapping>
81
82 <filter-mapping>
83 <filter-name>jersey</filter-name>
84 <url-pattern>/*</url-pattern>
85 </filter-mapping>
86
87 <!-- 欢迎页 -->
88 <welcome-file-list>
89 <welcome-file>jsp/status.jsp</welcome-file>
90 </welcome-file-list>
91
92 </web-app>
web.xml 中可以得知如下信息：
- eureka server 启动时首先通过 com.netflix.eureka.EurekaBootStrap 类来进行启动初始化相关的工作
- 配置了 StatusFilter（server 状态过滤器）、ServerRequestAuthFilter（认证过滤器）、RateLimitingFilter（限流过滤器） 等过滤器，但 RateLimitingFilter 默认没有启用
- 配置了 jersey 的 servlet 容器，其实就跟 springframework 的 DispatcherServlet 是类似的，用来拦截处理 http restful 请求，这块我们不用过于关注
- 最后还配置了 eureka server 的欢迎页为 jsp/status.jsp 页面，这个页面在 eureka-resources 模块下，也就是前面看到的 eureka 控制台页面
③ 单元测试类 EurekaClientServerRestIntegrationTest
首先看 setUp 方法，每个测试用例运行之前都会先运行 setUp 方法来初始化运行环境。
1 @BeforeClass
2 public static void setUp() throws Exception {
3 // 初始化 eureka 配置
4 injectEurekaConfiguration();
5 // 启动 eureka server，会找 build/libs 目录下的 eureka-server.*.war 包来运行
6 // 这一步启动时，就会加载 web.xm 配置文件，然后进入 EurekaBootStrap 初始化类
7 startServer();
8 // eureka server 配置
9 createEurekaServerConfig();
10
11 // 创建 jersey 客户端，使用 jersey 客户端来调用资源
12 httpClientFactory = JerseyEurekaHttpClientFactory.newBuilder()
13 .withClientName("testEurekaClient")
14 .withConnectionTimeout(1000)
15 .withReadTimeout(1000)
16 .withMaxConnectionsPerHost(1)
17 .withMaxTotalConnections(1)
18 .withConnectionIdleTimeout(1000)
19 .build();
20
21 jerseyEurekaClient = httpClientFactory.newClient(new DefaultEndpoint(eurekaServiceUrl));
22
23 ServerCodecs serverCodecs = new DefaultServerCodecs(eurekaServerConfig);
24 jerseyReplicationClient = JerseyReplicationClient.createReplicationClient(
25 eurekaServerConfig,
26 serverCodecs,
27 eurekaServiceUrl
28 );
29 }
这个类提供了如下的一些测试用例，我们可以运行这些测试用例来进行调试。
2、EurekaBootStrap 初始化
EurekaBootStrap 是监听器的入口，实现了 ServletContextListener 接口，主要完成了 eureka server 的启动初始化。
从 contextInitialized 方法进去，整体上来说，分为 eureka 环境初始化和 eureka server 上下文初始化。
1 @Override
2 public void contextInitialized(ServletContextEvent event) {
3 try {
4 // eureka 环境初始化
5 initEurekaEnvironment();
6 // eureka server 上下文初始化
7 initEurekaServerContext();
8
9 ServletContext sc = event.getServletContext();
10 sc.setAttribute(EurekaServerContext.class.getName(), serverContext);
11 } catch (Throwable e) {
12 logger.error("Cannot bootstrap eureka server :", e);
13 throw new RuntimeException("Cannot bootstrap eureka server :", e);
14 }
15 }
① eureka环境初始化
initEurekaEnvironment 方法内主要是设置数据中心和运行环境参数：
- archaius.deployment.datacenter = default
- archaius.deployment.environment = test
② eureka server 上下文初始化
initEurekaServerContext 上下文初始化则包含了很多阶段：
- 构造 eureka 注册中心配置：EurekaServerConfig
- 构造 eureka 实例配置：EurekaInstanceConfig
- 构造实例信息：InstanceInfo
- 构造实例管理器：ApplicationInfoManager
- 构造 eureka 客户端配置：EurekaClientConfig
- 创建 eureka 客户端：EurekaClient（DiscoveryClient）
- 创建注册表（可以感知eureka集群的注册表）：PeerAwareInstanceRegistry
- 创建集群：PeerEurekaNodes
- 将信息封装到eureka上下文：EurekaServerContext
- 将eureka上下文放到一个全局容器中：EurekaServerContextHolder
- 初始化eureka上下文
- 同步eureka server的注册表
- 开启追踪
- 注册监控统计
1 protected void initEurekaServerContext() throws Exception {
2 // 1、eureka 注册中心配置
3 EurekaServerConfig eurekaServerConfig = new DefaultEurekaServerConfig();
4
5 // For backward compatibility
6 JsonXStream.getInstance().registerConverter(new V1AwareInstanceInfoConverter(), XStream.PRIORITY_VERY_HIGH);
7 XmlXStream.getInstance().registerConverter(new V1AwareInstanceInfoConverter(), XStream.PRIORITY_VERY_HIGH);
8
9 logger.info("Initializing the eureka client...");
10 logger.info(eurekaServerConfig.getJsonCodecName());
11 ServerCodecs serverCodecs = new DefaultServerCodecs(eurekaServerConfig);
12
13 ApplicationInfoManager applicationInfoManager = null;
14
15 if (eurekaClient == null) {
16 // 2、eureka 实例配置
17 EurekaInstanceConfig instanceConfig = isCloud(ConfigurationManager.getDeploymentContext())
18 ? new CloudInstanceConfig()
19 : new MyDataCenterInstanceConfig();
20
21 // 3、构造 InstanceInfo 实例信息
22 // 4、构造 ApplicationInfoManager 应用管理器
23 applicationInfoManager = new ApplicationInfoManager(
24 instanceConfig, new EurekaConfigBasedInstanceInfoProvider(instanceConfig).get());
25
26 // 5、eureka 客户端配置
27 EurekaClientConfig eurekaClientConfig = new DefaultEurekaClientConfig();
28 // 6、构造 EurekaClient，DiscoveryClient 封装了客户端相关的操作
29 eurekaClient = new DiscoveryClient(applicationInfoManager, eurekaClientConfig);
30 } else {
31 applicationInfoManager = eurekaClient.getApplicationInfoManager();
32 }
33
34 PeerAwareInstanceRegistry registry;
35 if (isAws(applicationInfoManager.getInfo())) {
36 registry = new AwsInstanceRegistry(
37 eurekaServerConfig,
38 eurekaClient.getEurekaClientConfig(),
39 serverCodecs,
40 eurekaClient
41 );
42 awsBinder = new AwsBinderDelegate(eurekaServerConfig, eurekaClient.getEurekaClientConfig(), registry, applicationInfoManager);
43 awsBinder.start();
44 } else {
45 // 7、构造感知eureka集群的注册表
46 registry = new PeerAwareInstanceRegistryImpl(
47 eurekaServerConfig,
48 eurekaClient.getEurekaClientConfig(),
49 serverCodecs,
50 eurekaClient
51 );
52 }
53
54 // 8、构造eureka-server集群信息
55 PeerEurekaNodes peerEurekaNodes = getPeerEurekaNodes(
56 registry,
57 eurekaServerConfig,
58 eurekaClient.getEurekaClientConfig(),
59 serverCodecs,
60 applicationInfoManager
61 );
62
63 // 9、基于前面构造的对象创建 EurekaServerContext
64 serverContext = new DefaultEurekaServerContext(
65 eurekaServerConfig,
66 serverCodecs,
67 registry,
68 peerEurekaNodes,
69 applicationInfoManager
70 );
71
72 // 将 serverContext 放到 EurekaServerContextHolder 上下文中，
73 // 这样其它地方都可以通过 EurekaServerContextHolder 拿到 EurekaServerContext
74 EurekaServerContextHolder.initialize(serverContext);
75
76 // 10、初始化eureka-server上下文
77 serverContext.initialize();
78 logger.info("Initialized server context");
79
80 // 11、从相邻的eureka-server同步注册表
81 int registryCount = registry.syncUp();
82 ///12、启动注册表，启动一些定时任务
83 registry.openForTraffic(applicationInfoManager, registryCount);
84
85 ///13、注册监控统计
86 EurekaMonitors.registerAllStats();
87 }
3、面向接口的配置读取
初始化中有三个配置接口，EurekaServerConfig、EurekaInstanceConfig、EurekaClientConfig，分别对应了注册中心、eureka实例、eureka客户端的配置获取。
从它们默认实现类的构造方法进去可以看到，EurekaServerConfig 是读取的 eureka-server.properties 配置文件，命名前缀是 eureka.server；EurekaInstanceConfig、EurekaClientConfig 是读取的 eureka-client.properties 配置文件，命名前缀分别是 eureka.instance、eureka.client。
这里可以看到，eureka 在代码中获取配置的方式是通过接口方法的形式来获取的，在其默认的实现类里通过硬编码的方式定义了配置的编码以及默认值。这种基于接口的配置读取方式是可以借鉴的，这种方式读取配置更易于维护，不用维护一堆常量，如果配置编码变了只需更改实现类即可。
例如下面的配置：
1 @Override
2 public int getExpectedClientRenewalIntervalSeconds() {
3 final int configured = configInstance.getIntProperty(
4 namespace + "expectedClientRenewalIntervalSeconds",
5 30).get();
6 return configured > 0 ? configured : 30;
7 }
8
9 @Override
10 public double getRenewalPercentThreshold() {
11 return configInstance.getDoubleProperty(
12 namespace + "renewalPercentThreshold", 0.85).get();
13 }
14
15 @Override
16 public boolean shouldEnableReplicatedRequestCompression() {
17 return configInstance.getBooleanProperty(
18 namespace + "enableReplicatedRequestCompression", false).get();
19 }
4、基于建造者模式构造服务实例
看 new EurekaConfigBasedInstanceInfoProvider(instanceConfig).get() 这段代码，在 get 方法中完成了服务实例信息的构造。它这里主要用到了建造者设计模式来构建 LeaseInfo 和 InstanceInfo，以 InstanceInfo 为例，它的内部有一个静态的 Builder 类，通过 newBuilder() 方法创建了 InstanceInfo 对象，然后可以调用 Builder 的属性设置方法来设置属性，在设置这些属性的时候，会做一些关联性的校验，在设置完成后，就调用 build() 方法返回对象，也可以在 build 方法中再做一些最终的校验。建造者模式就很适合用于构建这种复杂的对象。
1 public synchronized InstanceInfo get() {
2 if (instanceInfo == null) {
3 // 续约信息：主要有续约间隔时间（默认30秒）和续约过期时间（默认90秒）
4 LeaseInfo.Builder leaseInfoBuilder = LeaseInfo.Builder.newBuilder()
5 .setRenewalIntervalInSecs(config.getLeaseRenewalIntervalInSeconds())
6 .setDurationInSecs(config.getLeaseExpirationDurationInSeconds());
7
8 if (vipAddressResolver == null) {
9 vipAddressResolver = new Archaius1VipAddressResolver();
10 }
11
12 // 基于建造者模式来创建 InstanceInfo
13 InstanceInfo.Builder builder = InstanceInfo.Builder.newBuilder(vipAddressResolver);
14
15 // set the appropriate id for the InstanceInfo, falling back to datacenter Id if applicable, else hostname
16 String instanceId = config.getInstanceId();
17 if (instanceId == null || instanceId.isEmpty()) {
18 DataCenterInfo dataCenterInfo = config.getDataCenterInfo();
19 if (dataCenterInfo instanceof UniqueIdentifier) {
20 instanceId = ((UniqueIdentifier) dataCenterInfo).getId();
21 } else {
22 instanceId = config.getHostName(false);
23 }
24 }
25
26 String defaultAddress;
27 if (config instanceof RefreshableInstanceConfig) {
28 // Refresh AWS data center info, and return up to date address
29 defaultAddress = ((RefreshableInstanceConfig) config).resolveDefaultAddress(false);
30 } else {
31 defaultAddress = config.getHostName(false);
32 }
33
34 // fail safe
35 if (defaultAddress == null || defaultAddress.isEmpty()) {
36 defaultAddress = config.getIpAddress();
37 }
38
39 // 设置属性
40 builder.setNamespace(config.getNamespace())
41 .setInstanceId(instanceId)
42 .setAppName(config.getAppname())
43 .setAppGroupName(config.getAppGroupName())
44 .setDataCenterInfo(config.getDataCenterInfo())
45 .setIPAddr(config.getIpAddress())
46 .setHostName(defaultAddress)
47 .setPort(config.getNonSecurePort())
48 .enablePort(PortType.UNSECURE, config.isNonSecurePortEnabled())
49 .setSecurePort(config.getSecurePort())
50 .enablePort(PortType.SECURE, config.getSecurePortEnabled())
51 .setVIPAddress(config.getVirtualHostName())
52 .setSecureVIPAddress(config.getSecureVirtualHostName())
53 .setHomePageUrl(config.getHomePageUrlPath(), config.getHomePageUrl())
54 .setStatusPageUrl(config.getStatusPageUrlPath(), config.getStatusPageUrl())
55 .setASGName(config.getASGName())
56 .setHealthCheckUrls(config.getHealthCheckUrlPath(),
57 config.getHealthCheckUrl(), config.getSecureHealthCheckUrl());
58
59
60 // Start off with the STARTING state to avoid traffic
61 if (!config.isInstanceEnabledOnit()) {
62 InstanceStatus initialStatus = InstanceStatus.STARTING;
63 LOG.info("Setting initial instance status as: {}", initialStatus);
64 builder.setStatus(initialStatus);
65 } else {
66 LOG.info("Setting initial instance status as: {}. This may be too early for the instance to advertise "
67 + "itself as available. You would instead want to control this via a healthcheck handler.",
68 InstanceStatus.UP);
69 }
70
71 // Add any user-specific metadata information
72 for (Map.Entry<String, String> mapEntry : config.getMetadataMap().entrySet()) {
73 String key = mapEntry.getKey();
74 String value = mapEntry.getValue();
75 // only add the metadata if the value is present
76 if (value != null && !value.isEmpty()) {
77 builder.add(key, value);
78 }
79 }
80
81 // 调用 build 方法做属性校验并创建 InstanceInfo 实例
82 instanceInfo = builder.build();
83 instanceInfo.setLeaseInfo(leaseInfoBuilder.build());
84 }
85 return instanceInfo;
86 }
LeaseInfo 就是续约信息，可以看到主要的两个配置就是续约间隔时间和多久未续约认为实例过期，实例过期就会被剔除。然后就是基于 config 设置 InstanceInfo，就是实例信息，包含了实例ID、主机名称、端口、LeaseInfo 等等。
5、注册中心构造客户端 DiscoveryClient
在集群模式下，eureka server 也会作为客户端注册到其它注册中心，此时，它本身就是一个 eureka client。因此会去构建 EurekaClient，其默认实现类是 DiscoveryClient。DiscoveryClient 包含了 eureka 客户端的大部分核心功能，比如服务注册、续约、维持心跳、拉取注册表等。
一步步进入到DiscoveryClient最复杂的那个构造方法，我们先整体分析下做了哪些事情，抓大放小，很多组件的细节等后面分析具体功能的时候再来看。
- 将 EurekaClientConfig、EurekaInstanceConfig、EurekaTransportConfig、InstanceInfo、ApplicationInfoManager 等保存到本地变量中
- 如果要获取注册表，就创建一个注册表状态度量器
- 如果要注册到注册中心，就创建一个心跳状态度量器
- 如果不获取注册表且不注册到注册中心，就不会创建调度器、心跳线程池这些了，释放一些资源
- 如果要注册到注册中心且要抓取注册表，就初始化一些调度的资源：
- 创建了支持调度的线程池，有两个核心线程，从后面可以看出，主要就是处理心跳和缓存刷新的任务
- 创建了维持心跳的线程池，核心线程数为1，最大线程数配置默认为5
- 创建了刷新缓存的线程池，核心线程数为1，最大线程数配置默认为5
- 创建了eureka client 与 eureka server 进行网络通信的组件 EurekaTransport，并进行了一些初始化 。EurekaTransport 里的客户端主要就是封装了对 server 的 api 调用接口，便于调用
- 接着，如果要抓取注册表，就会抓取注册表了，fetchRegistry 里面可以看到是分为全量抓取和增量抓取的，第一次启动的时候就是全量抓取注册表
- 开始初始化调度任务：
- 如果要抓取注册表，就创建刷新缓存的任务，并开始调度，默认每隔30秒抓取一次注册表
- 如果要注册到注册中心，就创建发送心跳的任务，并开始调度，默认每隔30秒发送一次心跳
- 如果要注册到注册中心，还会创建实例副本传播器（内部也是一个定时调度任务）、实例状态变更的监听器
1 DiscoveryClient(ApplicationInfoManager applicationInfoManager, EurekaClientConfig config, AbstractDiscoveryClientOptionalArgs args,
2 Provider<BackupRegistry> backupRegistryProvider, EndpointRandomizer endpointRandomizer) {
3 if (args != null) {
4 this.healthCheckHandlerProvider = args.healthCheckHandlerProvider;
5 this.healthCheckCallbackProvider = args.healthCheckCallbackProvider;
6 this.eventListeners.addAll(args.getEventListeners());
7 this.preRegistrationHandler = args.preRegistrationHandler;
8 } else {
9 this.healthCheckCallbackProvider = null;
10 this.healthCheckHandlerProvider = null;
11 this.preRegistrationHandler = null;
12 }
13
14 // 将实例信息、配置信息保存到本地
15 this.applicationInfoManager = applicationInfoManager;
16 InstanceInfo myInfo = applicationInfoManager.getInfo();
17 clientConfig = config;
18 staticClientConfig = clientConfig;
19 transportConfig = config.getTransportConfig();
20 instanceInfo = myInfo;
21 if (myInfo != null) {
22 appPathIdentifier = instanceInfo.getAppName() + "/" + instanceInfo.getId();
23 } else {
24 logger.warn("Setting instanceInfo to a passed in null value");
25 }
26
27 this.backupRegistryProvider = backupRegistryProvider;
28 this.endpointRandomizer = endpointRandomizer;
29 this.urlRandomizer = new EndpointUtils.InstanceInfoBasedUrlRandomizer(instanceInfo);
30 localRegionApps.set(new Applications());
31
32
33 fetchRegistryGeneration = new AtomicLong(0);
34 remoteRegionsToFetch = new AtomicReference<String>(clientConfig.fetchRegistryForRemoteRegions());
35 // 从远程拉取注册表的地址数组，使用的原子类，在运行中可能会动态更新地址
36 remoteRegionsRef = new AtomicReference<>(remoteRegionsToFetch.get() == null ? null : remoteRegionsToFetch.get().split(","));
37
38 // 如果要获取注册表，就会注册状态监视器
39 if (config.shouldFetchRegistry()) {
40 this.registryStalenessMonitor = new ThresholdLevelsMetric(this, METRIC_REGISTRY_PREFIX + "lastUpdateSec_", new long[]{15L, 30L, 60L, 120L, 240L, 480L});
41 } else {
42 this.registryStalenessMonitor = ThresholdLevelsMetric.NO_OP_METRIC;
43 }
44
45 // 如果要注册到 eureka-server，就会创建心跳状态监视器
46 if (config.shouldRegisterWithEureka()) {
47 this.heartbeatStalenessMonitor = new ThresholdLevelsMetric(this, METRIC_REGISTRATION_PREFIX + "lastHeartbeatSec_", new long[]{15L, 30L, 60L, 120L, 240L, 480L});
48 } else {
49 this.heartbeatStalenessMonitor = ThresholdLevelsMetric.NO_OP_METRIC;
50 }
51
52 logger.info("Initializing Eureka in region {}", clientConfig.getRegion());
53
54 // 如果不注册到注册中心，且不拉取注册表，就不创建调度器、线程池等资源了
55 if (!config.shouldRegisterWithEureka() && !config.shouldFetchRegistry()) {
56 logger.info("Client configured to neither register nor query for data.");
57 scheduler = null;
58 heartbeatExecutor = null;
59 cacheRefreshExecutor = null;
60 eurekaTransport = null;
61 instanceRegionChecker = new InstanceRegionChecker(new PropertyBasedAzToRegionMapper(config), clientConfig.getRegion());
62
63 // This is a bit of hack to allow for existing code using DiscoveryManager.getInstance()
64 // to work with DI'd DiscoveryClient
65 DiscoveryManager.getInstance().setDiscoveryClient(this);
66 DiscoveryManager.getInstance().setEurekaClientConfig(config);
67
68 initTimestampMs = System.currentTimeMillis();
69 initRegistrySize = this.getApplications().size();
70 registrySize = initRegistrySize;
71 logger.info("Discovery Client initialized at timestamp {} with initial instances count: {}",
72 initTimestampMs, initRegistrySize);
73
74 return; // no need to setup up an network tasks and we are done
75 }
76
77 try {
78 // 创建定时调度器，默认有2个核心线程，主要处理心跳任务和缓存刷新任务
79 scheduler = Executors.newScheduledThreadPool(2,
80 new ThreadFactoryBuilder()
81 .setNameFormat("DiscoveryClient-%d")
82 .setDaemon(true)
83 .build());
84
85 // 维持心跳的线程池，一个核心线程，最大线程数默认5。
86 // 注意其使用的队列是 SynchronousQueue 队列，这个队列只能放一个任务，一个线程将任务取走后，才能放入下一个任务，否则只能阻塞。
87 heartbeatExecutor = new ThreadPoolExecutor(
88 1, clientConfig.getHeartbeatExecutorThreadPoolSize(), 0, TimeUnit.SECONDS,
89 new SynchronousQueue<Runnable>(),
90 new ThreadFactoryBuilder()
91 .setNameFormat("DiscoveryClient-HeartbeatExecutor-%d")
92 .setDaemon(true)
93 .build()
94 ); // use direct handoff
95
96 // 刷新缓存的线程池，一个核心线程，最大线程数据默认为5
97 cacheRefreshExecutor = new ThreadPoolExecutor(
98 1, clientConfig.getCacheRefreshExecutorThreadPoolSize(), 0, TimeUnit.SECONDS,
99 new SynchronousQueue<Runnable>(),
100 new ThreadFactoryBuilder()
101 .setNameFormat("DiscoveryClient-CacheRefreshExecutor-%d")
102 .setDaemon(true)
103 .build()
104 ); // use direct handoff
105
106 // eureka http 调用客户端，支持 eureka client 与 eureka server 之间的通信
107 eurekaTransport = new EurekaTransport();
108 // 初始化 eurekaTransport
109 scheduleServerEndpointTask(eurekaTransport, args);
110
111 AzToRegionMapper azToRegionMapper;
112 if (clientConfig.shouldUseDnsForFetchingServiceUrls()) {
113 azToRegionMapper = new DNSBasedAzToRegionMapper(clientConfig);
114 } else {
115 azToRegionMapper = new PropertyBasedAzToRegionMapper(clientConfig);
116 }
117 if (null != remoteRegionsToFetch.get()) {
118 azToRegionMapper.setRegionsToFetch(remoteRegionsToFetch.get().split(","));
119 }
120 instanceRegionChecker = new InstanceRegionChecker(azToRegionMapper, clientConfig.getRegion());
121 } catch (Throwable e) {
122 throw new RuntimeException("Failed to initialize DiscoveryClient!", e);
123 }
124
125 if (clientConfig.shouldFetchRegistry()) {
126 try {
127 // 拉取注册表：全量抓取和增量抓取
128 boolean primaryFetchRegistryResult = fetchRegistry(false);
129 if (!primaryFetchRegistryResult) {
130 logger.info("Initial registry fetch from primary servers failed");
131 }
132 boolean backupFetchRegistryResult = true;
133 if (!primaryFetchRegistryResult && !fetchRegistryFromBackup()) {
134 backupFetchRegistryResult = false;
135 logger.info("Initial registry fetch from backup servers failed");
136 }
137 if (!primaryFetchRegistryResult && !backupFetchRegistryResult && clientConfig.shouldEnforceFetchRegistryAtInit()) {
138 throw new IllegalStateException("Fetch registry error at startup. Initial fetch failed.");
139 }
140 } catch (Throwable th) {
141 logger.error("Fetch registry error at startup: {}", th.getMessage());
142 throw new IllegalStateException(th);
143 }
144 }
145
146 // call and execute the pre registration handler before all background tasks (inc registration) is started
147 if (this.preRegistrationHandler != null) {
148 this.preRegistrationHandler.beforeRegistration();
149 }
150
151 if (clientConfig.shouldRegisterWithEureka() && clientConfig.shouldEnforceRegistrationAtInit()) {
152 try {
153 if (!register() ) {
154 throw new IllegalStateException("Registration error at startup. Invalid server response.");
155 }
156 } catch (Throwable th) {
157 logger.error("Registration error at startup: {}", th.getMessage());
158 throw new IllegalStateException(th);
159 }
160 }
161
162 // 初始化一些调度任务：刷新缓存的调度任务、发送心跳的调度任务、实例副本传播器
163 initScheduledTasks();
164
165 try {
166 Monitors.registerObject(this);
167 } catch (Throwable e) {
168 logger.warn("Cannot register timers", e);
169 }
170
171 // This is a bit of hack to allow for existing code using DiscoveryManager.getInstance()
172 // to work with DI'd DiscoveryClient
173 DiscoveryManager.getInstance().setDiscoveryClient(this);
174 DiscoveryManager.getInstance().setEurekaClientConfig(config);
175
176 // 初始化的时间
177 initTimestampMs = System.currentTimeMillis();
178 initRegistrySize = this.getApplications().size();
179 registrySize = initRegistrySize;
180 logger.info("Discovery Client initialized at timestamp {} with initial instances count: {}",
181 initTimestampMs, initRegistrySize);
182 }
183
184 ////////////////////////////////////////////////////////////////////
185
186 private void initScheduledTasks() {
187 if (clientConfig.shouldFetchRegistry()) {
188 // 抓取注册表的间隔时间，默认30秒
189 int registryFetchIntervalSeconds = clientConfig.getRegistryFetchIntervalSeconds();
190 // 刷新缓存调度器延迟时间扩大倍数，在任务超时的时候，将扩大延迟时间
191 // 这在出现网络抖动、eureka-sever 不可用时，可以避免频繁发起无效的调度
192 int expBackOffBound = clientConfig.getCacheRefreshExecutorExponentialBackOffBound();
193 // 注册表刷新的定时任务
194 cacheRefreshTask = new TimedSupervisorTask(
195 "cacheRefresh",
196 scheduler,
197 cacheRefreshExecutor,
198 registryFetchIntervalSeconds,
199 TimeUnit.SECONDS,
200 expBackOffBound,
201 new CacheRefreshThread() // 刷新注册表的任务
202 );
203 // 30秒后开始调度刷新注册表的任务
204 scheduler.schedule(
205 cacheRefreshTask,
206 registryFetchIntervalSeconds, TimeUnit.SECONDS);
207 }
208
209 if (clientConfig.shouldRegisterWithEureka()) {
210 // 续约间隔时间，默认30秒
211 int renewalIntervalInSecs = instanceInfo.getLeaseInfo().getRenewalIntervalInSecs();
212 // 心跳调度器的延迟时间扩大倍数，默认10
213 int expBackOffBound = clientConfig.getHeartbeatExecutorExponentialBackOffBound();
214 logger.info("Starting heartbeat executor: " + "renew interval is: {}", renewalIntervalInSecs);
215
216 // 心跳的定时任务
217 heartbeatTask = new TimedSupervisorTask(
218 "heartbeat",
219 scheduler,
220 heartbeatExecutor,
221 renewalIntervalInSecs,
222 TimeUnit.SECONDS,
223 expBackOffBound,
224 new HeartbeatThread()
225 );
226 // 30秒后开始调度心跳的任务
227 scheduler.schedule(
228 heartbeatTask,
229 renewalIntervalInSecs, TimeUnit.SECONDS);
230
231 // 实例副本传播器，用于定时更新自己状态
232 instanceInfoReplicator = new InstanceInfoReplicator(
233 this,
234 instanceInfo,
235 clientConfig.getInstanceInfoReplicationIntervalSeconds(),
236 2); // burstSize
237
238 // 实例状态变更的监听器
239 statusChangeListener = new ApplicationInfoManager.StatusChangeListener() {
240 @Override
241 public String getId() {
242 return "statusChangeListener";
243 }
244
245 @Override
246 public void notify(StatusChangeEvent statusChangeEvent) {
247 if (statusChangeEvent.getStatus() == InstanceStatus.DOWN) {
248 logger.error("Saw local status change event {}", statusChangeEvent);
249 } else {
250 logger.info("Saw local status change event {}", statusChangeEvent);
251 }
252 instanceInfoReplicator.onDemandUpdate();
253 }
254 };
255
256 // 向 ApplicationInfoManager 注册监听器
257 if (clientConfig.shouldOnDemandUpdateStatusChange()) {
258 applicationInfoManager.registerStatusChangeListener(statusChangeListener);
259 }
260
261 // 启动副本传播器，默认延迟时间40秒
262 instanceInfoReplicator.start(clientConfig.getInitialInstanceInfoReplicationIntervalSeconds());
263 } else {
264 logger.info("Not registering with Eureka server per configuration");
265 }
266 }
6、定时任务监管器的设计
可以看到，eureka client 为了定时发送心跳以及定时抓取注册表，使用了定时任务和调度器，我觉得它这里的定时调度的设计思想是可以参考和借鉴的。
以心跳任务的这段代码为例：
1 if (clientConfig.shouldFetchRegistry()) {
2 // 抓取注册表的间隔时间，默认30秒
3 int registryFetchIntervalSeconds = clientConfig.getRegistryFetchIntervalSeconds();
4 // 刷新缓存调度器延迟时间扩大倍数，在任务超时的时候，将扩大延迟时间
5 // 这在出现网络抖动、eureka-sever 不可用时，可以避免频繁发起无效的调度
6 int expBackOffBound = clientConfig.getCacheRefreshExecutorExponentialBackOffBound();
7 // 注册表刷新的定时任务
8 cacheRefreshTask = new TimedSupervisorTask(
9 "cacheRefresh",
10 scheduler,
11 cacheRefreshExecutor,
12 registryFetchIntervalSeconds,
13 TimeUnit.SECONDS,
14 expBackOffBound,
15 new CacheRefreshThread() // 刷新注册表的任务
16 );
17 // 30秒后开始调度刷新注册表的任务
18 scheduler.schedule(
19 cacheRefreshTask,
20 registryFetchIntervalSeconds, TimeUnit.SECONDS);
21 }
上面这段代码其实并不复杂，主要就是创建了一个定时任务，然后使用调度器在一定的延迟之后开始调度。但它这里并不是直接使用调度器调度任务（CacheRefreshThread），也不是以一个固定的频率调度（每隔30秒）。它定义了一个任务的监管器 TimedSupervisorTask，在创建这个监管器的时候，传入了调度器、要执行的任务、以及间隔时间等参数，然后调度器调度 TimedSupervisorTask。
看 TimedSupervisorTask 的构造方法，主要有以下几个点：
- 任务的超时时间等于间隔时间，也就是默认30秒的超时时间，然后延迟时间默认等于超时时间 如果 eureka server down 了，或者网络问题，就有可能出现超时
- 设置了最大的延迟时间，默认在超时时间的基础上扩大10倍，即300秒
- 最后创建了一些计数器，分别统计成功、超时、拒绝、异常的次数，可以看到，它这里对任务的调度是有做统计的
1 public TimedSupervisorTask(String name, ScheduledExecutorService scheduler, ThreadPoolExecutor executor,
2 int timeout, TimeUnit timeUnit, int expBackOffBound, Runnable task) {
3 this.name = name;
4 this.scheduler = scheduler;
5 this.executor = executor;
6 // 任务超时时间就等于任务调度的间隔时间
7 this.timeoutMillis = timeUnit.toMillis(timeout);
8 this.task = task;
9 // 延迟时间默认为超时时间
10 this.delay = new AtomicLong(timeoutMillis);
11 // 最大延迟时间，默认在超时时间的基础上扩大10倍
12 this.maxDelay = timeoutMillis * expBackOffBound;
13
14 // 初始化计数器并注册
15 successCounter = Monitors.newCounter("success");
16 timeoutCounter = Monitors.newCounter("timeouts");
17 rejectedCounter = Monitors.newCounter("rejectedExecutions");
18 throwableCounter = Monitors.newCounter("throwables");
19 threadPoolLevelGauge = new LongGauge(MonitorConfig.builder("threadPoolUsed").build());
20 Monitors.registerObject(name, this);
21 }
再看 TimedSupervisorTask 的 run 方法：
- 1）首先将任务异步提交到线程池去执行，它这里并不是直接运行任务，而是异步提交到线程池中，这样可以实现超时等待，不影响主任务
- 2）任务如果超时，比如出现网络延迟、eureka server 不可用等情况，超时了，它这个时候就会认为如果还是30秒后调度，可能 eureka server 还是不可用的状态，那么就增大延迟时间，那么第一次超时就会在300秒后再调度。如果300秒内 eureka server 可用了，然后有新的服务实例注册上去了，那这个客户端就不能及时感知到了，因此我觉得可以将 getCacheRefreshExecutorExponentialBackOffBound 对应的参数适当设置小一点（默认10倍）。
- 3）如果任务没有超时，在调度成功后，就会重置延迟时间为默认的超时时间。最后在 finally 中进行下一次的调度。
1 public void run() {
2 Future<?> future = null;
3 try {
4 // 提交任务到线程池
5 future = executor.submit(task);
6 threadPoolLevelGauge.set((long) executor.getActiveCount());
7 // 阻塞直到任务完成或超时
8 future.get(timeoutMillis, TimeUnit.MILLISECONDS);
9 // 任务完成后，重置延迟时间为超时时间，即30秒
10 delay.set(timeoutMillis);
11 threadPoolLevelGauge.set((long) executor.getActiveCount());
12 // 成功次数+1
13 successCounter.increment();
14 } catch (TimeoutException e) {
15 logger.warn("task supervisor timed out", e);
16 // 超时次数+1
17 timeoutCounter.increment();
18
19 // 如果任务超时了，就会增大延迟时间，当前延迟时间*2，然后取一个最大值
20 long currentDelay = delay.get();
21 long newDelay = Math.min(maxDelay, currentDelay * 2);
22 // 设置为最大的一个延迟时间
23 delay.compareAndSet(currentDelay, newDelay);
24
25 } catch (RejectedExecutionException e) {
26 if (executor.isShutdown() || scheduler.isShutdown()) {
27 logger.warn("task supervisor shutting down, reject the task", e);
28 } else {
29 logger.warn("task supervisor rejected the task", e);
30 }
31
32 rejectedCounter.increment();
33 } catch (Throwable e) {
34 if (executor.isShutdown() || scheduler.isShutdown()) {
35 logger.warn("task supervisor shutting down, can't accept the task");
36 } else {
37 logger.warn("task supervisor threw an exception", e);
38 }
39
40 throwableCounter.increment();
41 } finally {
42 if (future != null) {
43 future.cancel(true);
44 }
45
46 if (!scheduler.isShutdown()) {
47 // 延迟 delay 时间后，继续调度任务
48 scheduler.schedule(this, delay.get(), TimeUnit.MILLISECONDS);
49 }
50 }
51 }
总结一下这块设计：
- 1）首先在远程调用的时候要考虑到网络不可用、server 端 down 了等情况导致调用超时，可以使用线程池异步提交任务，实现等待超时机制。
- 2）超时之后，可以假想服务恢复可用状态可能需要一定的时间，如果还是按原来的时间间隔调度，可能还是会超时，因此增大延迟时间。如果调用成功，说明已经恢复了，则重置延迟时间。
- 3）定时任务的调度以一定的延迟时间来循环调度（schedule），延迟时间可以根据实际情况变化，而不是一开始就按一个固定的频率来调度（scheduleAtFixedRate）。
- 4）定时任务、线程池里的任务，最好做好任务执行状态的统计，便于观察任务的调度情况。
7、构造注册表
接着构造 PeerAwareInstanceRegistry，从命名来看，这是一个可以感知 eureka 集群的注册表，就是在集群模式下，eureka server 从其它 server 节点拉取注册表。它的默认实现类是 PeerAwareInstanceRegistryImpl，继承自 AbstractInstanceRegistry，就是实例注册表。
① 构造 PeerAwareInstanceRegistry
进入 PeerAwareInstanceRegistryImpl 的构造方法：
- 首先是将前面构造的 EurekaServerConfig、EurekaClientConfig、EurekaClient 等传入构造方法来构造 PeerAwareInstanceRegistry
- 调用了 super 的构造方法，主要初始化了如下几个东西：
- 保存最近下线实例的循环队列
- 保存最近注册实例的循环队列
- 最近一分钟续约次数的计数器
- 定时任务剔除 recentlyChangedQueue 中的实例
- 然后创建了一个最近一分钟集群同步次数的计数器 numberOfReplicationsLastMin。MeasuredRate 我们到后面再来分析它的设计。
1 public PeerAwareInstanceRegistryImpl(
2 EurekaServerConfig serverConfig,
3 EurekaClientConfig clientConfig,
4 ServerCodecs serverCodecs,
5 EurekaClient eurekaClient
6 ) {
7 super(serverConfig, clientConfig, serverCodecs);
8 this.eurekaClient = eurekaClient;
9 // 最近一分钟集群同步的次数计数器
10 this.numberOfReplicationsLastMin = new MeasuredRate(1000 * 60 * 1);
11 // We first check if the instance is STARTING or DOWN, then we check explicit overrides,
12 // then we check the status of a potentially existing lease.
13 this.instanceStatusOverrideRule = new FirstMatchWinsCompositeRule(new DownOrStartingRule(),
14 new OverrideExistsRule(overriddenInstanceStatusMap), new LeaseExistsRule());
15 }
16
17
18 ///////////////////////////////////////////////
19
20
21 protected AbstractInstanceRegistry(EurekaServerConfig serverConfig, EurekaClientConfig clientConfig, ServerCodecs serverCodecs) {
22 this.serverConfig = serverConfig;
23 this.clientConfig = clientConfig;
24 this.serverCodecs = serverCodecs;
25 // 最近下线的循环队列
26 this.recentCanceledQueue = new CircularQueue<Pair<Long, String>>(1000);
27 // 最近注册的循环队列
28 this.recentRegisteredQueue = new CircularQueue<Pair<Long, String>>(1000);
29
30 // 最近一分钟续约的计数器
31 this.renewsLastMin = new MeasuredRate(1000 * 60 * 1);
32 // 一个定时调度任务，定时剔除最近改变队列中过期的实例
33 this.deltaRetentionTimer.schedule(getDeltaRetentionTask(),
34 serverConfig.getDeltaRetentionTimerIntervalInMs(),
35 serverConfig.getDeltaRetentionTimerIntervalInMs());
36 }
这块的具体细节等后面分析具体功能的时候再来看，我们先知道有这些队列、计数器就行了。
② 循环队列 CircularQueue 的设计
从构造方法可以看到，它使用了循环队列来保存最近下线和最近注册的实例信息，容量固定为1000，这样就把最近的实例数量控制在1000以内。
CircularQueue 是它自定义的一个循环队列，继承自 AbstractQueue。其内部其实就是代理了 ArrayBlockingQueue，然后重写了入队的 offer 方法，当队列满了，就取出头部的一个元素，然后再放到队列尾部。
1 static class CircularQueue<E> extends AbstractQueue<E> {
2 private final ArrayBlockingQueue<E> delegate;
3 private final int capacity;
4
5 public CircularQueue(int capacity) {
6 this.capacity = capacity;
7 this.delegate = new ArrayBlockingQueue<>(capacity);
8 }
9
10 @Override
11 public Iterator<E> iterator() {
12 return delegate.iterator();
13 }
14
15 @Override
16 public int size() {
17 return delegate.size();
18 }
19
20 @Override
21 public boolean offer(E e) {
22 // 如果队列满了，就取出头部的一个元素，然后再放到尾部
23 while (!delegate.offer(e)) {
24 delegate.poll();
25 }
26 return true;
27 }
28
29 @Override
30 public E poll() {
31 return delegate.poll();
32 }
33
34 @Override
35 public E peek() {
36 return delegate.peek();
37 }
38 }
8、创建 Eureka Server 上下文并初始化
接下来先是创建了 PeerEurekaNodes，应该就是代表 eureka 集群的。然后基于前面创建的一些东西创建 eureka server 上下文 EurekaServerContext，从 DefaultEurekaServerContext 构造方法进去可以看到，只是将前面构造的东西封装起来，便于全局使用。然后将 serverContext 放到 EurekaServerContextHolder 中，这样其它地方就可以通过这个 holder 获取 serverContext 了。
接着就是初始化eureka server上下文：
- 启动 eureka 集群：
- 主要是启动一个定时任务（间隔时间默认10分钟）更新 eureka 集群节点的信息，根据配置的 eureka server 地址更新 PeerEurekaNode，这样当有 eureka server 下线或上线后，就可以及时感知到其它 server 节点。PeerEurekaNode 主要就是用于集群节点间的数据同步，这块后面分析集群的时候再具体分析。
- 注册表初始化：
- 首先启动了前面创建的计数器：numberOfReplicationsLastMin
- 初始化响应缓存，eureka server 构造了一个多级缓存来响应客户端抓取注册表的请求，这个多级缓存的设计就是响应频繁抓取注册表请求的核心所在，等后面分析客户端抓取注册表的时候再具体分析
- 定时调度任务更新续约阈值，主要就是更新 numberOfRenewsPerMinThreshold 这个值，即每分钟续约次数，等分析续约的时候再来分析
- 初始化 RemoteRegionRegistry，猜测是跟 eureka 多个区域（region）部署有关的
1 public void initialize() {
2 logger.info("Initializing ...");
3 // 启动eureka集群
4 peerEurekaNodes.start();
5 try {
6 // 注册表初始化
7 registry.init(peerEurekaNodes);
8 } catch (Exception e) {
9 throw new RuntimeException(e);
10 }
11 logger.info("Initialized");
12 }
PeerEurekaNodes 的 start 方法：
1 public void start() {
2 // 单个线程的线程池
3 taskExecutor = Executors.newSingleThreadScheduledExecutor(
4 new ThreadFactory() {
5 @Override
6 public Thread newThread(Runnable r) {
7 Thread thread = new Thread(r, "Eureka-PeerNodesUpdater");
8 thread.setDaemon(true);
9 return thread;
10 }
11 }
12 );
13 try {
14 // 根据集群地址更新 PeerEurekaNode，PeerEurekaNode 就包含了调度其它注册中心的客户端
15 updatePeerEurekaNodes(resolvePeerUrls());
16 Runnable peersUpdateTask = new Runnable() {
17 @Override
18 public void run() {
19 try {
20 updatePeerEurekaNodes(resolvePeerUrls());
21 } catch (Throwable e) {
22 logger.error("Cannot update the replica Nodes", e);
23 }
24
25 }
26 };
27 // 定时跟新集群信息 PeerEurekaNode，如果有eureka-server不可用了，就可以及时下线，或者新上线了eureka-server，可以及时感知到
28 taskExecutor.scheduleWithFixedDelay(
29 peersUpdateTask,
30 serverConfig.getPeerEurekaNodesUpdateIntervalMs(),
31 serverConfig.getPeerEurekaNodesUpdateIntervalMs(),
32 TimeUnit.MILLISECONDS
33 );
34 } catch (Exception e) {
35 throw new IllegalStateException(e);
36 }
37 for (PeerEurekaNode node : peerEurekaNodes) {
38 logger.info("Replica node URL: {}", node.getServiceUrl());
39 }
40 }
PeerAwareInstanceRegistryImpl 的 init 方法：
1 public void init(PeerEurekaNodes peerEurekaNodes) throws Exception {
2 // 启动计数器
3 this.numberOfReplicationsLastMin.start();
4 this.peerEurekaNodes = peerEurekaNodes;
5 // 初始化响应缓存，eureka server 构造了一个多级缓存来响应客户端抓取注册表的请求
6 initializedResponseCache();
7 // 定时调度任务更新续约阀值，主要就是更新 numberOfRenewsPerMinThreshold 这个值，即每分钟续约次数
8 scheduleRenewalThresholdUpdateTask();
9 // 初始化 RemoteRegionRegistry
10 initRemoteRegionRegistry();
11
12 try {
13 Monitors.registerObject(this);
14 } catch (Throwable e) {
15 logger.warn("Cannot register the JMX monitor for the InstanceRegistry :", e);
16 }
17 }
9、完成 Eureka Server 初始化
接下来看最后几步：
- 首先调用 registry.syncUp() 将 EurekaClient 本地的实例同步到注册表，在集群模式下，eureka server 也是一个客户端，因此会获取到其它注册中心的注册表同步到当前 server 的注册表中。它默认会重试5次，每次间隔30秒。在单机模式下，应该将重试次数设置为 0。
- 然后调用 registry.openForTraffic 做最后的一些初始化：
- 更新每分钟续约阈值
- 设置实例状态
- 启动统计最近一分钟续约次数的计数器
- 启动定时任务剔除下线的实例，定时任务默认每隔60秒调度一次
- 最后一步就是注册 eureka 自身的一些监控统计
syncUp 方法：
1 public int syncUp() {
2 // Copy entire entry from neighboring DS node
3 int count = 0;
4
5 // 注册表同步重试次数，默认5次
6 for (int i = 0; ((i < serverConfig.getRegistrySyncRetries()) && (count == 0)); i++) {
7 if (i > 0) {
8 try {
9 // 同步重试时间，默认30秒
10 Thread.sleep(serverConfig.getRegistrySyncRetryWaitMs());
11 } catch (InterruptedException e) {
12 logger.warn("Interrupted during registry transfer..");
13 break;
14 }
15 }
16 Applications apps = eurekaClient.getApplications();
17 for (Application app : apps.getRegisteredApplications()) {
18 for (InstanceInfo instance : app.getInstances()) {
19 try {
20 if (isRegisterable(instance)) {
21 // 注册实例
22 register(instance, instance.getLeaseInfo().getDurationInSecs(), true);
23 count++;
24 }
25 } catch (Throwable t) {
26 logger.error("During DS init copy", t);
27 }
28 }
29 }
30 }
31 return count;
32 }
openForTraffic 方法：
1 @Override
2 public void openForTraffic(ApplicationInfoManager applicationInfoManager, int count) {
3 // 期望的客户端每分钟的续约次数
4 this.expectedNumberOfClientsSendingRenews = count;
5 // 更新每分钟续约阀值
6 updateRenewsPerMinThreshold();
7 logger.info("Got {} instances from neighboring DS node", count);
8 logger.info("Renew threshold is: {}", numberOfRenewsPerMinThreshold);
9 this.startupTime = System.currentTimeMillis();
10 if (count > 0) {
11 this.peerInstancesTransferEmptyOnStartup = false;
12 }
13 DataCenterInfo.Name selfName = applicationInfoManager.getInfo().getDataCenterInfo().getName();
14 boolean isAws = Name.Amazon == selfName;
15 if (isAws && serverConfig.shouldPrimeAwsReplicaConnections()) {
16 logger.info("Priming AWS connections for all replicas..");
17 primeAwsReplicas(applicationInfoManager);
18 }
19 logger.info("Changing status to UP");
20 // 设置实例状态为已启动
21 applicationInfoManager.setInstanceStatus(InstanceStatus.UP);
22 super.postInit();
23 }
24
25 ///////////////////////////////////////
26
27 protected void postInit() {
28 // 启动 统计最近一分钟续约次数的计数器
29 renewsLastMin.start();
30 if (evictionTaskRef.get() != null) {
31 evictionTaskRef.get().cancel();
32 }
33 // 定时剔除任务
34 evictionTaskRef.set(new EvictionTask());
35 evictionTimer.schedule(evictionTaskRef.get(),
36 serverConfig.getEvictionIntervalTimerInMs(),
37 serverConfig.getEvictionIntervalTimerInMs());
38 }
10、Eureka Server 启动流程图
下面通过一张图来展示下 eureka server 的启动初始化流程。
四、Eureka Client 启动初始化
eureka client 的启动初始化我们看 eureka-examples 模块下的 ExampleEurekaClient 这个类，它的 main 方法中就模拟了作为一个 eureka client 启动初始化，并向注册中心发送请求。
eureka server 的初始化中其实已经包含了客户端的初始化，可以看出，客户端的初始化主要有如下的一些东西：
- 读取 eureka-client.properties 配置文件，创建 EurekaInstanceConfig
- 基于 InstanceConfig 创建实例信息 InstanceInfo
- 基于 InstanceConfig 和 InstanceInfo 创建应用实例管理器 ApplicationInfoManager
- 读取 eureka-client.properties 配置文件，创建 EurekaClientConfig
- 基于应用实例管理器和 clientConfig 创建 EurekaClient（DiscoveryClient），初始化流程跟 eureka server 初始化流程中 DiscoveryClient 的创建是一样的
1 public static void main(String[] args) {
2 ExampleEurekaClient sampleClient = new ExampleEurekaClient();
3
4 // 基于实例配置和实例信息创建应用实例管理器
5 ApplicationInfoManager applicationInfoManager = initializeApplicationInfoManager(new MyDataCenterInstanceConfig());
6 // 基于应用实例管理器和客户端配置创建 EurekaClient（DiscoveryClient）
7 EurekaClient client = initializeEurekaClient(applicationInfoManager, new DefaultEurekaClientConfig());
8
9 // use the client
10 sampleClient.sendRequestToServiceUsingEureka(client);
11
12 // shutdown the client
13 eurekaClient.shutdown();
14 }
最新文章
- 【干货分享】流程DEMO-事务呈批表
- python读取文本文件
- JavaScript函数的概念
- 获得APP当前显示的viewController
- Comparable接口
- PHP学习笔记：数据库学习心得
- php setcookie(name, value, expires, path, domain, secure) 参数详解
- 【大坑】DataGridView多线程更新修改Cell单元格卡死
- java 读入换行
- C#部分方法定义
- 整个Html内容以邮件的方式发送出去（取出标签包含的用户输入信息）
- 操作百度API
- 【转】Android内存机制分析1——了解Android堆和栈
- Django 学习笔记（七）数据库基本操作（增查改删）
- C GOTO使用示例
- C语言实现循环队列
- JS 页面表格的操作
- 共轭函数Fenchel不等式
- 设计模式 策略模式2 c++11
- css - bootstrap3下拉菜单点击之后怎么改变背景颜色?