10
SpringCloud 源码系列（3）—— 注册中心 Eureka（下）
source link: http://www.cnblogs.com/chiangchou/p/eureka-3.html
Go to the source link to view the article. You can view the picture content, updated content and better typesetting reading experience. If the link is broken, please click the button below to view the snapshot at that time.
SpringCloud 源码系列（1）—— 注册中心 Eureka（上）
SpringCloud 源码系列（2）—— 注册中心 Eureka（中）
SpringCloud 源码系列（3）—— 注册中心 Eureka（下）
十一、Eureka Server 集群
在实际的生产环境中，可能有几十个或者几百个的微服务实例，Eureka Server 承担了非常高的负载，而且为了保证注册中心高可用，一般都要部署成集群的，下面就来看看 eureka server 的集群。
1、搭建 Eureka Server 集群
首先来搭建一个三个节点的 eureka-server 集群，看看效果。
① 集群配置
首先在本地 hosts 文件中配置如下映射：
1 127.0.0.1 peer1 2 127.0.0.1 peer2 3 127.0.0.1 peer3
更改注册中心的 application.yml 配置文件，增加三个 profile，分别对应三个 eureka-server 的客户端配置。
eureka-server 在集群中作为客户端就需要抓取注册表，并配置 eureka-server 的地址。
1 spring: 2 application: 3 name: sunny-register 4 5 --- 6 spring: 7 profiles: peer1 8 server: 9 port: 8001 10 11 eureka: 12 instance: 13 hostname: peer1 14 client: 15 # 是否向注册中心注册自己 16 register-with-eureka: false 17 # 是否抓取注册表 18 fetch-registry: true 19 service-url: 20 defaultZone: http://peer1:8001/eureka,http://peer2:8002/eureka,http://peer3:8003/eureka 21 22 23 --- 24 spring: 25 profiles: peer2 26 server: 27 port: 8002 28 29 eureka: 30 instance: 31 hostname: peer2 32 client: 33 # 是否向注册中心注册自己 34 register-with-eureka: false 35 # 是否抓取注册表 36 fetch-registry: true 37 service-url: 38 defaultZone: http://peer1:8001/eureka,http://peer2:8002/eureka,http://peer3:8003/eureka 39 40 --- 41 spring: 42 profiles: peer3 43 server: 44 port: 8003 45 46 eureka: 47 instance: 48 hostname: peer3 49 client: 50 # 是否向注册中心注册自己 51 register-with-eureka: false 52 # 是否抓取注册表 53 fetch-registry: true 54 service-url: 55 defaultZone: http://peer1:8001/eureka,http://peer2:8002/eureka,http://peer3:8003/eureka
② 启动集群
分别启动三个注册中心，环境变量 spring.profiles.active 激活对应的集群配置。
启动之后访问 http://peer1:8001/ 进入 peer1 这个注册中心，就可以看到另外两个分片 peer2、peer3，说明集群中有3个节点了。
③ 启动客户端
首先客户端配置增加集群地址：
1 eureka: 2 client: 3 serviceUrl: 4 defaultZone: http://peer1:8001/eureka,http://peer2:8002/eureka,http://peer3:8003/eureka
启动几个客户端实例，过一会 之后，会发现三个 eureka-server 上都注册上去了：
到此 eureka-server 集群就搭建起来了，可以看到注册中心的实例会互相同步，每隔注册注册都可以接收注册、续约、下线等请求，它们是对等的。
2、Eureka Server 集群架构
一般来说，分布式系统的数据在多个副本之间的复制方式，可分为主从复制和对等复制。
① 主从复制
主从复制就是 Master-Slave 模式，即一个主副本，其它副本都为从副本。所有对数据的写操作都提交到主副本，然后再由主副本同步到从副本。
对于主从复制模式来说，写操作的压力都在主副本上，它是整个系统的瓶颈，而从副本则可以帮助主副本分担读请求。
② 对等复制
对等复制就是 Peer to Peer 的模式，副本之间不分主从，任何副本都可以接收写操作，每个副本之间相互进行数据更新同步。
Peer to Peer 模式每个副本之间都可以接收写请求，不存在写操作压力瓶颈。但是由于每个副本都可以进行写操作，各个副本之间的数据同步及冲突处理是一个棘手的问题。
③ Eureka Server 集群架构
Eureka Server 采用的就是 Peer to Peer 的复制模式，比如一个客户端实例随机向其中一个server注册，然后它就会同步到其它节点中。
3、Eureka Server 启动时抓取注册表
前面已经分析过了，在 eureka server 启动初始化的时候，即 EurekaBootStrap 初始化类，先初始化了 DiscoveryClient，DiscoveryClient 会向注册中心全量抓取注册表到本地。
初始化的最后调用了 registry.syncUp() 来同步注册表，就是将 DiscoveryClient 缓存的实例注册到 eureka-server 的注册表里去。
需要注意的是 eureka 配置的注册表同步重试次数默认为5，springcloud 中默认为 0，因此需要添加如下配置来开启注册表同步。
1 eureka: 2 server: 3 registry-sync-retries: 5
将 DiscoveryClient 本地的实例注册到注册表中：
4、集群节点同步
① 注册、续约、下线
前面也分析过了，在客户端注册、续约、下线的时候，都会同步到集群其它节点。可以看到都调用了 replicateToPeers 方法来复制到其它集群。
1 /////////////////////// 注册 ///////////////////////
2 public void register(final InstanceInfo info, final boolean isReplication) {
3 int leaseDuration = Lease.DEFAULT_DURATION_IN_SECS;
4 // 如果实例中没有周期的配置，就设置为默认的 90 秒
5 if (info.getLeaseInfo() != null && info.getLeaseInfo().getDurationInSecs() > 0) {
6 leaseDuration = info.getLeaseInfo().getDurationInSecs();
7 }
8 // 注册实例
9 super.register(info, leaseDuration, isReplication);
10 // 复制到集群其它 server 节点
11 replicateToPeers(Action.Register, info.getAppName(), info.getId(), info, null, isReplication);
12 }
13
14
15 /////////////////////// 下线 ///////////////////////
16 public boolean cancel(final String appName, final String id,
17 final boolean isReplication) {
18 if (super.cancel(appName, id, isReplication)) {
19 replicateToPeers(Action.Cancel, appName, id, null, null, isReplication);
20
21 return true;
22 }
23 return false;
24 }
25
26
27 /////////////////////// 续约 ///////////////////////
28 public boolean renew(final String appName, final String id, final boolean isReplication) {
29 // 调用父类（AbstractInstanceRegistry）的 renew 续约
30 if (super.renew(appName, id, isReplication)) {
31 // 续约完成后同步到集群其它节点
32 replicateToPeers(Action.Heartbeat, appName, id, null, null, isReplication);
33 return true;
34 }
35 return false;
36 }
② 同步到其它节点
来看看 replicateToPeers 方法：
- 首先判断 isReplication 参数，如果是集群复制操作，最近一分钟复制次数 numberOfReplicationsLastMin + 1。isReplication 是在请求头中指定的，请求头为 PeerEurekaNode.HEADER_REPLICATION（x-netflix-discovery-replication）。
- 接着遍历集群列表，复制实例操作到集群节点中。前面也分析过了，PeerEurekaNode 就代表了一个 eureka-server，PeerEurekaNodes 就代表了 eureka-server 集群。
- 复制实例操作到集群的方法 replicateInstanceActionsToPeers 就是根据不同的操作类型调用集群 PeerEurekaNode 对应的方法完成操作复制。
1 private void replicateToPeers(Action action, String appName, String id,
2 InstanceInfo info /* optional */,
3 InstanceStatus newStatus /* optional */, boolean isReplication) {
4 Stopwatch tracer = action.getTimer().start();
5 try {
6 if (isReplication) {
7 // 如果是来自其它server节点的注册请求，则最近一分钟集群同步次数+1
8 numberOfReplicationsLastMin.increment();
9 }
10 // If it is a replication already, do not replicate again as this will create a poison replication
11 if (peerEurekaNodes == Collections.EMPTY_LIST || isReplication) {
12 return;
13 }
14
15 // 如果是来自客户端的注册请求，就同步到集群中其它server节点
16 for (final PeerEurekaNode node : peerEurekaNodes.getPeerEurekaNodes()) {
17 // If the url represents this host, do not replicate to yourself.
18 if (peerEurekaNodes.isThisMyUrl(node.getServiceUrl())) {
19 continue;
20 }
21
22 replicateInstanceActionsToPeers(action, appName, id, info, newStatus, node);
23 }
24 } finally {
25 tracer.stop();
26 }
27 }
1 private void replicateInstanceActionsToPeers(Action action, String appName,
2 String id, InstanceInfo info, InstanceStatus newStatus,
3 PeerEurekaNode node) {
4 try {
5 InstanceInfo infoFromRegistry;
6 CurrentRequestVersion.set(Version.V2);
7 switch (action) {
8 case Cancel:
9 // 下线
10 node.cancel(appName, id);
11 break;
12 case Heartbeat:
13 InstanceStatus overriddenStatus = overriddenInstanceStatusMap.get(id);
14 infoFromRegistry = getInstanceByAppAndId(appName, id, false);
15 // 续约
16 node.heartbeat(appName, id, infoFromRegistry, overriddenStatus, false);
17 break;
18 case Register:
19 // 注册
20 node.register(info);
21 break;
22 case StatusUpdate:
23 infoFromRegistry = getInstanceByAppAndId(appName, id, false);
24 node.statusUpdate(appName, id, newStatus, infoFromRegistry);
25 break;
26 case DeleteStatusOverride:
27 infoFromRegistry = getInstanceByAppAndId(appName, id, false);
28 node.deleteStatusOverride(appName, id, infoFromRegistry);
29 break;
30 }
31 } catch (Throwable t) {
32 logger.error("Cannot replicate information to {} for action {}", node.getServiceUrl(), action.name(), t);
33 } finally {
34 CurrentRequestVersion.remove();
35 }
36 }
③ isReplication
PeerEurekaNode 与 eureka-server 通信的组件是 JerseyReplicationClient，这个类重写了 addExtraHeaders 方法，并添加了请求头 PeerEurekaNode.HEADER_REPLICATION，设置为 true。
这样其它 eureka-server 收到这个复制操作后，就知道是来自集群节点的同步操作，就不会再同步给其它节点了，从而避免死循环。
1 @Override
2 protected void addExtraHeaders(Builder webResource) {
3 webResource.header(PeerEurekaNode.HEADER_REPLICATION, "true");
4 }
十二、集群同步机制
Eureka Server 集群间同步机制还是比较复杂的，试想如果每次客户端的请求一过来，比如注册、心跳，然后 eureka-server 就立马同步给集群中其它 server 节点，那 eureka-server 这种 Peer to Peer 的模式实际上就无法分担客户端的写操作压力，相当于每个 eureka-server 接收到的请求量都是一样的。那 eureka server 为了避免这种情况，底层采用了三层队列，加批量任务的方式来进行集群间的同步。简单来说就是先将客户端操作放入队列中，然后从队列中取出一批操作，然后将这一批操作发送给其它 Server 节点，Server节点接收到之后再将这批操作解析到本地。下面就来详细看看是如何实现的。
1、集群节点 PeerEurekaNode
之前分析 eureka-server 启动初始化的时候，EurekaBootStrap 初始化了代表集群的 PeerEurekaNodes，它里面又根据配置的注册中心地址构造了 PeerEurekaNode，集群间同步核心的组件就是这个 PeerEurekaNode 了。下面以客户端注册为例来看下是如何同步的。
① 注册同步
replicateInstanceActionsToPeers 中调用了 PeerEurekaNode 的 register 方法来同步注册操作到集群。
node.register 方法：
- 可以看到先计算了过期时间，为当前时间 + 租约间隔时间（默认90秒）
- 然后调用了 batchingDispatcher 批量任务分发器来处理任务，提交了一个 InstanceReplicationTask 的实例，其 execute 方法中调用了 replicationClient 来向这个 server 注册同步。
1 public void register(final InstanceInfo info) throws Exception {
2 // 过期时间：当前时间 + 租约时间（默认90秒）
3 long expiryTime = System.currentTimeMillis() + getLeaseRenewalOf(info);
4 batchingDispatcher.process(
5 taskId("register", info),
6 new InstanceReplicationTask(targetHost, Action.Register, info, null, true) {
7 public EurekaHttpResponse<Void> execute() {
8 return replicationClient.register(info);
9 }
10 },
11 expiryTime
12 );
13 }
再看下 getLeaseRenewalOf 这个方法，这里应该是有bug的，这个方法返回的是毫秒数，可以看到它的卫语句的else部分是乘以 1000 了的，而 if 部分则没有，返回的是 90，不过这里 info.getLeaseInfo() 应该都不会为 null。
1 private static int getLeaseRenewalOf(InstanceInfo info) {
2 // bug : Lease.DEFAULT_DURATION_IN_SECS * 1000
3 return (info.getLeaseInfo() == null ? Lease.DEFAULT_DURATION_IN_SECS : info.getLeaseInfo().getRenewalIntervalInSecs()) * 1000;
4 }
② PeerEurekaNode 的构造
batchingDispatcher 是在 PeerEurekaNode 的构造方法中初始化的，来看下它的构造方法：
- registry：本地注册表
- targetHost：eureka-server host
- replicationClient：基于 jersey 的集群复制客户端通信组件，它在请求头中设置了 PeerEurekaNode.HEADER_REPLICATION 为 true
- serviceUrl：eureka-server 地址
- maxProcessingDelayMs：最大处理延迟毫秒数，默认为30000毫秒，即30秒，在下线的时候有用到
- batcherName：批处理器名称
- taskProcessor：复制任务处理器，它封装了 targetHost 和 replicationClient，主要就是 ReplicationTaskProcessor 在处理批量任务的提交
- batchingDispatcher：批量任务分发器，它会将任务打成一个批次提交到 eureka-server，避免多次请求eureka-server，注册时就是先用这个分发器提交的任务
- nonBatchingDispatcher：非批量任务分发器，就是一个任务一个任务的提交
1 public PeerEurekaNode(PeerAwareInstanceRegistry registry, String targetHost, String serviceUrl, HttpReplicationClient replicationClient, EurekaServerConfig config) {
2 this(registry, targetHost, serviceUrl, replicationClient, config, BATCH_SIZE, MAX_BATCHING_DELAY_MS, RETRY_SLEEP_TIME_MS, SERVER_UNAVAILABLE_SLEEP_TIME_MS);
3 }
4
5 PeerEurekaNode(PeerAwareInstanceRegistry registry, String targetHost, String serviceUrl,
6 HttpReplicationClient replicationClient, EurekaServerConfig config,
7 int batchSize, long maxBatchingDelayMs,
8 long retrySleepTimeMs, long serverUnavailableSleepTimeMs) {
9 this.registry = registry;
10 // 集群节点 host
11 this.targetHost = targetHost;
12 this.replicationClient = replicationClient;
13
14 // 集群节点地址
15 this.serviceUrl = serviceUrl;
16 this.config = config;
17 // 最大延迟时间 默认30秒
18 this.maxProcessingDelayMs = config.getMaxTimeForReplication();
19
20 // 批处理器名称
21 String batcherName = getBatcherName();
22
23 // 复制任务处理器
24 ReplicationTaskProcessor taskProcessor = new ReplicationTaskProcessor(targetHost, replicationClient);
25 // 批量任务分发器
26 this.batchingDispatcher = TaskDispatchers.createBatchingTaskDispatcher(
27 batcherName,
28 // 复制池里最大容量，默认 10000
29 config.getMaxElementsInPeerReplicationPool(),
30 batchSize, // 250
31 // 同步使用的最大线程数 默认 20
32 config.getMaxThreadsForPeerReplication(),
33 maxBatchingDelayMs, // 500
34 serverUnavailableSleepTimeMs, // 1000
35 retrySleepTimeMs, // 100
36 taskProcessor
37 );
38 // 单个任务分发器
39 this.nonBatchingDispatcher = TaskDispatchers.createNonBatchingTaskDispatcher(
40 targetHost,
41 config.getMaxElementsInStatusReplicationPool(),
42 config.getMaxThreadsForStatusReplication(),
43 maxBatchingDelayMs,
44 serverUnavailableSleepTimeMs,
45 retrySleepTimeMs,
46 taskProcessor
47 );
48 }
2、批量分发器 TaskDispatcher
创建 batchingDispatcher 时调用了 TaskDispatchers.createBatchingTaskDispatcher 方法创建了 batchingDispatcher。
首先看下 createBatchingTaskDispatcher 的参数及默认值，后面分析代码的时候会用到这些参数：
- id：批量分发器的名称
- maxBufferSize：缓存池最大数量，默认 10000
- workloadSize：工作负载数量，即一个批次最多多少任务，默认 250
- workerCount：工作者数量，这个是线程池线程工作线程的数量，默认20
- maxBatchingDelay：批量任务最大延迟毫秒数，默认为 500 毫秒
- congestionRetryDelayMs：阻塞重试延迟毫秒数，默认为 1000 毫秒
- networkFailureRetryMs：网络失败重试延迟毫秒数，默认为 100 毫秒
- taskProcessor：任务处理器，即 ReplicationTaskProcessor
再看下这个方法：
- 首先创建了一个接收者执行器 AcceptorExecutor，主要的参数是缓存、时间相关的
- 再创建了一个任务处理器 TaskExecutors，主要的参数是工作线程数、任务处理器以及接收者执行器，可以猜测这应该就是最终执行批量任务提交的执行器
- 最后创建了任务分发器 TaskDispatcher，从它的 process 方法可以看出，分发器提交的任务实际上又提交给了 AcceptorExecutor
从这里可以知道，前面注册时 batchingDispatcher.process() 提交的任务其实就是分发到 acceptorExecutor 这个接收者执行器了。创建的这个分发器 TaskDispatcher 主要有接收者执行器 AcceptorExecutor 和 任务处理器 TaskExecutors 这两个组件，核心的分发功能就在这两个组件中。
1 public static <ID, T> TaskDispatcher<ID, T> createBatchingTaskDispatcher(String id, int maxBufferSize, int workloadSize,
2 int workerCount, long maxBatchingDelay, long congestionRetryDelayMs,
3 long networkFailureRetryMs, TaskProcessor<T> taskProcessor) {
4 // 接收者执行器 AcceptorExecutor
5 final AcceptorExecutor<ID, T> acceptorExecutor = new AcceptorExecutor<>(
6 id, maxBufferSize, workloadSize, maxBatchingDelay, congestionRetryDelayMs, networkFailureRetryMs
7 );
8
9 // 任务处理器 TaskExecutors， workerCount = 20
10 final TaskExecutors<ID, T> taskExecutor = TaskExecutors.batchExecutors(id, workerCount, taskProcessor, acceptorExecutor);
11
12 return new TaskDispatcher<ID, T>() {
13 @Override
14 public void process(ID id, T task, long expiryTime) {
15 // 任务由 acceptorExecutor 处理
16 acceptorExecutor.process(id, task, expiryTime);
17 }
18
19 @Override
20 public void shutdown() {
21 acceptorExecutor.shutdown();
22 taskExecutor.shutdown();
23 }
24 };
25 }
3、接收者执行器 AcceptorExecutor
先看下创建 AcceptorExecutor 的构造方法：
- 根据 congestionRetryDelayMs、networkFailureRetryMs 创建了一个时间调整器 TrafficShaper，应该主要就是用来调整补偿时间的
- 然后创建了一个后台线程 acceptorThread，它运行的任务是 AcceptorRunner，主要就是将任务转成批量任务的
- 最后就是注册了一些监控统计之类的
1 AcceptorExecutor(String id,
2 int maxBufferSize,
3 int maxBatchingSize,
4 long maxBatchingDelay,
5 long congestionRetryDelayMs,
6 long networkFailureRetryMs) {
7 // 批处理器名称
8 this.id = id;
9 // 最大缓冲数：10000
10 this.maxBufferSize = maxBufferSize;
11 // 每批最大数量：250
12 this.maxBatchingSize = maxBatchingSize;
13 // 最大延迟时间：500 ms
14 this.maxBatchingDelay = maxBatchingDelay;
15 // 时间调整器
16 // congestionRetryDelayMs 阻塞重试延迟时间，1000ms
17 // networkFailureRetryMs 网络异常重试时间，100ms
18 this.trafficShaper = new TrafficShaper(congestionRetryDelayMs, networkFailureRetryMs);
19
20 // 接收者后台处理线程
21 ThreadGroup threadGroup = new ThreadGroup("eurekaTaskExecutors");
22 this.acceptorThread = new Thread(threadGroup, new AcceptorRunner(), "TaskAcceptor-" + id);
23 this.acceptorThread.setDaemon(true);
24 this.acceptorThread.start();
25
26 // 监控统计相关
27 final double[] percentiles = {50.0, 95.0, 99.0, 99.5};
28 final StatsConfig statsConfig = new StatsConfig.Builder()
29 .withSampleSize(1000)
30 .withPercentiles(percentiles)
31 .withPublishStdDev(true)
32 .build();
33 final MonitorConfig config = MonitorConfig.builder(METRIC_REPLICATION_PREFIX + "batchSize").build();
34 this.batchSizeMetric = new StatsTimer(config, statsConfig);
35 try {
36 Monitors.registerObject(id, this);
37 } catch (Throwable e) {
38 logger.warn("Cannot register servo monitor for this object", e);
39 }
40 }
然后看看 AcceptorExecutor 的属性，它定义了几个队列以及容器来处理批量任务，我们先知道有这些东西，后面再来看看都怎么使用的。
然后可以看到 AcceptorExecutor 大量使用了并发包下的一些类，以及队列的特性，这里我们需要了解下这些类的特性：
- LinkedBlockingQueue：基于链表的单端阻塞队列，就是队尾入队，队首出队
- Deque：双端队列，就是队首、队尾都可以入队、出队
- Semaphore：信号量，需要通过 acquire（或 tryAcquire） 获取到许可证之后才可以进入临界区，通过 release 释放许可证。只要能拿到许可证，Semaphore 是可以允许多个线程进入临界区的。另外注意它这里设置的许可证数量是0，说明要先调用了 release 放入一个许可证，才有可能调用 acquire 获取到许可证。
1 // 接收任务的队列 2 private final BlockingQueue<TaskHolder<ID, T>> acceptorQueue = new LinkedBlockingQueue<>(); 3 // 重试任务的队列 4 private final BlockingDeque<TaskHolder<ID, T>> reprocessQueue = new LinkedBlockingDeque<>(); 5 // 后台接收者线程 6 private final Thread acceptorThread; 7 // 待处理任务容器 8 private final Map<ID, TaskHolder<ID, T>> pendingTasks = new HashMap<>(); 9 // 处理中的队列 10 private final Deque<ID> processingOrder = new LinkedList<>(); 11 12 // 单项队列请求的信号量 13 private final Semaphore singleItemWorkRequests = new Semaphore(0); 14 // 单项任务队列 15 private final BlockingQueue<TaskHolder<ID, T>> singleItemWorkQueue = new LinkedBlockingQueue<>(); 16 17 // 批量队列请求的信号量 18 private final Semaphore batchWorkRequests = new Semaphore(0); 19 // 批量任务队列 20 private final BlockingQueue<List<TaskHolder<ID, T>>> batchWorkQueue = new LinkedBlockingQueue<>(); 21 // 时间调整器 22 private final TrafficShaper trafficShaper;
TaskDispatcher 调用 acceptorExecutor.process 将任务转给 AcceptorExecutor，可以看到就是将任务添加到接收者队列 acceptorQueue 的队尾了。
1 void process(ID id, T task, long expiryTime) {
2 acceptorQueue.add(new TaskHolder<ID, T>(id, task, expiryTime));
3 acceptedTasks++;
4 }
4、接收者任务 AcceptorRunner
任务添加到 acceptorQueue 了，那任务在哪处理的呢？这就是在 AcceptorRunner 这个任务里去处理的了，这个任务比较复杂，我先把整个代码放出来，再来分析。
1 class AcceptorRunner implements Runnable {
2 @Override
3 public void run() {
4 long scheduleTime = 0;
5 while (!isShutdown.get()) {
6 try {
7 // 排出输入队列的任务：将 reprocessQueue、acceptorQueue 队列的任务转移到 pendingTasks
8 drainInputQueues();
9
10 // 待处理的数量
11 int totalItems = processingOrder.size();
12
13 long now = System.currentTimeMillis();
14 if (scheduleTime < now) {
15 // 时间补偿，正常情况下 transmissionDelay() 返回 0
16 scheduleTime = now + trafficShaper.transmissionDelay();
17 }
18 if (scheduleTime <= now) {
19 // 分配批量工作任务：将 pendingTasks 的任务分一批到（最多250个） batchWorkQueue 队列中
20 assignBatchWork();
21 // 分配单项工作任务：pendingTasks 如果还有剩余任务，将没有过期的转移到 singleItemWorkQueue 队列中
22 assignSingleItemWork();
23 }
24
25 // If no worker is requesting data or there is a delay injected by the traffic shaper,
26 // sleep for some time to avoid tight loop.
27 if (totalItems == processingOrder.size()) {
28 Thread.sleep(10);
29 }
30 } catch (InterruptedException ex) {
31 // Ignore
32 } catch (Throwable e) {
33 // Safe-guard, so we never exit this loop in an uncontrolled way.
34 logger.warn("Discovery AcceptorThread error", e);
35 }
36 }
37 }
38
39 private boolean isFull() {
40 // 待处理的任务 >= 10000，也就是说 pendingTasks 最多放 10000 个任务
41 return pendingTasks.size() >= maxBufferSize;
42 }
43
44 private void drainInputQueues() throws InterruptedException {
45 do {
46 // 排出 reprocessQueue，将 reprocessQueue 队列的任务转移到 pendingTasks
47 drainReprocessQueue();
48 // 排出 acceptorQueue，将 acceptorQueue 队列的任务转移到 pendingTasks
49 drainAcceptorQueue();
50
51 if (isShutdown.get()) {
52 break;
53 }
54 // If all queues are empty, block for a while on the acceptor queue
55 if (reprocessQueue.isEmpty() && acceptorQueue.isEmpty() && pendingTasks.isEmpty()) {
56 // 等待任务放入 acceptorQueue，等待 10 毫秒
57 TaskHolder<ID, T> taskHolder = acceptorQueue.poll(10, TimeUnit.MILLISECONDS);
58 if (taskHolder != null) {
59 // 放入之后 acceptorQueue、pendingTasks 就不为空了
60 appendTaskHolder(taskHolder);
61 }
62 }
63 // pendingTasks 为空、acceptorQueue 不为空、reprocessQueue不为空时，就会一直循环
64 // 如果所有任务都处理完了，reprocessQueue、acceptorQueue、pendingTasks 都是空的，
65 // 这时就会循环等待任务进入 acceptorQueue，每次等待 10 毫秒
66 } while (!reprocessQueue.isEmpty() || !acceptorQueue.isEmpty() || pendingTasks.isEmpty());
67 }
68
69 private void drainAcceptorQueue() {
70 while (!acceptorQueue.isEmpty()) {
71 // 将 acceptorQueue 的任务转移到 pendingTasks
72 appendTaskHolder(acceptorQueue.poll());
73 }
74 }
75
76 private void drainReprocessQueue() {
77 long now = System.currentTimeMillis();
78 while (!reprocessQueue.isEmpty() && !isFull()) {
79 // 从 reprocessQueue 队尾取出任务
80 TaskHolder<ID, T> taskHolder = reprocessQueue.pollLast();
81 ID id = taskHolder.getId();
82 if (taskHolder.getExpiryTime() <= now) {
83 // 任务过期
84 expiredTasks++;
85 } else if (pendingTasks.containsKey(id)) {
86 // pendingTasks 已存在
87 overriddenTasks++;
88 } else {
89 // 将 reprocessQueue 队列的任务放到 pendingTasks
90 pendingTasks.put(id, taskHolder);
91 // 添加到 processingOrder 队列的头部，reprocessQueue 是失败重试的队列，所以优先级高一些
92 processingOrder.addFirst(id);
93 }
94 }
95 if (isFull()) {
96 queueOverflows += reprocessQueue.size();
97 // pendingTasks 满了，就清空 reprocessQueue
98 reprocessQueue.clear();
99 }
100 }
101
102 private void appendTaskHolder(TaskHolder<ID, T> taskHolder) {
103 if (isFull()) {
104 // pendingTasks 满了就移除一个元素
105 pendingTasks.remove(processingOrder.poll());
106 queueOverflows++;
107 }
108 // 将 acceptorQueue 里的任务放到 pendingTasks
109 TaskHolder<ID, T> previousTask = pendingTasks.put(taskHolder.getId(), taskHolder);
110 if (previousTask == null) {
111 // 原本不存在，将任务ID添加到 processingOrder 队列的最后
112 processingOrder.add(taskHolder.getId());
113 } else {
114 // 已经存在了，就是覆盖
115 overriddenTasks++;
116 }
117 }
118
119 void assignSingleItemWork() {
120 if (!processingOrder.isEmpty()) {
121 if (singleItemWorkRequests.tryAcquire(1)) {
122 long now = System.currentTimeMillis();
123 while (!processingOrder.isEmpty()) {
124 ID id = processingOrder.poll();
125 TaskHolder<ID, T> holder = pendingTasks.remove(id);
126 if (holder.getExpiryTime() > now) {
127 // 将 pendingTasks 的任务移到 singleItemWorkQueue
128 singleItemWorkQueue.add(holder);
129 return;
130 }
131 expiredTasks++;
132 }
133 singleItemWorkRequests.release();
134 }
135 }
136 }
137
138 void assignBatchWork() {
139 // 有足够的任务做一个批处理
140 if (hasEnoughTasksForNextBatch()) {
141 if (batchWorkRequests.tryAcquire(1)) {
142 long now = System.currentTimeMillis();
143 // 一批任务最多 250 个
144 int len = Math.min(maxBatchingSize, processingOrder.size());
145 List<TaskHolder<ID, T>> holders = new ArrayList<>(len);
146 // 将 pendingTasks 中的任务移动一批到 holders 中
147 // 也就是说，如果队列中有500个任务，这一批任务最多也是250个
148 while (holders.size() < len && !processingOrder.isEmpty()) {
149 ID id = processingOrder.poll();
150 TaskHolder<ID, T> holder = pendingTasks.remove(id);
151 if (holder.getExpiryTime() > now) {
152 holders.add(holder);
153 } else {
154 expiredTasks++;
155 }
156 }
157 if (holders.isEmpty()) {
158 batchWorkRequests.release();
159 } else {
160 batchSizeMetric.record(holders.size(), TimeUnit.MILLISECONDS);
161 // 添加到批量队列中
162 batchWorkQueue.add(holders);
163 }
164 }
165 }
166 }
167
168 // 是否有足够的任务做一个批处理
169 private boolean hasEnoughTasksForNextBatch() {
170 if (processingOrder.isEmpty()) {
171 return false;
172 }
173 if (pendingTasks.size() >= maxBufferSize) {
174 return true;
175 }
176
177 // 从 processingOrder 队首取一个任务ID，然后从 pendingTasks 读取这个任务。注意 peek() 只是取出元素，并不会移除队首的元素
178 TaskHolder<ID, T> nextHolder = pendingTasks.get(processingOrder.peek());
179 // 判断任务提交到现在的时间差是否超过最大批任务延迟时间（500毫秒）
180 long delay = System.currentTimeMillis() - nextHolder.getSubmitTimestamp();
181 return delay >= maxBatchingDelay;
182 }
183 }
View Code
先看它的 run 方法：
① 队列中的任务转移到待处理容器中
drainInputQueues 将输入队列（reprocessQueue、acceptorQueue）的任务转移到 pendingTasks 这个待处理容器中。
先是 drainReprocessQueue 将重处理队列 reprocessQueue 中的任务转移到 pendingTasks：
- 如果 pendingTasks 已满（超过10000），就直接清空了 reprocessQueue。任务丢弃会不会有影响呢？
-
否则，如果 reprocessQueue 非空，就从 reprocessQueue 队尾一个个取出来：
- 如果过期了就丢掉这个任务，说明已经超过续约周期了（90秒）。比如实例注册，如果多次同步失败后，然后就直接丢弃，那不是其它 server 永远无法知道注册的这个实例？后面再分析这个问题。
- 如果 pendingTasks 已经存在了，也丢弃这个重试任务
- 否则就添加到 pendingTasks 中，并且往 processingOrder 的头部添加了任务ID
- 注意它这里是从 reprocessQueue 队尾一个个取出，放入 processingOrder 头部，最终任务在 processingOrder 中的顺序跟 reprocessQueue 是一样的
然后是 drainAcceptorQueue 将接收者队列 acceptorQueue 中的任务转移到 pendingTasks：
- 只要 acceptorQueue 非空，就从队首取出任务
- 如果 pendingTasks 已满，则从 processingOrder 队首取出第一个任务的ID，并从 pendingTasks 中移除这个任务
- 否则就将任务添加到 pendingTasks，如果之前不存在相同ID的任务，就将任务ID添加到 processingOrder 队尾
- 注意它这里是从 acceptorQueue 队首取出任务，放到 processingOrder 队尾，最终任务在 processingOrder 中的顺序跟 acceptorQueue 是一样的
从这段任务转移以及后面的使用来看，processingOrder 将决定任务的处理顺序，最前面的将最先处理，也说明了 reprocessQueue 的优先级比 acceptorQueue 更高。而 pendingTasks 是一个 key-value 的队列，便于快速通过ID读取任务。
1 private void drainAcceptorQueue() {
2 while (!acceptorQueue.isEmpty()) {
3 // 将 acceptorQueue 的任务转移到 pendingTasks
4 appendTaskHolder(acceptorQueue.poll());
5 }
6 }
7
8 private void drainReprocessQueue() {
9 long now = System.currentTimeMillis();
10 while (!reprocessQueue.isEmpty() && !isFull()) {
11 // 从 reprocessQueue 队尾取出任务
12 TaskHolder<ID, T> taskHolder = reprocessQueue.pollLast();
13 ID id = taskHolder.getId();
14 if (taskHolder.getExpiryTime() <= now) {
15 // 任务过期
16 expiredTasks++;
17 } else if (pendingTasks.containsKey(id)) {
18 // pendingTasks 已存在
19 overriddenTasks++;
20 } else {
21 // 将 reprocessQueue 队列的任务放到 pendingTasks
22 pendingTasks.put(id, taskHolder);
23 // 添加到 processingOrder 队列的头部，reprocessQueue 是失败重试的队列，所以优先级高一些
24 processingOrder.addFirst(id);
25 }
26 }
27 if (isFull()) {
28 queueOverflows += reprocessQueue.size();
29 // pendingTasks 满了，就清空 reprocessQueue
30 reprocessQueue.clear();
31 }
32 }
33
34 private void appendTaskHolder(TaskHolder<ID, T> taskHolder) {
35 if (isFull()) {
36 // pendingTasks 满了就移除一个元素
37 pendingTasks.remove(processingOrder.poll());
38 queueOverflows++;
39 }
40 // 将 acceptorQueue 里的任务放到 pendingTasks
41 TaskHolder<ID, T> previousTask = pendingTasks.put(taskHolder.getId(), taskHolder);
42 if (previousTask == null) {
43 // 原本不存在，将任务ID添加到 processingOrder 队列的最后
44 processingOrder.add(taskHolder.getId());
45 } else {
46 // 已经存在了，就是覆盖
47 overriddenTasks++;
48 }
49 }
② 接下来通过 trafficShaper 获取了一个补偿时间，它主要是在发生阻塞或网络异常导致任务提交失败后，在任务调度周期内做一个时间补偿，这块等分析到提交任务失败的时候再回来看看。
1 long now = System.currentTimeMillis();
2 if (scheduleTime < now) {
3 // 时间补偿，正常情况下 transmissionDelay() 返回 0
4 scheduleTime = now + trafficShaper.transmissionDelay();
5 }
③ 任务打包
接着看 assignBatchWork ，它就是将任务打包成一个批次：
- 首先调用 hasEnoughTasksForNextBatch 判断是否有足够的任务来打成一个批次，注意它判断了最新提交的任务的时间是否超过了延迟时间 maxBatchingDelay（500ms），也就是说批次任务每隔500毫秒运行一次。
- 能够打包后，要获取 batchWorkRequests 信号量的一个许可证，因为许可证默认数量是 0，那一定是先有地方调用了 batchWorkRequests.release() 放入许可证，否则这里就不会打包了。
- 然后可以看出，一个批次的任务数量最多是250个
- 它从 processingOrder 的队首取出这个批次的任务ID，并从 pendingTasks 中取出任务，如果是过期的任务就直接丢弃了。
- 然后如果这个批次并没有任务，他才调用 batchWorkRequests.release() 释放了许可证，否则就把这个批次任务添加到批量工作队列 batchWorkQueue 中，注意并没有释放许可证。
1 void assignBatchWork() {
2 // 有足够的任务做一个批处理
3 if (hasEnoughTasksForNextBatch()) {
4 // 获取许可证
5 if (batchWorkRequests.tryAcquire(1)) {
6 long now = System.currentTimeMillis();
7 // 一批任务最多 250 个
8 int len = Math.min(maxBatchingSize, processingOrder.size());
9 List<TaskHolder<ID, T>> holders = new ArrayList<>(len);
10 // 将 pendingTasks 中的任务移动一批到 holders 中
11 // 也就是说，如果队列中有500个任务，这一批任务最多也是250个
12 while (holders.size() < len && !processingOrder.isEmpty()) {
13 ID id = processingOrder.poll();
14 TaskHolder<ID, T> holder = pendingTasks.remove(id);
15 if (holder.getExpiryTime() > now) {
16 holders.add(holder);
17 } else {
18 expiredTasks++;
19 }
20 }
21 if (holders.isEmpty()) {
22 batchWorkRequests.release();
23 } else {
24 batchSizeMetric.record(holders.size(), TimeUnit.MILLISECONDS);
25 // 添加到批量队列中
26 batchWorkQueue.add(holders);
27 }
28 }
29 }
30 }
31
32 // 是否有足够的任务做一个批处理
33 private boolean hasEnoughTasksForNextBatch() {
34 if (processingOrder.isEmpty()) {
35 return false;
36 }
37 if (pendingTasks.size() >= maxBufferSize) {
38 return true;
39 }
40
41 // 从 processingOrder 队首取一个任务ID，然后从 pendingTasks 读取这个任务。注意 peek() 只是取出元素，并不会移除队首的元素
42 TaskHolder<ID, T> nextHolder = pendingTasks.get(processingOrder.peek());
43 // 判断任务提交到现在的时间差是否超过最大批任务延迟时间（500毫秒）
44 long delay = System.currentTimeMillis() - nextHolder.getSubmitTimestamp();
45 return delay >= maxBatchingDelay;
46 }
接着看分配单项任务的方法 assignSingleItemWork：
- 如果 processingOrder 非空且获取到了 singleItemWorkRequests 信号量的许可证，就将 processingOrder 队列剩余的任务都取出来，放入单项工作队列 singleItemWorkQueue 中
- 也就是前面已经打了一批任务（250个）之后，processingOrder 中还有任务，就全部取出来放到 singleItemWorkQueue 队列中
1 void assignSingleItemWork() {
2 if (!processingOrder.isEmpty()) {
3 if (singleItemWorkRequests.tryAcquire(1)) {
4 long now = System.currentTimeMillis();
5 while (!processingOrder.isEmpty()) {
6 ID id = processingOrder.poll();
7 TaskHolder<ID, T> holder = pendingTasks.remove(id);
8 if (holder.getExpiryTime() > now) {
9 // 将 pendingTasks 的任务移到 singleItemWorkQueue
10 singleItemWorkQueue.add(holder);
11 return;
12 }
13 expiredTasks++;
14 }
15 singleItemWorkRequests.release();
16 }
17 }
18 }
5、任务处理器 TaskExecutors
batchWorkQueue 中的批量任务以及 singleItemWorkQueue 中的单项任务都已经准备好了，那是在哪里发送到集群节点的呢，那就是任务执行器 TaskExecutors 了。
① 创建 TaskExecutors
从创建 TaskExecutors 的方法中可以看出：
- 批量处理任务的类是 BatchWorkerRunnable，它主要就是处理批量任务队列 batchWorkQueue 中的任务
- 处理单项任务的类是 SingleTaskWorkerRunnable，它主要就是处理单项任务队列 singleItemWorkQueue 中的任务
- TaskExecutors 创建了一个线程池，batchExecutors 默认有20个工作线程（不太理解他为什么不用JDK现成的线程池。。），singleItemExecutors 默认只有一个工作线程。
1 static <ID, T> TaskExecutors<ID, T> singleItemExecutors(final String name, int workerCount, final TaskProcessor<T> processor, final AcceptorExecutor<ID, T> acceptorExecutor) {
2 final AtomicBoolean isShutdown = new AtomicBoolean();
3 final TaskExecutorMetrics metrics = new TaskExecutorMetrics(name);
4 registeredMonitors.put(name, metrics);
5 // workerCount = 1
6 return new TaskExecutors<>(idx -> new SingleTaskWorkerRunnable<>("TaskNonBatchingWorker-" + name + '-' + idx, isShutdown, metrics, processor, acceptorExecutor), workerCount, isShutdown);
7 }
8
9 ////////////////////////////////////////////////
10
11 static <ID, T> TaskExecutors<ID, T> batchExecutors(final String name, int workerCount, final TaskProcessor<T> processor, final AcceptorExecutor<ID, T> acceptorExecutor) {
12 final AtomicBoolean isShutdown = new AtomicBoolean();
13 final TaskExecutorMetrics metrics = new TaskExecutorMetrics(name);
14 registeredMonitors.put(name, metrics);
15 // BatchWorkerRunnable 批量任务处理
16 return new TaskExecutors<>(idx -> new BatchWorkerRunnable<>("TaskBatchingWorker-" + name + '-' + idx, isShutdown, metrics, processor, acceptorExecutor), workerCount, isShutdown);
17 }
18
19 ////////////////////////////////////////////////
20
21 private final List<Thread> workerThreads;
22
23 TaskExecutors(WorkerRunnableFactory<ID, T> workerRunnableFactory, int workerCount, AtomicBoolean isShutdown) {
24 this.isShutdown = isShutdown;
25 // 工作线程集合
26 this.workerThreads = new ArrayList<>();
27
28 // 创建20个线程，相当于是搞了一个线程池
29 ThreadGroup threadGroup = new ThreadGroup("eurekaTaskExecutors");
30 for (int i = 0; i < workerCount; i++) {
31 WorkerRunnable<ID, T> runnable = workerRunnableFactory.create(i);
32 Thread workerThread = new Thread(threadGroup, runnable, runnable.getWorkerName());
33 workerThreads.add(workerThread);
34 workerThread.setDaemon(true);
35 workerThread.start();
36 }
37 }
② BatchWorkerRunnable
看批量处理的任务：
- 首先 getWork 获取批量任务，它调用 taskDispatcher.requestWorkItems()，实际就是返回了 taskDispatcher 的 batchWorkQueue，并且调用 batchWorkRequests.release() 往信号量放入一个许可证，这样前面 AcceptorRunner 就可以得到许可证然后去打包批量任务了
- 如果 batchWorkQueue 中没有批量任务，可以看到是一直在 while 循环等待的，直到拿到一个批量任务。它这个 BatchWorkerRunnable 任务和前面的 AcceptorRunner 任务，感觉通过信号量的方式就形成了一个等待通知的机制，BatchWorkerRunnable 放入一个许可证，让 AcceptorRunner 拿到这个许可证去打个批次的任务过来。
- 拿到这个批次任务后，就调用 processor（ReplicationTaskProcessor）来处理任务。
- 如果任务处理结果是 Congestion（阻塞）、TransientError（传输失败）就要重处理，调用了 taskDispatcher.reprocess 将这个批次的任务提交到重处理队列 reprocessQueue 中。
1 static class BatchWorkerRunnable<ID, T> extends WorkerRunnable<ID, T> {
2
3 BatchWorkerRunnable(String workerName, AtomicBoolean isShutdown, TaskExecutorMetrics metrics, TaskProcessor<T> processor, AcceptorExecutor<ID, T> acceptorExecutor) {
4 super(workerName, isShutdown, metrics, processor, acceptorExecutor);
5 }
6
7 @Override
8 public void run() {
9 try {
10 while (!isShutdown.get()) {
11 // 获取一个批量任务
12 List<TaskHolder<ID, T>> holders = getWork();
13 metrics.registerExpiryTimes(holders);
14 // TaskHolder 提取 ReplicationTask
15 List<T> tasks = getTasksOf(holders);
16 // processor => 任务复制处理器 ReplicationTaskProcessor
17 ProcessingResult result = processor.process(tasks);
18 switch (result) {
19 case Success:
20 break;
21 case Congestion:
22 case TransientError:
23 // 阻塞或网络失败就重新处理这批任务
24 taskDispatcher.reprocess(holders, result);
25 break;
26 case PermanentError:
27 logger.warn("Discarding {} tasks of {} due to permanent error", holders.size(), workerName);
28 }
29 metrics.registerTaskResult(result, tasks.size());
30 }
31 } catch (InterruptedException e) {
32 // Ignore
33 } catch (Throwable e) {
34 // Safe-guard, so we never exit this loop in an uncontrolled way.
35 logger.warn("Discovery WorkerThread error", e);
36 }
37 }
38
39 private List<TaskHolder<ID, T>> getWork() throws InterruptedException {
40 // 获取批量队列 batchWorkQueue
41 BlockingQueue<List<TaskHolder<ID, T>>> workQueue = taskDispatcher.requestWorkItems();
42 List<TaskHolder<ID, T>> result;
43 do {
44 result = workQueue.poll(1, TimeUnit.SECONDS);
45 // 循环等待，直到取到一个批量任务
46 } while (!isShutdown.get() && result == null);
47 return (result == null) ? new ArrayList<>() : result;
48 }
49
50 private List<T> getTasksOf(List<TaskHolder<ID, T>> holders) {
51 List<T> tasks = new ArrayList<>(holders.size());
52 for (TaskHolder<ID, T> holder : holders) {
53 tasks.add(holder.getTask());
54 }
55 return tasks;
56 }
57 }
1 BlockingQueue<TaskHolder<ID, T>> requestWorkItem() {
2 singleItemWorkRequests.release();
3 return singleItemWorkQueue;
4 }
5
6 BlockingQueue<List<TaskHolder<ID, T>>> requestWorkItems() {
7 batchWorkRequests.release();
8 return batchWorkQueue;
9 }
③ 任务重处理
可以看到处理失败后，就是将这批任务添加到重处理队列 reprocessQueue 中去，然后向时间调整期注册失败，这就和前面 AcceptorRunner 处理 reprocessQueue 对应起来了。
1 void reprocess(List<TaskHolder<ID, T>> holders, ProcessingResult processingResult) {
2 // 添加到重处理队列 reprocessQueue
3 reprocessQueue.addAll(holders);
4 replayedTasks += holders.size();
5 // 时间调整器注册失败
6 trafficShaper.registerFailure(processingResult);
7 }
④ TrafficShaper
还记得前面 AcceptorRunner 中又这样一段代码，可以看到是通过 trafficShaper 计算了一个延迟时间，这里就来看看是如何计算的。
1 long now = System.currentTimeMillis();
2 if (scheduleTime < now) {
3 // 时间补偿，正常情况下 transmissionDelay() 返回 0
4 scheduleTime = now + trafficShaper.transmissionDelay();
5 }
6 if (scheduleTime <= now) {
7 // 分配批量工作任务：将 pendingTasks 的任务分一批到（最多250个） batchWorkQueue 队列中
8 assignBatchWork();
9 // 分配单项工作任务：pendingTasks 如果还有剩余任务，将没有过期的转移到 singleItemWorkQueue 队列中
10 assignSingleItemWork();
11 }
时间调整器 TrafficShaper：
- registerFailure 就是设置了失败的最后时间
- 然后看 transmissionDelay，以阻塞为例，如果上一次阻塞失败到现在 500 毫秒，那么 transmissionDelay 返回 500，那么 transmissionDelay 就大于 now 了，就不会打包任务了。
- 总结下来就是如果上一次阻塞导致批量任务提交失败，就延迟1000毫秒后执行。如果上一次网络导致批量任务提交失败，就延迟100毫秒执行。
1 TrafficShaper(long congestionRetryDelayMs, long networkFailureRetryMs) {
2 // 1000
3 this.congestionRetryDelayMs = Math.min(MAX_DELAY, congestionRetryDelayMs);
4 // 100
5 this.networkFailureRetryMs = Math.min(MAX_DELAY, networkFailureRetryMs);
6 }
7
8 void registerFailure(ProcessingResult processingResult) {
9 if (processingResult == ProcessingResult.Congestion) {
10 // 最后一次阻塞导致提交批处理失败的时间
11 lastCongestionError = System.currentTimeMillis();
12 } else if (processingResult == ProcessingResult.TransientError) {
13 // 最后一次网络原因导致提交批处理失败的时间
14 lastNetworkFailure = System.currentTimeMillis();
15 }
16 }
17
18 // 计算传输延迟的时间
19 long transmissionDelay() {
20 if (lastCongestionError == -1 && lastNetworkFailure == -1) {
21 return 0;
22 }
23
24 long now = System.currentTimeMillis();
25 if (lastCongestionError != -1) {
26 // 阻塞延迟时间
27 long congestionDelay = now - lastCongestionError;
28 if (congestionDelay >= 0 && congestionDelay < congestionRetryDelayMs) {
29 return congestionRetryDelayMs - congestionDelay;
30 }
31 lastCongestionError = -1;
32 }
33
34 if (lastNetworkFailure != -1) {
35 // 网络延迟时间
36 long failureDelay = now - lastNetworkFailure;
37 if (failureDelay >= 0 && failureDelay < networkFailureRetryMs) {
38 return networkFailureRetryMs - failureDelay;
39 }
40 lastNetworkFailure = -1;
41 }
42 return 0;
43 }
⑤ SingleTaskWorkerRunnable
单项任务处理跟批量任务处理的流程是类似的，只不过是一个个的发送同步操作，处理失败同样也会放入重处理队列中。
一个批量任务250个对于大部分场景来说其实不会触发单项任务的处理，如果微服务集群中有很多的实例，eureka 通过不断的轮询也能尽量使用批量处理，我觉得单项任务处理更像是对批量任务处理的一种补充。
6、复制任务处理器 ReplicationTaskProcessor
批量任务最终是提交到 ReplicationTaskProcessor 去处理的，可以看到，就是调用了 replicationClient 提交了批量任务，提交的接口是 POST peerreplication/batch，那我们就可以从这个入口去看 eureka-server 如何接收批量任务的。
1 public ProcessingResult process(List<ReplicationTask> tasks) {
2 // 任务封装到 ReplicationList
3 ReplicationList list = createReplicationListOf(tasks);
4 try {
5 // 提交批量任务：POST peerreplication/batch/
6 EurekaHttpResponse<ReplicationListResponse> response = replicationClient.submitBatchUpdates(list);
7 int statusCode = response.getStatusCode();
8 if (!isSuccess(statusCode)) {
9 if (statusCode == 503) {
10 logger.warn("Server busy (503) HTTP status code received from the peer {}; rescheduling tasks after delay", peerId);
11 return ProcessingResult.Congestion;
12 } else {
13 // Unexpected error returned from the server. This should ideally never happen.
14 logger.error("Batch update failure with HTTP status code {}; discarding {} replication tasks", statusCode, tasks.size());
15 return ProcessingResult.PermanentError;
16 }
17 } else {
18 // 处理批量任务结果
19 handleBatchResponse(tasks, response.getEntity().getResponseList());
20 }
21 } catch (Throwable e) {
22 if (maybeReadTimeOut(e)) {
23 //read timeout exception is more Congestion then TransientError, return Congestion for longer delay
24 return ProcessingResult.Congestion;
25 } else if (isNetworkConnectException(e)) {
26 logNetworkErrorSample(null, e);
27 return ProcessingResult.TransientError;
28 } else {
29 logger.error("Not re-trying this exception because it does not seem to be a network exception", e);
30 return ProcessingResult.PermanentError;
31 }
32 }
33 return ProcessingResult.Success;
34 }
7、接收复制同步请求
很容易找到批量任务提交的接口在 PeerReplicationResource 的 batchReplication 方法中。
可以看到，其实遍历批量任务，然后根据不同的操作类型，调用 XxxResource 接口进行对应的操作。比如注册，就是调用 applicationResource.addInstance 完成实例的注册。
1 @Path("/{version}/peerreplication")
2 @Produces({"application/xml", "application/json"})
3 public class PeerReplicationResource {
4
5 private static final Logger logger = LoggerFactory.getLogger(PeerReplicationResource.class);
6
7 private static final String REPLICATION = "true";
8
9 private final EurekaServerConfig serverConfig;
10 private final PeerAwareInstanceRegistry registry;
11
12 @Inject
13 PeerReplicationResource(EurekaServerContext server) {
14 this.serverConfig = server.getServerConfig();
15 this.registry = server.getRegistry();
16 }
17
18 public PeerReplicationResource() {
19 this(EurekaServerContextHolder.getInstance().getServerContext());
20 }
21
22 @Path("batch")
23 @POST
24 public Response batchReplication(ReplicationList replicationList) {
25 try {
26 ReplicationListResponse batchResponse = new ReplicationListResponse();
27 for (ReplicationInstance instanceInfo : replicationList.getReplicationList()) {
28 try {
29 // dispatch 分发任务
30 batchResponse.addResponse(dispatch(instanceInfo));
31 } catch (Exception e) {
32 batchResponse.addResponse(new ReplicationInstanceResponse(Status.INTERNAL_SERVER_ERROR.getStatusCode(), null));
33 logger.error("{} request processing failed for batch item {}/{}",
34 instanceInfo.getAction(), instanceInfo.getAppName(), instanceInfo.getId(), e);
35 }
36 }
37 return Response.ok(batchResponse).build();
38 } catch (Throwable e) {
39 logger.error("Cannot execute batch Request", e);
40 return Response.status(Status.INTERNAL_SERVER_ERROR).build();
41 }
42 }
43
44 private ReplicationInstanceResponse dispatch(ReplicationInstance instanceInfo) {
45 ApplicationResource applicationResource = createApplicationResource(instanceInfo);
46 InstanceResource resource = createInstanceResource(instanceInfo, applicationResource);
47
48 String lastDirtyTimestamp = toString(instanceInfo.getLastDirtyTimestamp());
49 String overriddenStatus = toString(instanceInfo.getOverriddenStatus());
50 String instanceStatus = toString(instanceInfo.getStatus());
51
52 Builder singleResponseBuilder = new Builder();
53 // 根据不同的类型分别处理
54 switch (instanceInfo.getAction()) {
55 case Register:
56 singleResponseBuilder = handleRegister(instanceInfo, applicationResource);
57 break;
58 case Heartbeat:
59 singleResponseBuilder = handleHeartbeat(serverConfig, resource, lastDirtyTimestamp, overriddenStatus, instanceStatus);
60 break;
61 case Cancel:
62 singleResponseBuilder = handleCancel(resource);
63 break;
64 case StatusUpdate:
65 singleResponseBuilder = handleStatusUpdate(instanceInfo, resource);
66 break;
67 case DeleteStatusOverride:
68 singleResponseBuilder = handleDeleteStatusOverride(instanceInfo, resource);
69 break;
70 }
71 return singleResponseBuilder.build();
72 }
73
74 /* Visible for testing */ ApplicationResource createApplicationResource(ReplicationInstance instanceInfo) {
75 return new ApplicationResource(instanceInfo.getAppName(), serverConfig, registry);
76 }
77
78 /* Visible for testing */ InstanceResource createInstanceResource(ReplicationInstance instanceInfo,
79 ApplicationResource applicationResource) {
80 return new InstanceResource(applicationResource, instanceInfo.getId(), serverConfig, registry);
81 }
82
83 private static Builder handleRegister(ReplicationInstance instanceInfo, ApplicationResource applicationResource) {
84 // addInstance
85 applicationResource.addInstance(instanceInfo.getInstanceInfo(), REPLICATION);
86 return new Builder().setStatusCode(Status.OK.getStatusCode());
87 }
88
89 private static Builder handleCancel(InstanceResource resource) {
90 // cancelLease
91 Response response = resource.cancelLease(REPLICATION);
92 return new Builder().setStatusCode(response.getStatus());
93 }
94
95 private static Builder handleHeartbeat(EurekaServerConfig config, InstanceResource resource, String lastDirtyTimestamp, String overriddenStatus, String instanceStatus) {
96 Response response = resource.renewLease(REPLICATION, overriddenStatus, instanceStatus, lastDirtyTimestamp);
97 int responseStatus = response.getStatus();
98 Builder responseBuilder = new Builder().setStatusCode(responseStatus);
99
100 if ("false".equals(config.getExperimental("bugfix.934"))) {
101 if (responseStatus == Status.OK.getStatusCode() && response.getEntity() != null) {
102 responseBuilder.setResponseEntity((InstanceInfo) response.getEntity());
103 }
104 } else {
105 if ((responseStatus == Status.OK.getStatusCode() || responseStatus == Status.CONFLICT.getStatusCode())
106 && response.getEntity() != null) {
107 responseBuilder.setResponseEntity((InstanceInfo) response.getEntity());
108 }
109 }
110 return responseBuilder;
111 }
112
113 private static Builder handleStatusUpdate(ReplicationInstance instanceInfo, InstanceResource resource) {
114 Response response = resource.statusUpdate(instanceInfo.getStatus(), REPLICATION, toString(instanceInfo.getLastDirtyTimestamp()));
115 return new Builder().setStatusCode(response.getStatus());
116 }
117
118 private static Builder handleDeleteStatusOverride(ReplicationInstance instanceInfo, InstanceResource resource) {
119 Response response = resource.deleteStatusUpdate(REPLICATION, instanceInfo.getStatus(),
120 instanceInfo.getLastDirtyTimestamp().toString());
121 return new Builder().setStatusCode(response.getStatus());
122 }
123
124 private static <T> String toString(T value) {
125 if (value == null) {
126 return null;
127 }
128 return value.toString();
129 }
130 }
View Code
8、集群数据同步冲突问题
Peer to Peer 模式重点要解决的一个问题是数据复制冲突的问题，因为 peer 节点间的相互复制并不能保证所有操作都成功。eureka 主要通过 lastDirtyTimestamp 标识和心跳来进行数据的最终修复，下面就来看下 eureka 如何处理数据冲突问题的。
① 先看续约的这个方法
- 在续约 renewLease 里，如果 lastDirtyTimestamp 不为空且允许时间戳不一致时进行同步（默认开启），就调用了 validateDirtyTimestamp 方法校验 lastDirtyTimestamp。
- 接着看 validateDirtyTimestamp，如果 lastDirtyTimestamp 与本地实例的 lastDirtyTimestamp 一致，说明数据是一致的，就续约成功，返回 OK（200）。
- 如果 lastDirtyTimestamp 大于 本地实例的 lastDirtyTimestamp，说明复制的实例最新更新的，出现数据冲突，返回 NOT_FOUND（404）。
- 如果 lastDirtyTimestamp 小于 本地实例的 lastDirtyTimestamp ，说明复制的实例是旧的，出现数据冲突，返回 CONFLICT（409），并且返回了本地的实例。
1 @PUT
2 public Response renewLease(
3 @HeaderParam(PeerEurekaNode.HEADER_REPLICATION) String isReplication,
4 @QueryParam("overriddenstatus") String overriddenStatus,
5 @QueryParam("status") String status,
6 @QueryParam("lastDirtyTimestamp") String lastDirtyTimestamp) {
7 boolean isFromReplicaNode = "true".equals(isReplication);
8 // 调用注册表的 renew 进行服务续约
9 boolean isSuccess = registry.renew(app.getName(), id, isFromReplicaNode);
10
11 // Not found in the registry, immediately ask for a register
12 if (!isSuccess) {
13 logger.warn("Not Found (Renew): {} - {}", app.getName(), id);
14 return Response.status(Status.NOT_FOUND).build();
15 }
16 // Check if we need to sync based on dirty time stamp, the client
17 // instance might have changed some value
18 Response response;
19 // 如果是复制操作，就校验 lastDirtyTimestamp
20 if (lastDirtyTimestamp != null && serverConfig.shouldSyncWhenTimestampDiffers()) {
21 response = this.validateDirtyTimestamp(Long.valueOf(lastDirtyTimestamp), isFromReplicaNode);
22 // Store the overridden status since the validation found out the node that replicates wins
23 if (response.getStatus() == Response.Status.NOT_FOUND.getStatusCode()
24 && (overriddenStatus != null)
25 && !(InstanceStatus.UNKNOWN.name().equals(overriddenStatus))
26 && isFromReplicaNode) {
27 registry.storeOverriddenStatusIfRequired(app.getAppName(), id, InstanceStatus.valueOf(overriddenStatus));
28 }
29 } else {
30 response = Response.ok().build();
31 }
32 logger.debug("Found (Renew): {} - {}; reply status={}", app.getName(), id, response.getStatus());
33 return response;
34 }
35
36 ///////////////////////////////////////////
37
38 private Response validateDirtyTimestamp(Long lastDirtyTimestamp, boolean isReplication) {
39 InstanceInfo appInfo = registry.getInstanceByAppAndId(app.getName(), id, false);
40 if (appInfo != null) {
41 // 如果复制传过来的实例中 lastDirtyTimestamp 不等于本地实例的 lastDirtyTimestamp
42 if ((lastDirtyTimestamp != null) && (!lastDirtyTimestamp.equals(appInfo.getLastDirtyTimestamp()))) {
43 Object[] args = {id, appInfo.getLastDirtyTimestamp(), lastDirtyTimestamp, isReplication};
44
45 if (lastDirtyTimestamp > appInfo.getLastDirtyTimestamp()) {
46 logger.debug(
47 "Time to sync, since the last dirty timestamp differs -"
48 + " ReplicationInstance id : {},Registry : {} Incoming: {} Replication: {}",
49 args);
50 // 如果复制实例的 lastDirtyTimestamp > 本地实例的 lastDirtyTimestamp，表示数据出现冲突，返回 404，要求应用实例重新进行 register 操作
51 return Response.status(Status.NOT_FOUND).build();
52 } else if (appInfo.getLastDirtyTimestamp() > lastDirtyTimestamp) {
53 // In the case of replication, send the current instance info in the registry for the
54 // replicating node to sync itself with this one.
55 if (isReplication) {
56 logger.debug(
57 "Time to sync, since the last dirty timestamp differs -"
58 + " ReplicationInstance id : {},Registry : {} Incoming: {} Replication: {}",
59 args);
60 // 如果本地实例的 lastDirtyTimestamp > 复制实例的 lastDirtyTimestamp，就返回 CONFLICT(409)，说明数据冲突，要求其同步自己最新的数据
61 // 注意这里将本地实例 appInfo 放入 Response entity 中了
62 return Response.status(Status.CONFLICT).entity(appInfo).build();
63 } else {
64 return Response.ok().build();
65 }
66 }
67 }
68
69 }
70 return Response.ok().build();
71 }
② 接着看 PeerReplicationResource 处理心跳的方法
- 首先就是调用了续约的方法 renewLease 进行续约
- 如果返回的状态是 OK 或者 CONFLICT，就在 resposeEntity 中返回本地实例
1 private static Builder handleHeartbeat(EurekaServerConfig config, InstanceResource resource, String lastDirtyTimestamp, String overriddenStatus, String instanceStatus) {
2 // 调用 renewLease 续约
3 Response response = resource.renewLease(REPLICATION, overriddenStatus, instanceStatus, lastDirtyTimestamp);
4 int responseStatus = response.getStatus();
5 Builder responseBuilder = new Builder().setStatusCode(responseStatus);
6
7 if ("false".equals(config.getExperimental("bugfix.934"))) {
8 if (responseStatus == Status.OK.getStatusCode() && response.getEntity() != null) {
9 responseBuilder.setResponseEntity((InstanceInfo) response.getEntity());
10 }
11 } else {
12 if ((responseStatus == Status.OK.getStatusCode() || responseStatus == Status.CONFLICT.getStatusCode())
13 && response.getEntity() != null) {
14 // 续约成功或 CONFLICT 冲突时，将本地实例 appInfo 返回到客户端
15 responseBuilder.setResponseEntity((InstanceInfo) response.getEntity());
16 }
17 }
18 return responseBuilder;
19 }
③ PeerEurekaNode 发送心跳
ReplicationTaskProcessor 收到批量任务返回结果后，会处理响应结果，对于心跳任务，可以找到，失败后就会回调 handleFailure 方法。
- 如果返回状态是 404（NOT_FOUND），就会重新注册，也是提交到队列中。通过重新注册来实现数据同步。
- 如果是其它状态（409 CONFLICT）并且开启了时间戳不一致就同步的配置，就将服务端返回的实例注册到本地，实现数据的同步。
1 public void heartbeat(final String appName, final String id,
2 final InstanceInfo info, final InstanceStatus overriddenStatus,
3 boolean primeConnection) throws Throwable {
4 if (primeConnection) {
5 // We do not care about the result for priming request.
6 replicationClient.sendHeartBeat(appName, id, info, overriddenStatus);
7 return;
8 }
9 ReplicationTask replicationTask = new InstanceReplicationTask(targetHost, Action.Heartbeat, info, overriddenStatus, false) {
10 @Override
11 public EurekaHttpResponse<InstanceInfo> execute() throws Throwable {
12 return replicationClient.sendHeartBeat(appName, id, info, overriddenStatus);
13 }
14
15 @Override
16 public void handleFailure(int statusCode, Object responseEntity) throws Throwable {
17 super.handleFailure(statusCode, responseEntity);
18 if (statusCode == 404) {
19 logger.warn("{}: missing entry.", getTaskName());
20 if (info != null) {
21 logger.warn("{}: cannot find instance id {} and hence replicating the instance with status {}",
22 getTaskName(), info.getId(), info.getStatus());
23 // 复制返回 404 时，重新注册
24 register(info);
25 }
26 } else if (config.shouldSyncWhenTimestampDiffers()) {
27 // 409（CONFLICT）
28 InstanceInfo peerInstanceInfo = (InstanceInfo) responseEntity;
29 if (peerInstanceInfo != null) {
30 syncInstancesIfTimestampDiffers(appName, id, info, peerInstanceInfo);
31 }
32 }
33 }
34 };
35 long expiryTime = System.currentTimeMillis() + getLeaseRenewalOf(info);
36 batchingDispatcher.process(taskId("heartbeat", info), replicationTask, expiryTime);
37 }
38
39 //////////////////////////////////////////
40
41 private void syncInstancesIfTimestampDiffers(String appName, String id, InstanceInfo info, InstanceInfo infoFromPeer) {
42 try {
43 if (infoFromPeer != null) {
44 if (infoFromPeer.getOverriddenStatus() != null && !InstanceStatus.UNKNOWN.equals(infoFromPeer.getOverriddenStatus())) {
45 logger.warn("Overridden Status info -id {}, mine {}, peer's {}", id, info.getOverriddenStatus(), infoFromPeer.getOverriddenStatus());
46 registry.storeOverriddenStatusIfRequired(appName, id, infoFromPeer.getOverriddenStatus());
47 }
48 // 将服务端的实例注册到本地，实现数据同步
49 registry.register(infoFromPeer, true);
50 }
51 } catch (Throwable e) {
52 logger.warn("Exception when trying to set information from peer :", e);
53 }
54 }
至此，我们就可以总结出，eureka server 通过对比 lastDirtyTimestamp 和心跳操作来实现集群数据的复制和最终同步。
前面提到的实例过期就丢弃任务这样看来就没问题，它也不保证peer节点间相互复制的所有操作都成功，eureka 采用的是最终一致性，它是通过心跳的方式实现集群数据的最终修复和同步，只是集群间可能会同步延迟。
9、集群同步总结
下面总结下 eureka-server 集群节点间的同步：
- 首先 eureka-server 集群采用的是 Peer To Peer 的模式，即对等复制，各个 server 不分主从，每个 server 都可以接收写请求，然后互相之间进行数据更新同步。
-
数据同步采用了多层任务队列+批量处理的机制：
- eureka-server 接收到客户端请求（注册、下线、续约）后都会调用集群 PeerEurekaNode 进行操作的同步
- PeerEurekaNode 将操作封装成 InstanceReplicationTask 实例复制任务，并用批量分发器 batchingDispatcher（TaskDispatcher）来分发处理
- batchingDispatcher 内部则将任务交给接收者执行器 AcceptorExecutor 处理，任务首先进入到 AcceptorExecutor 内的接收者队列 acceptorQueue 中
- AcceptorExecutor 有个后台工作线程（AcceptorRunner）不断轮询，将接收者队列 acceptorQueue 和 重处理队列 reprocessQueue 中的任务转移到处理中队列中（processingOrder + pendingTasks）
- 接着将处理中队列中的任务打包，一次最多 250 个任务，然后放到批量工作队列 batchWorkQueue。如果处理中队列中还有任务，就将任务放到单项任务队列 singleItemWorkQueue
- 任务都打包好了，任务执行器 TaskExecutors 内分别有批量任务处理器（BatchWorkerRunnable）和单项任务处理器（SingleTaskWorkerRunnable）来处理 batchWorkQueue 和 singleItemWorkQueue 中的任务
- 处理器会利用任务复制处理器（ReplicationTaskProcessor）来提交任务，批量任务会提交给 server 节点的批量接口（peerreplication/batch/），单项任务则提交到对应的操作接口
- 任务提交如果阻塞或者网络失败就会被放入重处理队列 reprocessQueue，然后再次被 AcceptorRunner 轮询处理，不过过期（超过90秒）的任务会被丢弃掉
-
其它 eureka-server 同步：
- 其它 eureka-server 接收到批量复制请求后，会轮询批量任务列表，根据不同的操作类型（Register、Heartbeat、Cancel 等）分别调用 Resource 的接口进行处理
- 如果是续约操作，会判断复制实例的 lastDirtyTimestamp 与本地实例的 lastDirtyTimestamp，如果是一致的，就任务数据一致
- 如果复制实例的 lastDirtyTimestamp > 本地实例的 lastDirtyTimestamp，则复制实例的数据是最新的，返回 404（NOT_FOUND） 要求客户端重新发送一个注册操作过来
- 如果复制实例的 lastDirtyTimestamp < 本地实例的 lastDirtyTimestamp，则本地实例的数据是最新的，返回 409（CONFLICT）和本地实例，客户端用返回来的实例覆盖本地的实例
下面再用一张图总结集群同步：
十二、SpringCloud Eureka
到这里，对 Eureka 核心源码的研究就差不多了，这节来看下 Spring cloud eureka。Spring cloud eureka 提供了服务端的依赖 spring-cloud-starter-netflix-eureka-server 和客户端的依赖 spring-cloud-starter-netflix-eureka-client，这两个依赖包本身是比较简单的，只是对 netflix 的 eureka-server 和 eureka-client 的封装，它通过一些注解和配置类将 eureka 整合到 springboot 技术栈中，便于使用。
1、spring-cloud-starter-netflix-eureka-server
看 spring-cloud-starter-netflix-eureka-server 我们从 @EnableEurekaServer 这个注解来看，因为我们的注册中心是基于 springboot 的，在启动类上加上了 @EnableEurekaServer 注解就启用了 eureka-server 注册中心。
① Eureka Server 自动化配置
看这个注解的定义，从注释中可以了解到，这个注解会激活 EurekaServerAutoConfiguration 的自动化配置类。
1 /**
2 * Annotation to activate Eureka Server related configuration.
3 * {@link EurekaServerAutoConfiguration}
4 *
5 * @author Dave Syer
6 * @author Biju Kunjummen
7 */
8 @Target(ElementType.TYPE)
9 @Retention(RetentionPolicy.RUNTIME)
10 @Documented
11 @Import(EurekaServerMarkerConfiguration.class)
12 public @interface EnableEurekaServer {
13
14 }
看 EurekaServerAutoConfiguration 这个类，可以发现 springcloud 几乎是将 com.netflix.eureka.EurekaBootStrap 中初始化组件的代码拷贝到了 EurekaServerAutoConfiguration，然后以 springboot 创建 bean 的方式来创建相关的组件。
1 @Configuration(proxyBeanMethods = false)
2 @Import(EurekaServerInitializerConfiguration.class)
3 @ConditionalOnBean(EurekaServerMarkerConfiguration.Marker.class)
4 @EnableConfigurationProperties({ EurekaDashboardProperties.class,
5 InstanceRegistryProperties.class })
6 @PropertySource("classpath:/eureka/server.properties")
7 public class EurekaServerAutoConfiguration implements WebMvcConfigurer {
8
9 /**
10 * List of packages containing Jersey resources required by the Eureka server.
11 */
12 private static final String[] EUREKA_PACKAGES = new String[] {
13 "com.netflix.discovery", "com.netflix.eureka" };
14
15 @Autowired
16 private ApplicationInfoManager applicationInfoManager;
17
18 @Autowired
19 private EurekaServerConfig eurekaServerConfig;
20
21 @Autowired
22 private EurekaClientConfig eurekaClientConfig;
23
24 @Autowired
25 private EurekaClient eurekaClient;
26
27 @Autowired
28 private InstanceRegistryProperties instanceRegistryProperties;
29
30 /**
31 * A {@link CloudJacksonJson} instance.
32 */
33 public static final CloudJacksonJson JACKSON_JSON = new CloudJacksonJson();
34
35 @Bean
36 public HasFeatures eurekaServerFeature() {
37 return HasFeatures.namedFeature("Eureka Server",
38 EurekaServerAutoConfiguration.class);
39 }
40
41 @Bean
42 @ConditionalOnProperty(prefix = "eureka.dashboard", name = "enabled",
43 matchIfMissing = true)
44 public EurekaController eurekaController() {
45 return new EurekaController(this.applicationInfoManager);
46 }
47
48 static {
49 CodecWrappers.registerWrapper(JACKSON_JSON);
50 EurekaJacksonCodec.setInstance(JACKSON_JSON.getCodec());
51 }
52
53 @Bean
54 public ServerCodecs serverCodecs() {
55 return new CloudServerCodecs(this.eurekaServerConfig);
56 }
57
58 private static CodecWrapper getFullJson(EurekaServerConfig serverConfig) {
59 CodecWrapper codec = CodecWrappers.getCodec(serverConfig.getJsonCodecName());
60 return codec == null ? CodecWrappers.getCodec(JACKSON_JSON.codecName()) : codec;
61 }
62
63 private static CodecWrapper getFullXml(EurekaServerConfig serverConfig) {
64 CodecWrapper codec = CodecWrappers.getCodec(serverConfig.getXmlCodecName());
65 return codec == null ? CodecWrappers.getCodec(CodecWrappers.XStreamXml.class)
66 : codec;
67 }
68
69 @Bean
70 @ConditionalOnMissingBean
71 public ReplicationClientAdditionalFilters replicationClientAdditionalFilters() {
72 return new ReplicationClientAdditionalFilters(Collections.emptySet());
73 }
74
75 @Bean
76 public PeerAwareInstanceRegistry peerAwareInstanceRegistry(
77 ServerCodecs serverCodecs) {
78 this.eurekaClient.getApplications(); // force initialization
79 return new InstanceRegistry(this.eurekaServerConfig, this.eurekaClientConfig,
80 serverCodecs, this.eurekaClient,
81 this.instanceRegistryProperties.getExpectedNumberOfClientsSendingRenews(),
82 this.instanceRegistryProperties.getDefaultOpenForTrafficCount());
83 }
84
85 @Bean
86 @ConditionalOnMissingBean
87 public PeerEurekaNodes peerEurekaNodes(PeerAwareInstanceRegistry registry,
88 ServerCodecs serverCodecs,
89 ReplicationClientAdditionalFilters replicationClientAdditionalFilters) {
90 return new RefreshablePeerEurekaNodes(registry, this.eurekaServerConfig,
91 this.eurekaClientConfig, serverCodecs, this.applicationInfoManager,
92 replicationClientAdditionalFilters);
93 }
94
95 @Bean
96 @ConditionalOnMissingBean
97 public EurekaServerContext eurekaServerContext(ServerCodecs serverCodecs,
98 PeerAwareInstanceRegistry registry, PeerEurekaNodes peerEurekaNodes) {
99 return new DefaultEurekaServerContext(this.eurekaServerConfig, serverCodecs,
100 registry, peerEurekaNodes, this.applicationInfoManager);
101 }
102
103 @Bean
104 public EurekaServerBootstrap eurekaServerBootstrap(PeerAwareInstanceRegistry registry,
105 EurekaServerContext serverContext) {
106 return new EurekaServerBootstrap(this.applicationInfoManager,
107 this.eurekaClientConfig, this.eurekaServerConfig, registry,
108 serverContext);
109 }
110
111 /**
112 * Register the Jersey filter.
113 * @param eurekaJerseyApp an {@link Application} for the filter to be registered
114 * @return a jersey {@link FilterRegistrationBean}
115 */
116 @Bean
117 public FilterRegistrationBean<?> jerseyFilterRegistration(
118 javax.ws.rs.core.Application eurekaJerseyApp) {
119 FilterRegistrationBean<Filter> bean = new FilterRegistrationBean<Filter>();
120 bean.setFilter(new ServletContainer(eurekaJerseyApp));
121 bean.setOrder(Ordered.LOWEST_PRECEDENCE);
122 bean.setUrlPatterns(
123 Collections.singletonList(EurekaConstants.DEFAULT_PREFIX + "/*"));
124
125 return bean;
126 }
127
128 /**
129 * Construct a Jersey {@link javax.ws.rs.core.Application} with all the resources
130 * required by the Eureka server.
131 * @param environment an {@link Environment} instance to retrieve classpath resources
132 * @param resourceLoader a {@link ResourceLoader} instance to get classloader from
133 * @return created {@link Application} object
134 */
135 @Bean
136 public javax.ws.rs.core.Application jerseyApplication(Environment environment,
137 ResourceLoader resourceLoader) {
138
139 ClassPathScanningCandidateComponentProvider provider = new ClassPathScanningCandidateComponentProvider(
140 false, environment);
141
142 // Filter to include only classes that have a particular annotation.
143 //
144 provider.addIncludeFilter(new AnnotationTypeFilter(Path.class));
145 provider.addIncludeFilter(new AnnotationTypeFilter(Provider.class));
146
147 // Find classes in Eureka packages (or subpackages)
148 //
149 Set<Class<?>> classes = new HashSet<>();
150 for (String basePackage : EUREKA_PACKAGES) {
151 Set<BeanDefinition> beans = provider.findCandidateComponents(basePackage);
152 for (BeanDefinition bd : beans) {
153 Class<?> cls = ClassUtils.resolveClassName(bd.getBeanClassName(),
154 resourceLoader.getClassLoader());
155 classes.add(cls);
156 }
157 }
158
159 // Construct the Jersey ResourceConfig
160 Map<String, Object> propsAndFeatures = new HashMap<>();
161 propsAndFeatures.put(
162 // Skip static content used by the webapp
163 ServletContainer.PROPERTY_WEB_PAGE_CONTENT_REGEX,
164 EurekaConstants.DEFAULT_PREFIX + "/(fonts|images|css|js)/.*");
165
166 DefaultResourceConfig rc = new DefaultResourceConfig(classes);
167 rc.setPropertiesAndFeatures(propsAndFeatures);
168
169 return rc;
170 }
171
172 @Bean
173 @ConditionalOnBean(name = "httpTraceFilter")
174 public FilterRegistrationBean<?> traceFilterRegistration(
175 @Qualifier("httpTraceFilter") Filter filter) {
176 FilterRegistrationBean<Filter> bean = new FilterRegistrationBean<Filter>();
177 bean.setFilter(filter);
178 bean.setOrder(Ordered.LOWEST_PRECEDENCE - 10);
179 return bean;
180 }
181
182 @Configuration(proxyBeanMethods = false)
183 protected static class EurekaServerConfigBeanConfiguration {
184
185 @Bean
186 @ConditionalOnMissingBean
187 public EurekaServerConfig eurekaServerConfig(EurekaClientConfig clientConfig) {
188 EurekaServerConfigBean server = new EurekaServerConfigBean();
189 if (clientConfig.shouldRegisterWithEureka()) {
190 // Set a sensible default if we are supposed to replicate
191 server.setRegistrySyncRetries(5);
192 }
193 return server;
194 }
195
196 }
197
198 /**
199 * {@link PeerEurekaNodes} which updates peers when /refresh is invoked. Peers are
200 * updated only if <code>eureka.client.use-dns-for-fetching-service-urls</code> is
201 * <code>false</code> and one of following properties have changed.
202 * <p>
203 * </p>
204 * <ul>
205 * <li><code>eureka.client.availability-zones</code></li>
206 * <li><code>eureka.client.region</code></li>
207 * <li><code>eureka.client.service-url.<zone></code></li>
208 * </ul>
209 */
210 static class RefreshablePeerEurekaNodes extends PeerEurekaNodes
211 implements ApplicationListener<EnvironmentChangeEvent> {
212
213 private ReplicationClientAdditionalFilters replicationClientAdditionalFilters;
214
215 RefreshablePeerEurekaNodes(final PeerAwareInstanceRegistry registry,
216 final EurekaServerConfig serverConfig,
217 final EurekaClientConfig clientConfig, final ServerCodecs serverCodecs,
218 final ApplicationInfoManager applicationInfoManager,
219 final ReplicationClientAdditionalFilters replicationClientAdditionalFilters) {
220 super(registry, serverConfig, clientConfig, serverCodecs,
221 applicationInfoManager);
222 this.replicationClientAdditionalFilters = replicationClientAdditionalFilters;
223 }
224
225 @Override
226 protected PeerEurekaNode createPeerEurekaNode(String peerEurekaNodeUrl) {
227 JerseyReplicationClient replicationClient = JerseyReplicationClient
228 .createReplicationClient(serverConfig, serverCodecs,
229 peerEurekaNodeUrl);
230
231 this.replicationClientAdditionalFilters.getFilters()
232 .forEach(replicationClient::addReplicationClientFilter);
233
234 String targetHost = hostFromUrl(peerEurekaNodeUrl);
235 if (targetHost == null) {
236 targetHost = "host";
237 }
238 return new PeerEurekaNode(registry, targetHost, peerEurekaNodeUrl,
239 replicationClient, serverConfig);
240 }
241
242 @Override
243 public void onApplicationEvent(final EnvironmentChangeEvent event) {
244 if (shouldUpdate(event.getKeys())) {
245 updatePeerEurekaNodes(resolvePeerUrls());
246 }
247 }
248
249 /*
250 * Check whether specific properties have changed.
251 */
252 protected boolean shouldUpdate(final Set<String> changedKeys) {
253 assert changedKeys != null;
254
255 // if eureka.client.use-dns-for-fetching-service-urls is true, then
256 // service-url will not be fetched from environment.
257 if (this.clientConfig.shouldUseDnsForFetchingServiceUrls()) {
258 return false;
259 }
260
261 if (changedKeys.contains("eureka.client.region")) {
262 return true;
263 }
264
265 for (final String key : changedKeys) {
266 // property keys are not expected to be null.
267 if (key.startsWith("eureka.client.service-url.")
268 || key.startsWith("eureka.client.availability-zones.")) {
269 return true;
270 }
271 }
272 return false;
273 }
274
275 }
276
277 class CloudServerCodecs extends DefaultServerCodecs {
278
279 CloudServerCodecs(EurekaServerConfig serverConfig) {
280 super(getFullJson(serverConfig),
281 CodecWrappers.getCodec(CodecWrappers.JacksonJsonMini.class),
282 getFullXml(serverConfig),
283 CodecWrappers.getCodec(CodecWrappers.JacksonXmlMini.class));
284 }
285
286 }
287
288 }
View Code
但是要注意，springcloud 中，感知集群的注册表组件是 InstanceRegistry，集群 PeerEurekaNodes 组件是 RefreshablePeerEurekaNodes。
② Springboot 方式的配置
EurekaServerAutoConfiguration 中注入了 EurekaServerConfig、EurekaClientConfig，在 Netflix 中，EurekaServerConfig 默认读取的是 eureka-server.properties 配置文件，EurekaClientConfig 默认读取的是 eureka-client.properties 配置文件。而在 springcloud 中，它们的实现类为 EurekaServerConfigBean、EurekaClientConfigBean，可以看到，就是基于 springboot 的配置方式来的了，读取的是 application.yml 配置文件中 eureka 的配置了，并且每个配置也提供了默认值。
例如 EurekaServerConfigBean：
1 @ConfigurationProperties(EurekaServerConfigBean.PREFIX)
2 public class EurekaServerConfigBean implements EurekaServerConfig {
3
4 /**
5 * Eureka server configuration properties prefix.
6 */
7 public static final String PREFIX = "eureka.server";
8
9 private static final int MINUTES = 60 * 1000;
10
11 @Autowired(required = false)
12 PropertyResolver propertyResolver;
13
14 private String aWSAccessId;
15
16 private String aWSSecretKey;
17
18 private int eIPBindRebindRetries = 3;
19
20 private int eIPBindingRetryIntervalMs = 5 * MINUTES;
21
22 private int eIPBindingRetryIntervalMsWhenUnbound = 1 * MINUTES;
23
24 private boolean enableSelfPreservation = true;
25
26 private double renewalPercentThreshold = 0.85;
27
28 private int renewalThresholdUpdateIntervalMs = 15 * MINUTES;
29
30 private int peerEurekaNodesUpdateIntervalMs = 10 * MINUTES;
31
32 private int numberOfReplicationRetries = 5;
33
34 private int peerEurekaStatusRefreshTimeIntervalMs = 30 * 1000;
35
36 private int waitTimeInMsWhenSyncEmpty = 5 * MINUTES;
37
38 private int peerNodeConnectTimeoutMs = 200;
39
40 private int peerNodeReadTimeoutMs = 200;
41
42 private int peerNodeTotalConnections = 1000;
43
44 private int peerNodeTotalConnectionsPerHost = 500;
45
46 private int peerNodeConnectionIdleTimeoutSeconds = 30;
47
48 private long retentionTimeInMSInDeltaQueue = 3 * MINUTES;
49
50 private long deltaRetentionTimerIntervalInMs = 30 * 1000;
51
52 private long evictionIntervalTimerInMs = 60 * 1000;
53
54 private int aSGQueryTimeoutMs = 300;
55
56 private long aSGUpdateIntervalMs = 5 * MINUTES;
57
58 private long aSGCacheExpiryTimeoutMs = 10 * MINUTES; // defaults to longer than the
59
60 // asg update interval
61
62 private long responseCacheAutoExpirationInSeconds = 180;
63
64 private long responseCacheUpdateIntervalMs = 30 * 1000;
65
66 private boolean useReadOnlyResponseCache = true;
67
68 private boolean disableDelta;
69
70 private long maxIdleThreadInMinutesAgeForStatusReplication = 10;
71
72 private int minThreadsForStatusReplication = 1;
73
74 private int maxThreadsForStatusReplication = 1;
75
76 private int maxElementsInStatusReplicationPool = 10000;
77
78 private boolean syncWhenTimestampDiffers = true;
79
80 private int registrySyncRetries = 0;
81
82 private long registrySyncRetryWaitMs = 30 * 1000;
83
84 private int maxElementsInPeerReplicationPool = 10000;
85
86 private long maxIdleThreadAgeInMinutesForPeerReplication = 15;
87
88 private int minThreadsForPeerReplication = 5;
89
90 private int maxThreadsForPeerReplication = 20;
91
92 private int maxTimeForReplication = 30000;
93
94 private boolean primeAwsReplicaConnections = true;
95
96 private boolean disableDeltaForRemoteRegions;
97
98 private int remoteRegionConnectTimeoutMs = 1000;
99
100 private int remoteRegionReadTimeoutMs = 1000;
101
102 private int remoteRegionTotalConnections = 1000;
103
104 private int remoteRegionTotalConnectionsPerHost = 500;
105
106 private int remoteRegionConnectionIdleTimeoutSeconds = 30;
107
108 private boolean gZipContentFromRemoteRegion = true;
109
110 private Map<String, String> remoteRegionUrlsWithName = new HashMap<>();
111
112 private String[] remoteRegionUrls;
113
114 private Map<String, Set<String>> remoteRegionAppWhitelist = new HashMap<>();
115
116 private int remoteRegionRegistryFetchInterval = 30;
117
118 private int remoteRegionFetchThreadPoolSize = 20;
119
120 private String remoteRegionTrustStore = "";
121
122 private String remoteRegionTrustStorePassword = "changeit";
123
124 private boolean disableTransparentFallbackToOtherRegion;
125
126 private boolean batchReplication;
127
128 private boolean rateLimiterEnabled = false;
129
130 private boolean rateLimiterThrottleStandardClients = false;
131
132 private Set<String> rateLimiterPrivilegedClients = Collections.emptySet();
133
134 private int rateLimiterBurstSize = 10;
135
136 private int rateLimiterRegistryFetchAverageRate = 500;
137
138 private int rateLimiterFullFetchAverageRate = 100;
139
140 private boolean logIdentityHeaders = true;
141
142 private String listAutoScalingGroupsRoleName = "ListAutoScalingGroups";
143
144 private boolean enableReplicatedRequestCompression = false;
145
146 private String jsonCodecName;
147
148 private String xmlCodecName;
149
150 private int route53BindRebindRetries = 3;
151
152 private int route53BindingRetryIntervalMs = 5 * MINUTES;
153
154 private long route53DomainTTL = 30;
155
156 private AwsBindingStrategy bindingStrategy = AwsBindingStrategy.EIP;
157
158 private int minAvailableInstancesForPeerReplication = -1;
159
160 private int initialCapacityOfResponseCache = 1000;
161
162 private int expectedClientRenewalIntervalSeconds = 30;
163
164 private boolean useAwsAsgApi = true;
165
166 private String myUrl;
167
168 @Override
169 public boolean shouldEnableSelfPreservation() {
170 return this.enableSelfPreservation;
171 }
172
173 @Override
174 public boolean shouldDisableDelta() {
175 return this.disableDelta;
176 }
177
178 @Override
179 public boolean shouldSyncWhenTimestampDiffers() {
180 return this.syncWhenTimestampDiffers;
181 }
182
183 @Override
184 public boolean shouldPrimeAwsReplicaConnections() {
185 return this.primeAwsReplicaConnections;
186 }
187
188 @Override
189 public boolean shouldDisableDeltaForRemoteRegions() {
190 return this.disableDeltaForRemoteRegions;
191 }
192
193 @Override
194 public boolean shouldGZipContentFromRemoteRegion() {
195 return this.gZipContentFromRemoteRegion;
196 }
197
198 @Override
199 public Set<String> getRemoteRegionAppWhitelist(String regionName) {
200 return this.remoteRegionAppWhitelist
201 .get(regionName == null ? "global" : regionName.trim().toLowerCase());
202 }
203
204 @Override
205 public boolean disableTransparentFallbackToOtherRegion() {
206 return this.disableTransparentFallbackToOtherRegion;
207 }
208
209 @Override
210 public boolean shouldBatchReplication() {
211 return this.batchReplication;
212 }
213
214 @Override
215 public String getMyUrl() {
216 return this.myUrl;
217 }
218
219 public void setMyUrl(String myUrl) {
220 this.myUrl = myUrl;
221 }
222
223 @Override
224 public boolean shouldLogIdentityHeaders() {
225 return this.logIdentityHeaders;
226 }
227
228 @Override
229 public String getJsonCodecName() {
230 return this.jsonCodecName;
231 }
232
233 @Override
234 public String getXmlCodecName() {
235 return this.xmlCodecName;
236 }
237
238 @Override
239 public boolean shouldUseReadOnlyResponseCache() {
240 return this.useReadOnlyResponseCache;
241 }
242
243 @Override
244 public boolean shouldEnableReplicatedRequestCompression() {
245 return this.enableReplicatedRequestCompression;
246 }
247
248 @Override
249 public String getExperimental(String name) {
250 if (this.propertyResolver != null) {
251 return this.propertyResolver.getProperty(PREFIX + ".experimental." + name,
252 String.class, null);
253 }
254 return null;
255 }
256
257 @Override
258 public int getInitialCapacityOfResponseCache() {
259 return this.initialCapacityOfResponseCache;
260 }
261
262 public void setInitialCapacityOfResponseCache(int initialCapacityOfResponseCache) {
263 this.initialCapacityOfResponseCache = initialCapacityOfResponseCache;
264 }
265
266 @Override
267 public int getHealthStatusMinNumberOfAvailablePeers() {
268 return this.minAvailableInstancesForPeerReplication;
269 }
270
271 public PropertyResolver getPropertyResolver() {
272 return propertyResolver;
273 }
274
275 public void setPropertyResolver(PropertyResolver propertyResolver) {
276 this.propertyResolver = propertyResolver;
277 }
278
279 public String getAWSAccessId() {
280 return aWSAccessId;
281 }
282
283 public void setAWSAccessId(String aWSAccessId) {
284 this.aWSAccessId = aWSAccessId;
285 }
286
287 public String getAWSSecretKey() {
288 return aWSSecretKey;
289 }
290
291 public void setAWSSecretKey(String aWSSecretKey) {
292 this.aWSSecretKey = aWSSecretKey;
293 }
294
295 public int getEIPBindRebindRetries() {
296 return eIPBindRebindRetries;
297 }
298
299 public void setEIPBindRebindRetries(int eIPBindRebindRetries) {
300 this.eIPBindRebindRetries = eIPBindRebindRetries;
301 }
302
303 public int getEIPBindingRetryIntervalMs() {
304 return eIPBindingRetryIntervalMs;
305 }
306
307 public void setEIPBindingRetryIntervalMs(int eIPBindingRetryIntervalMs) {
308 this.eIPBindingRetryIntervalMs = eIPBindingRetryIntervalMs;
309 }
310
311 public int getEIPBindingRetryIntervalMsWhenUnbound() {
312 return eIPBindingRetryIntervalMsWhenUnbound;
313 }
314
315 public void setEIPBindingRetryIntervalMsWhenUnbound(
316 int eIPBindingRetryIntervalMsWhenUnbound) {
317 this.eIPBindingRetryIntervalMsWhenUnbound = eIPBindingRetryIntervalMsWhenUnbound;
318 }
319
320 public boolean isEnableSelfPreservation() {
321 return enableSelfPreservation;
322 }
323
324 public void setEnableSelfPreservation(boolean enableSelfPreservation) {
325 this.enableSelfPreservation = enableSelfPreservation;
326 }
327
328 @Override
329 public double getRenewalPercentThreshold() {
330 return renewalPercentThreshold;
331 }
332
333 public void setRenewalPercentThreshold(double renewalPercentThreshold) {
334 this.renewalPercentThreshold = renewalPercentThreshold;
335 }
336
337 @Override
338 public int getRenewalThresholdUpdateIntervalMs() {
339 return renewalThresholdUpdateIntervalMs;
340 }
341
342 @Override
343 public int getExpectedClientRenewalIntervalSeconds() {
344 return this.expectedClientRenewalIntervalSeconds;
345 }
346
347 public void setExpectedClientRenewalIntervalSeconds(
348 int expectedClientRenewalIntervalSeconds) {
349 this.expectedClientRenewalIntervalSeconds = expectedClientRenewalIntervalSeconds;
350 }
351
352 public void setRenewalThresholdUpdateIntervalMs(
353 int renewalThresholdUpdateIntervalMs) {
354 this.renewalThresholdUpdateIntervalMs = renewalThresholdUpdateIntervalMs;
355 }
356
357 @Override
358 public int getPeerEurekaNodesUpdateIntervalMs() {
359 return peerEurekaNodesUpdateIntervalMs;
360 }
361
362 public void setPeerEurekaNodesUpdateIntervalMs(int peerEurekaNodesUpdateIntervalMs) {
363 this.peerEurekaNodesUpdateIntervalMs = peerEurekaNodesUpdateIntervalMs;
364 }
365
366 @Override
367 public int getNumberOfReplicationRetries() {
368 return numberOfReplicationRetries;
369 }
370
371 public void setNumberOfReplicationRetries(int numberOfReplicationRetries) {
372 this.numberOfReplicationRetries = numberOfReplicationRetries;
373 }
374
375 @Override
376 public int getPeerEurekaStatusRefreshTimeIntervalMs() {
377 return peerEurekaStatusRefreshTimeIntervalMs;
378 }
379
380 public void setPeerEurekaStatusRefreshTimeIntervalMs(
381 int peerEurekaStatusRefreshTimeIntervalMs) {
382 this.peerEurekaStatusRefreshTimeIntervalMs = peerEurekaStatusRefreshTimeIntervalMs;
383 }
384
385 @Override
386 public int getWaitTimeInMsWhenSyncEmpty() {
387 return waitTimeInMsWhenSyncEmpty;
388 }
389
390 public void setWaitTimeInMsWhenSyncEmpty(int waitTimeInMsWhenSyncEmpty) {
391 this.waitTimeInMsWhenSyncEmpty = waitTimeInMsWhenSyncEmpty;
392 }
393
394 @Override
395 public int getPeerNodeConnectTimeoutMs() {
396 return peerNodeConnectTimeoutMs;
397 }
398
399 public void setPeerNodeConnectTimeoutMs(int peerNodeConnectTimeoutMs) {
400 this.peerNodeConnectTimeoutMs = peerNodeConnectTimeoutMs;
401 }
402
403 @Override
404 public int getPeerNodeReadTimeoutMs() {
405 return peerNodeReadTimeoutMs;
406 }
407
408 public void setPeerNodeReadTimeoutMs(int peerNodeReadTimeoutMs) {
409 this.peerNodeReadTimeoutMs = peerNodeReadTimeoutMs;
410 }
411
412 @Override
413 public int getPeerNodeTotalConnections() {
414 return peerNodeTotalConnections;
415 }
416
417 public void setPeerNodeTotalConnections(int peerNodeTotalConnections) {
418 this.peerNodeTotalConnections = peerNodeTotalConnections;
419 }
420
421 @Override
422 public int getPeerNodeTotalConnectionsPerHost() {
423 return peerNodeTotalConnectionsPerHost;
424 }
425
426 public void setPeerNodeTotalConnectionsPerHost(int peerNodeTotalConnectionsPerHost) {
427 this.peerNodeTotalConnectionsPerHost = peerNodeTotalConnectionsPerHost;
428 }
429
430 @Override
431 public int getPeerNodeConnectionIdleTimeoutSeconds() {
432 return peerNodeConnectionIdleTimeoutSeconds;
433 }
434
435 public void setPeerNodeConnectionIdleTimeoutSeconds(
436 int peerNodeConnectionIdleTimeoutSeconds) {
437 this.peerNodeConnectionIdleTimeoutSeconds = peerNodeConnectionIdleTimeoutSeconds;
438 }
439
440 @Override
441 public long getRetentionTimeInMSInDeltaQueue() {
442 return retentionTimeInMSInDeltaQueue;
443 }
444
445 public void setRetentionTimeInMSInDeltaQueue(long retentionTimeInMSInDeltaQueue) {
446 this.retentionTimeInMSInDeltaQueue = retentionTimeInMSInDeltaQueue;
447 }
448
449 @Override
450 public long getDeltaRetentionTimerIntervalInMs() {
451 return deltaRetentionTimerIntervalInMs;
452 }
453
454 public void setDeltaRetentionTimerIntervalInMs(long deltaRetentionTimerIntervalInMs) {
455 this.deltaRetentionTimerIntervalInMs = deltaRetentionTimerIntervalInMs;
456 }
457
458 @Override
459 public long getEvictionIntervalTimerInMs() {
460 return evictionIntervalTimerInMs;
461 }
462
463 @Override
464 public boolean shouldUseAwsAsgApi() {
465 return this.useAwsAsgApi;
466 }
467
468 public void setUseAwsAsgApi(boolean useAwsAsgApi) {
469 this.useAwsAsgApi = useAwsAsgApi;
470 }
471
472 public void setEvictionIntervalTimerInMs(long evictionIntervalTimerInMs) {
473 this.evictionIntervalTimerInMs = evictionIntervalTimerInMs;
474 }
475
476 public int getASGQueryTimeoutMs() {
477 return aSGQueryTimeoutMs;
478 }
479
480 public void setASGQueryTimeoutMs(int aSGQueryTimeoutMs) {
481 this.aSGQueryTimeoutMs = aSGQueryTimeoutMs;
482 }
483
484 public long getASGUpdateIntervalMs() {
485 return aSGUpdateIntervalMs;
486 }
487
488 public void setASGUpdateIntervalMs(long aSGUpdateIntervalMs) {
489 this.aSGUpdateIntervalMs = aSGUpdateIntervalMs;
490 }
491
492 public long getASGCacheExpiryTimeoutMs() {
493 return aSGCacheExpiryTimeoutMs;
494 }
495
496 public void setASGCacheExpiryTimeoutMs(long aSGCacheExpiryTimeoutMs) {
497 this.aSGCacheExpiryTimeoutMs = aSGCacheExpiryTimeoutMs;
498 }
499
500 @Override
501 public long getResponseCacheAutoExpirationInSeconds() {
502 return responseCacheAutoExpirationInSeconds;
503 }
504
505 public void setResponseCacheAutoExpirationInSeconds(
506 long responseCacheAutoExpirationInSeconds) {
507 this.responseCacheAutoExpirationInSeconds = responseCacheAutoExpirationInSeconds;
508 }
509
510 @Override
511 public long getResponseCacheUpdateIntervalMs() {
512 return responseCacheUpdateIntervalMs;
513 }
514
515 public void setResponseCacheUpdateIntervalMs(long responseCacheUpdateIntervalMs) {
516 this.responseCacheUpdateIntervalMs = responseCacheUpdateIntervalMs;
517 }
518
519 public boolean isUseReadOnlyResponseCache() {
520 return useReadOnlyResponseCache;
521 }
522
523 public void setUseReadOnlyResponseCache(boolean useReadOnlyResponseCache) {
524 this.useReadOnlyResponseCache = useReadOnlyResponseCache;
525 }
526
527 public boolean isDisableDelta() {
528 return disableDelta;
529 }
530
531 public void setDisableDelta(boolean disableDelta) {
532 this.disableDelta = disableDelta;
533 }
534
535 @Override
536 public long getMaxIdleThreadInMinutesAgeForStatusReplication() {
537 return maxIdleThreadInMinutesAgeForStatusReplication;
538 }
539
540 public void setMaxIdleThreadInMinutesAgeForStatusReplication(
541 long maxIdleThreadInMinutesAgeForStatusReplication) {
542 this.maxIdleThreadInMinutesAgeForStatusReplication = maxIdleThreadInMinutesAgeForStatusReplication;
543 }
544
545 @Override
546 public int getMinThreadsForStatusReplication() {
547 return minThreadsForStatusReplication;
548 }
549
550 public void setMinThreadsForStatusReplication(int minThreadsForStatusReplication) {
551 this.minThreadsForStatusReplication = minThreadsForStatusReplication;
552 }
553
554 @Override
555 public int getMaxThreadsForStatusReplication() {
556 return maxThreadsForStatusReplication;
557 }
558
559 public void setMaxThreadsForStatusReplication(int maxThreadsForStatusReplication) {
560 this.maxThreadsForStatusReplication = maxThreadsForStatusReplication;
561 }
562
563 @Override
564 public int getMaxElementsInStatusReplicationPool() {
565 return maxElementsInStatusReplicationPool;
566 }
567
568 public void setMaxElementsInStatusReplicationPool(
569 int maxElementsInStatusReplicationPool) {
570 this.maxElementsInStatusReplicationPool = maxElementsInStatusReplicationPool;
571 }
572
573 public boolean isSyncWhenTimestampDiffers() {
574 return syncWhenTimestampDiffers;
575 }
576
577 public void setSyncWhenTimestampDiffers(boolean syncWhenTimestampDiffers) {
578 this.syncWhenTimestampDiffers = syncWhenTimestampDiffers;
579 }
580
581 @Override
582 public int getRegistrySyncRetries() {
583 return registrySyncRetries;
584 }
585
586 public void setRegistrySyncRetries(int registrySyncRetries) {
587 this.registrySyncRetries = registrySyncRetries;
588 }
589
590 @Override
591 public long getRegistrySyncRetryWaitMs() {
592 return registrySyncRetryWaitMs;
593 }
594
595 public void setRegistrySyncRetryWaitMs(long registrySyncRetryWaitMs) {
596 this.registrySyncRetryWaitMs = registrySyncRetryWaitMs;
597 }
598
599 @Override
600 public int getMaxElementsInPeerReplicationPool() {
601 return maxElementsInPeerReplicationPool;
602 }
603
604 public void setMaxElementsInPeerReplicationPool(
605 int maxElementsInPeerReplicationPool) {
606 this.maxElementsInPeerReplicationPool = maxElementsInPeerReplicationPool;
607 }
608
609 @Override
610 public long getMaxIdleThreadAgeInMinutesForPeerReplication() {
611 return maxIdleThreadAgeInMinutesForPeerReplication;
612 }
613
614 public void setMaxIdleThreadAgeInMinutesForPeerReplication(
615 long maxIdleThreadAgeInMinutesForPeerReplication) {
616 this.maxIdleThreadAgeInMinutesForPeerReplication = maxIdleThreadAgeInMinutesForPeerReplication;
617 }
618
619 @Override
620 public int getMinThreadsForPeerReplication() {
621 return minThreadsForPeerReplication;
622 }
623
624 public void setMinThreadsForPeerReplication(int minThreadsForPeerReplication) {
625 this.minThreadsForPeerReplication = minThreadsForPeerReplication;
626 }
627
628 @Override
629 public int getMaxThreadsForPeerReplication() {
630 return maxThreadsForPeerReplication;
631 }
632
633 public void setMaxThreadsForPeerReplication(int maxThreadsForPeerReplication) {
634 this.maxThreadsForPeerReplication = maxThreadsForPeerReplication;
635 }
636
637 @Override
638 public int getMaxTimeForReplication() {
639 return maxTimeForReplication;
640 }
641
642 public void setMaxTimeForReplication(int maxTimeForReplication) {
643 this.maxTimeForReplication = maxTimeForReplication;
644 }
645
646 public boolean isPrimeAwsReplicaConnections() {
647 return primeAwsReplicaConnections;
648 }
649
650 public void setPrimeAwsReplicaConnections(boolean primeAwsReplicaConnections) {
651 this.primeAwsReplicaConnections = primeAwsReplicaConnections;
652 }
653
654 public boolean isDisableDeltaForRemoteRegions() {
655 return disableDeltaForRemoteRegions;
656 }
657
658 public void setDisableDeltaForRemoteRegions(boolean disableDeltaForRemoteRegions) {
659 this.disableDeltaForRemoteRegions = disableDeltaForRemoteRegions;
660 }
661
662 @Override
663 public int getRemoteRegionConnectTimeoutMs() {
664 return remoteRegionConnectTimeoutMs;
665 }
666
667 public void setRemoteRegionConnectTimeoutMs(int remoteRegionConnectTimeoutMs) {
668 this.remoteRegionConnectTimeoutMs = remoteRegionConnectTimeoutMs;
669 }
670
671 @Override
672 public int getRemoteRegionReadTimeoutMs() {
673 return remoteRegionReadTimeoutMs;
674 }
675
676 public void setRemoteRegionReadTimeoutMs(int remoteRegionReadTimeoutMs) {
677 this.remoteRegionReadTimeoutMs = remoteRegionReadTimeoutMs;
678 }
679
680 @Override
681 public int getRemoteRegionTotalConnections() {
682 return remoteRegionTotalConnections;
683 }
684
685 public void setRemoteRegionTotalConnections(int remoteRegionTotalConnections) {
686 this.remoteRegionTotalConnections = remoteRegionTotalConnections;
687 }
688
689 @Override
690 public int getRemoteRegionTotalConnectionsPerHost() {
691 return remoteRegionTotalConnectionsPerHost;
692 }
693
694 public void setRemoteRegionTotalConnectionsPerHost(
695 int remoteRegionTotalConnectionsPerHost) {
696 this.remoteRegionTotalConnectionsPerHost = remoteRegionTotalConnectionsPerHost;
697 }
698
699 @Override
700 public int getRemoteRegionConnectionIdleTimeoutSeconds() {
701 return remoteRegionConnectionIdleTimeoutSeconds;
702 }
703
704 public void setRemoteRegionConnectionIdleTimeoutSeconds(
705 int remoteRegionConnectionIdleTimeoutSeconds) {
706 this.remoteRegionConnectionIdleTimeoutSeconds = remoteRegionConnectionIdleTimeoutSeconds;
707 }
708
709 public boolean isgZipContentFromRemoteRegion() {
710 return gZipContentFromRemoteRegion;
711 }
712
713 public void setgZipContentFromRemoteRegion(boolean gZipContentFromRemoteRegion) {
714 this.gZipContentFromRemoteRegion = gZipContentFromRemoteRegion;
715 }
716
717 @Override
718 public Map<String, String> getRemoteRegionUrlsWithName() {
719 return remoteRegionUrlsWithName;
720 }
721
722 public void setRemoteRegionUrlsWithName(
723 Map<String, String> remoteRegionUrlsWithName) {
724 this.remoteRegionUrlsWithName = remoteRegionUrlsWithName;
725 }
726
727 @Override
728 public String[] getRemoteRegionUrls() {
729 return remoteRegionUrls;
730 }
731
732 public void setRemoteRegionUrls(String[] remoteRegionUrls) {
733 this.remoteRegionUrls = remoteRegionUrls;
734 }
735
736 public Map<String, Set<String>> getRemoteRegionAppWhitelist() {
737 return remoteRegionAppWhitelist;
738 }
739
740 public void setRemoteRegionAppWhitelist(
741 Map<String, Set<String>> remoteRegionAppWhitelist) {
742 this.remoteRegionAppWhitelist = remoteRegionAppWhitelist;
743 }
744
745 @Override
746 public int getRemoteRegionRegistryFetchInterval() {
747 return remoteRegionRegistryFetchInterval;
748 }
749
750 public void setRemoteRegionRegistryFetchInterval(
751 int remoteRegionRegistryFetchInterval) {
752 this.remoteRegionRegistryFetchInterval = remoteRegionRegistryFetchInterval;
753 }
754
755 @Override
756 public int getRemoteRegionFetchThreadPoolSize() {
757 return remoteRegionFetchThreadPoolSize;
758 }
759
760 public void setRemoteRegionFetchThreadPoolSize(int remoteRegionFetchThreadPoolSize) {
761 this.remoteRegionFetchThreadPoolSize = remoteRegionFetchThreadPoolSize;
762 }
763
764 @Override
765 public String getRemoteRegionTrustStore() {
766 return remoteRegionTrustStore;
767 }
768
769 public void setRemoteRegionTrustStore(String remoteRegionTrustStore) {
770 this.remoteRegionTrustStore = remoteRegionTrustStore;
771 }
772
773 @Override
774 public String getRemoteRegionTrustStorePassword() {
775 return remoteRegionTrustStorePassword;
776 }
777
778 public void setRemoteRegionTrustStorePassword(String remoteRegionTrustStorePassword) {
779 this.remoteRegionTrustStorePassword = remoteRegionTrustStorePassword;
780 }
781
782 public boolean isDisableTransparentFallbackToOtherRegion() {
783 return disableTransparentFallbackToOtherRegion;
784 }
785
786 public void setDisableTransparentFallbackToOtherRegion(
787 boolean disableTransparentFallbackToOtherRegion) {
788 this.disableTransparentFallbackToOtherRegion = disableTransparentFallbackToOtherRegion;
789 }
790
791 public boolean isBatchReplication() {
792 return batchReplication;
793 }
794
795 public void setBatchReplication(boolean batchReplication) {
796 this.batchReplication = batchReplication;
797 }
798
799 @Override
800 public boolean isRateLimiterEnabled() {
801 return rateLimiterEnabled;
802 }
803
804 public void setRateLimiterEnabled(boolean rateLimiterEnabled) {
805 this.rateLimiterEnabled = rateLimiterEnabled;
806 }
807
808 @Override
809 public boolean isRateLimiterThrottleStandardClients() {
810 return rateLimiterThrottleStandardClients;
811 }
812
813 public void setRateLimiterThrottleStandardClients(
814 boolean rateLimiterThrottleStandardClients) {
815 this.rateLimiterThrottleStandardClients = rateLimiterThrottleStandardClients;
816 }
817
818 @Override
819 public Set<String> getRateLimiterPrivilegedClients() {
820 return rateLimiterPrivilegedClients;
821 }
822
823 public void setRateLimiterPrivilegedClients(
824 Set<String> rateLimiterPrivilegedClients) {
825 this.rateLimiterPrivilegedClients = rateLimiterPrivilegedClients;
826 }
827
828 @Override
829 public int getRateLimiterBurstSize() {
830 return rateLimiterBurstSize;
831 }
832
833 public void setRateLimiterBurstSize(int rateLimiterBurstSize) {
834 this.rateLimiterBurstSize = rateLimiterBurstSize;
835 }
836
837 @Override
838 public int getRateLimiterRegistryFetchAverageRate() {
839 return rateLimiterRegistryFetchAverageRate;
840 }
841
842 public void setRateLimiterRegistryFetchAverageRate(
843 int rateLimiterRegistryFetchAverageRate) {
844 this.rateLimiterRegistryFetchAverageRate = rateLimiterRegistryFetchAverageRate;
845 }
846
847 @Override
848 public int getRateLimiterFullFetchAverageRate() {
849 return rateLimiterFullFetchAverageRate;
850 }
851
852 public void setRateLimiterFullFetchAverageRate(int rateLimiterFullFetchAverageRate) {
853 this.rateLimiterFullFetchAverageRate = rateLimiterFullFetchAverageRate;
854 }
855
856 public boolean isLogIdentityHeaders() {
857 return logIdentityHeaders;
858 }
859
860 public void setLogIdentityHeaders(boolean logIdentityHeaders) {
861 this.logIdentityHeaders = logIdentityHeaders;
862 }
863
864 @Override
865 public String getListAutoScalingGroupsRoleName() {
866 return listAutoScalingGroupsRoleName;
867 }
868
869 public void setListAutoScalingGroupsRoleName(String listAutoScalingGroupsRoleName) {
870 this.listAutoScalingGroupsRoleName = listAutoScalingGroupsRoleName;
871 }
872
873 public boolean isEnableReplicatedRequestCompression() {
874 return enableReplicatedRequestCompression;
875 }
876
877 public void setEnableReplicatedRequestCompression(
878 boolean enableReplicatedRequestCompression) {
879 this.enableReplicatedRequestCompression = enableReplicatedRequestCompression;
880 }
881
882 public void setJsonCodecName(String jsonCodecName) {
883 this.jsonCodecName = jsonCodecName;
884 }
885
886 public void setXmlCodecName(String xmlCodecName) {
887 this.xmlCodecName = xmlCodecName;
888 }
889
890 @Override
891 public int getRoute53BindRebindRetries() {
892 return route53BindRebindRetries;
893 }
894
895 public void setRoute53BindRebindRetries(int route53BindRebindRetries) {
896 this.route53BindRebindRetries = route53BindRebindRetries;
897 }
898
899 @Override
900 public int getRoute53BindingRetryIntervalMs() {
901 return route53BindingRetryIntervalMs;
902 }
903
904 public void setRoute53BindingRetryIntervalMs(int route53BindingRetryIntervalMs) {
905 this.route53BindingRetryIntervalMs = route53BindingRetryIntervalMs;
906 }
907
908 @Override
909 public long getRoute53DomainTTL() {
910 return route53DomainTTL;
911 }
912
913 public void setRoute53DomainTTL(long route53DomainTTL) {
914 this.route53DomainTTL = route53DomainTTL;
915 }
916
917 @Override
918 public AwsBindingStrategy getBindingStrategy() {
919 return bindingStrategy;
920 }
921
922 public void setBindingStrategy(AwsBindingStrategy bindingStrategy) {
923 this.bindingStrategy = bindingStrategy;
924 }
925
926 public int getMinAvailableInstancesForPeerReplication() {
927 return minAvailableInstancesForPeerReplication;
928 }
929
930 public void setMinAvailableInstancesForPeerReplication(
931 int minAvailableInstancesForPeerReplication) {
932 this.minAvailableInstancesForPeerReplication = minAvailableInstancesForPeerReplication;
933 }
934
935 @Override
936 public boolean equals(Object o) {
937 if (this == o) {
938 return true;
939 }
940 if (o == null || getClass() != o.getClass()) {
941 return false;
942 }
943 EurekaServerConfigBean that = (EurekaServerConfigBean) o;
944 return aSGCacheExpiryTimeoutMs == that.aSGCacheExpiryTimeoutMs
945 && aSGQueryTimeoutMs == that.aSGQueryTimeoutMs
946 && aSGUpdateIntervalMs == that.aSGUpdateIntervalMs
947 && Objects.equals(aWSAccessId, that.aWSAccessId)
948 && Objects.equals(aWSSecretKey, that.aWSSecretKey)
949 && batchReplication == that.batchReplication
950 && bindingStrategy == that.bindingStrategy
951 && deltaRetentionTimerIntervalInMs == that.deltaRetentionTimerIntervalInMs
952 && disableDelta == that.disableDelta
953 && disableDeltaForRemoteRegions == that.disableDeltaForRemoteRegions
954 && disableTransparentFallbackToOtherRegion == that.disableTransparentFallbackToOtherRegion
955 && eIPBindingRetryIntervalMs == that.eIPBindingRetryIntervalMs
956 && eIPBindingRetryIntervalMsWhenUnbound == that.eIPBindingRetryIntervalMsWhenUnbound
957 && eIPBindRebindRetries == that.eIPBindRebindRetries
958 && enableReplicatedRequestCompression == that.enableReplicatedRequestCompression
959 && enableSelfPreservation == that.enableSelfPreservation
960 && evictionIntervalTimerInMs == that.evictionIntervalTimerInMs
961 && gZipContentFromRemoteRegion == that.gZipContentFromRemoteRegion
962 && Objects.equals(jsonCodecName, that.jsonCodecName)
963 && Objects.equals(listAutoScalingGroupsRoleName,
964 that.listAutoScalingGroupsRoleName)
965 && logIdentityHeaders == that.logIdentityHeaders
966 && maxElementsInPeerReplicationPool == that.maxElementsInPeerReplicationPool
967 && maxElementsInStatusReplicationPool == that.maxElementsInStatusReplicationPool
968 && maxIdleThreadAgeInMinutesForPeerReplication == that.maxIdleThreadAgeInMinutesForPeerReplication
969 && maxIdleThreadInMinutesAgeForStatusReplication == that.maxIdleThreadInMinutesAgeForStatusReplication
970 && maxThreadsForPeerReplication == that.maxThreadsForPeerReplication
971 && maxThreadsForStatusReplication == that.maxThreadsForStatusReplication
972 && maxTimeForReplication == that.maxTimeForReplication
973 && minAvailableInstancesForPeerReplication == that.minAvailableInstancesForPeerReplication
974 && minThreadsForPeerReplication == that.minThreadsForPeerReplication
975 && minThreadsForStatusReplication == that.minThreadsForStatusReplication
976 && numberOfReplicationRetries == that.numberOfReplicationRetries
977 && peerEurekaNodesUpdateIntervalMs == that.peerEurekaNodesUpdateIntervalMs
978 && peerEurekaStatusRefreshTimeIntervalMs == that.peerEurekaStatusRefreshTimeIntervalMs
979 && peerNodeConnectionIdleTimeoutSeconds == that.peerNodeConnectionIdleTimeoutSeconds
980 && peerNodeConnectTimeoutMs == that.peerNodeConnectTimeoutMs
981 && peerNodeReadTimeoutMs == that.peerNodeReadTimeoutMs
982 && peerNodeTotalConnections == that.peerNodeTotalConnections
983 && peerNodeTotalConnectionsPerHost == that.peerNodeTotalConnectionsPerHost
984 && primeAwsReplicaConnections == that.primeAwsReplicaConnections
985 && Objects.equals(propertyResolver, that.propertyResolver)
986 && rateLimiterBurstSize == that.rateLimiterBurstSize
987 && rateLimiterEnabled == that.rateLimiterEnabled
988 && rateLimiterFullFetchAverageRate == that.rateLimiterFullFetchAverageRate
989 && Objects.equals(rateLimiterPrivilegedClients,
990 that.rateLimiterPrivilegedClients)
991 && rateLimiterRegistryFetchAverageRate == that.rateLimiterRegistryFetchAverageRate
992 && rateLimiterThrottleStandardClients == that.rateLimiterThrottleStandardClients
993 && registrySyncRetries == that.registrySyncRetries
994 && registrySyncRetryWaitMs == that.registrySyncRetryWaitMs
995 && Objects.equals(remoteRegionAppWhitelist, that.remoteRegionAppWhitelist)
996 && remoteRegionConnectionIdleTimeoutSeconds == that.remoteRegionConnectionIdleTimeoutSeconds
997 && remoteRegionConnectTimeoutMs == that.remoteRegionConnectTimeoutMs
998 && remoteRegionFetchThreadPoolSize == that.remoteRegionFetchThreadPoolSize
999 && remoteRegionReadTimeoutMs == that.remoteRegionReadTimeoutMs
1000 && remoteRegionRegistryFetchInterval == that.remoteRegionRegistryFetchInterval
1001 && remoteRegionTotalConnections == that.remoteRegionTotalConnections
1002 && remoteRegionTotalConnectionsPerHost == that.remoteRegionTotalConnectionsPerHost
1003 && Objects.equals(remoteRegionTrustStore, that.remoteRegionTrustStore)
1004 && Objects.equals(remoteRegionTrustStorePassword,
1005 that.remoteRegionTrustStorePassword)
1006 && Arrays.equals(remoteRegionUrls, that.remoteRegionUrls)
1007 && Objects.equals(remoteRegionUrlsWithName, that.remoteRegionUrlsWithName)
1008 && Double.compare(that.renewalPercentThreshold,
1009 renewalPercentThreshold) == 0
1010 && renewalThresholdUpdateIntervalMs == that.renewalThresholdUpdateIntervalMs
1011 && responseCacheAutoExpirationInSeconds == that.responseCacheAutoExpirationInSeconds
1012 && responseCacheUpdateIntervalMs == that.responseCacheUpdateIntervalMs
1013 && retentionTimeInMSInDeltaQueue == that.retentionTimeInMSInDeltaQueue
1014 && route53BindingRetryIntervalMs == that.route53BindingRetryIntervalMs
1015 && route53BindRebindRetries == that.route53BindRebindRetries
1016 && route53DomainTTL == that.route53DomainTTL
1017 && syncWhenTimestampDiffers == that.syncWhenTimestampDiffers
1018 && useReadOnlyResponseCache == that.useReadOnlyResponseCache
1019 && waitTimeInMsWhenSyncEmpty == that.waitTimeInMsWhenSyncEmpty
1020 && Objects.equals(xmlCodecName, that.xmlCodecName)
1021 && initialCapacityOfResponseCache == that.initialCapacityOfResponseCache
1022 && expectedClientRenewalIntervalSeconds == that.expectedClientRenewalIntervalSeconds
1023 && useAwsAsgApi == that.useAwsAsgApi && Objects.equals(myUrl, that.myUrl);
1024 }
1025
1026 @Override
1027 public int hashCode() {
1028 return Objects.hash(aSGCacheExpiryTimeoutMs, aSGQueryTimeoutMs,
1029 aSGUpdateIntervalMs, aWSAccessId, aWSSecretKey, batchReplication,
1030 bindingStrategy, deltaRetentionTimerIntervalInMs, disableDelta,
1031 disableDeltaForRemoteRegions, disableTransparentFallbackToOtherRegion,
1032 eIPBindRebindRetries, eIPBindingRetryIntervalMs,
1033 eIPBindingRetryIntervalMsWhenUnbound, enableReplicatedRequestCompression,
1034 enableSelfPreservation, evictionIntervalTimerInMs,
1035 gZipContentFromRemoteRegion, jsonCodecName, listAutoScalingGroupsRoleName,
1036 logIdentityHeaders, maxElementsInPeerReplicationPool,
1037 maxElementsInStatusReplicationPool,
1038 maxIdleThreadAgeInMinutesForPeerReplication,
1039 maxIdleThreadInMinutesAgeForStatusReplication,
1040 maxThreadsForPeerReplication, maxThreadsForStatusReplication,
1041 maxTimeForReplication, minAvailableInstancesForPeerReplication,
1042 minThreadsForPeerReplication, minThreadsForStatusReplication,
1043 numberOfReplicationRetries, peerEurekaNodesUpdateIntervalMs,
1044 peerEurekaStatusRefreshTimeIntervalMs, peerNodeConnectTimeoutMs,
1045 peerNodeConnectionIdleTimeoutSeconds, peerNodeReadTimeoutMs,
1046 peerNodeTotalConnections, peerNodeTotalConnectionsPerHost,
1047 primeAwsReplicaConnections, propertyResolver, rateLimiterBurstSize,
1048 rateLimiterEnabled, rateLimiterFullFetchAverageRate,
1049 rateLimiterPrivilegedClients, rateLimiterRegistryFetchAverageRate,
1050 rateLimiterThrottleStandardClients, registrySyncRetries,
1051 registrySyncRetryWaitMs, remoteRegionAppWhitelist,
1052 remoteRegionConnectTimeoutMs, remoteRegionConnectionIdleTimeoutSeconds,
1053 remoteRegionFetchThreadPoolSize, remoteRegionReadTimeoutMs,
1054 remoteRegionRegistryFetchInterval, remoteRegionTotalConnections,
1055 remoteRegionTotalConnectionsPerHost, remoteRegionTrustStore,
1056 remoteRegionTrustStorePassword, remoteRegionUrls,
1057 remoteRegionUrlsWithName, renewalPercentThreshold,
1058 renewalThresholdUpdateIntervalMs, responseCacheAutoExpirationInSeconds,
1059 responseCacheUpdateIntervalMs, retentionTimeInMSInDeltaQueue,
1060 route53BindRebindRetries, route53BindingRetryIntervalMs, route53DomainTTL,
1061 syncWhenTimestampDiffers, useReadOnlyResponseCache,
1062 waitTimeInMsWhenSyncEmpty, xmlCodecName, initialCapacityOfResponseCache,
1063 expectedClientRenewalIntervalSeconds, useAwsAsgApi, myUrl);
1064 }
1065
1066 @Override
1067 public String toString() {
1068 return new ToStringCreator(this)
1069 .append("aSGCacheExpiryTimeoutMs", this.aSGCacheExpiryTimeoutMs)
1070 .append("aSGQueryTimeoutMs", this.aSGQueryTimeoutMs)
1071 .append("aSGUpdateIntervalMs", this.aSGUpdateIntervalMs)
1072 .append("aWSAccessId", this.aWSAccessId)
1073 .append("aWSSecretKey", this.aWSSecretKey)
1074 .append("batchReplication", this.batchReplication)
1075 .append("bindingStrategy", this.bindingStrategy)
1076 .append("deltaRetentionTimerIntervalInMs",
1077 this.deltaRetentionTimerIntervalInMs)
1078 .append("disableDelta", this.disableDelta)
1079 .append("disableDeltaForRemoteRegions", this.disableDeltaForRemoteRegions)
1080 .append("disableTransparentFallbackToOtherRegion",
1081 this.disableTransparentFallbackToOtherRegion)
1082 .append("eIPBindRebindRetries", this.eIPBindRebindRetries)
1083 .append("eIPBindingRetryIntervalMs", this.eIPBindingRetryIntervalMs)
1084 .append("eIPBindingRetryIntervalMsWhenUnbound",
1085 this.eIPBindingRetryIntervalMsWhenUnbound)
1086 .append("enableReplicatedRequestCompression",
1087 this.enableReplicatedRequestCompression)
1088 .append("enableSelfPreservation", this.enableSelfPreservation)
1089 .append("evictionIntervalTimerInMs", this.evictionIntervalTimerInMs)
1090 .append("gZipContentFromRemoteRegion", this.gZipContentFromRemoteRegion)
1091 .append("jsonCodecName", this.jsonCodecName)
1092 .append("listAutoScalingGroupsRoleName",
1093 this.listAutoScalingGroupsRoleName)
1094 .append("logIdentityHeaders", this.logIdentityHeaders)
1095 .append("maxElementsInPeerReplicationPool",
1096 this.maxElementsInPeerReplicationPool)
1097 .append("maxElementsInStatusReplicationPool",
1098 this.maxElementsInStatusReplicationPool)
1099 .append("maxIdleThreadAgeInMinutesForPeerReplication",
1100 this.maxIdleThreadAgeInMinutesForPeerReplication)
1101 .append("maxIdleThreadInMinutesAgeForStatusReplication",
1102 this.maxIdleThreadInMinutesAgeForStatusReplication)
1103 .append("maxThreadsForPeerReplication", this.maxThreadsForPeerReplication)
1104 .append("maxThreadsForStatusReplication",
1105 this.maxThreadsForStatusReplication)
1106 .append("maxTimeForReplication", this.maxTimeForReplication)
1107 .append("minAvailableInstancesForPeerReplication",
1108 this.minAvailableInstancesForPeerReplication)
1109 .append("minThreadsForPeerReplication", this.minThreadsForPeerReplication)
1110 .append("minThreadsForStatusReplication",
1111 this.minThreadsForStatusReplication)
1112 .append("numberOfReplicationRetries", this.numberOfReplicationRetries)
1113 .append("peerEurekaNodesUpdateIntervalMs",
1114 this.peerEurekaNodesUpdateIntervalMs)
1115 .append("peerEurekaStatusRefreshTimeIntervalMs",
1116 this.peerEurekaStatusRefreshTimeIntervalMs)
1117 .append("peerNodeConnectTimeoutMs", this.peerNodeConnectTimeoutMs)
1118 .append("peerNodeConnectionIdleTimeoutSeconds",
1119 this.peerNodeConnectionIdleTimeoutSeconds)
1120 .append("peerNodeReadTimeoutMs", this.peerNodeReadTimeoutMs)
1121 .append("peerNodeTotalConnections", this.peerNodeTotalConnections)
1122 .append("peerNodeTotalConnectionsPerHost",
1123 this.peerNodeTotalConnectionsPerHost)
1124 .append("primeAwsReplicaConnections", this.primeAwsReplicaConnections)
1125 .append("propertyResolver", this.propertyResolver)
1126 .append("rateLimiterBurstSize", this.rateLimiterBurstSize)
1127 .append("rateLimiterEnabled", this.rateLimiterEnabled)
1128 .append("rateLimiterFullFetchAverageRate",
1129 this.rateLimiterFullFetchAverageRate)
1130 .append("rateLimiterPrivilegedClients", this.rateLimiterPrivilegedClients)
1131 .append("rateLimiterRegistryFetchAverageRate",
1132 this.rateLimiterRegistryFetchAverageRate)
1133 .append("rateLimiterThrottleStandardClients",
1134 this.rateLimiterThrottleStandardClients)
1135 .append("registrySyncRetries", this.registrySyncRetries)
1136 .append("registrySyncRetryWaitMs", this.registrySyncRetryWaitMs)
1137 .append("remoteRegionAppWhitelist", this.remoteRegionAppWhitelist)
1138 .append("remoteRegionConnectTimeoutMs", this.remoteRegionConnectTimeoutMs)
1139 .append("remoteRegionConnectionIdleTimeoutSeconds",
1140 this.remoteRegionConnectionIdleTimeoutSeconds)
1141 .append("remoteRegionFetchThreadPoolSize",
1142 this.remoteRegionFetchThreadPoolSize)
1143 .append("remoteRegionReadTimeoutMs", this.remoteRegionReadTimeoutMs)
1144 .append("remoteRegionRegistryFetchInterval",
1145 this.remoteRegionRegistryFetchInterval)
1146 .append("remoteRegionTotalConnections", this.remoteRegionTotalConnections)
1147 .append("remoteRegionTotalConnectionsPerHost",
1148 this.remoteRegionTotalConnectionsPerHost)
1149 .append("remoteRegionTrustStore", this.remoteRegionTrustStore)
1150 .append("remoteRegionTrustStorePassword",
1151 this.remoteRegionTrustStorePassword)
1152 .append("remoteRegionUrls", this.remoteRegionUrls)
1153 .append("remoteRegionUrlsWithName", this.remoteRegionUrlsWithName)
1154 .append("renewalPercentThreshold", this.renewalPercentThreshold)
1155 .append("renewalThresholdUpdateIntervalMs",
1156 this.renewalThresholdUpdateIntervalMs)
1157 .append("responseCacheAutoExpirationInSeconds",
1158 this.responseCacheAutoExpirationInSeconds)
1159 .append("responseCacheUpdateIntervalMs",
1160 this.responseCacheUpdateIntervalMs)
1161 .append("retentionTimeInMSInDeltaQueue",
1162 this.retentionTimeInMSInDeltaQueue)
1163 .append("route53BindRebindRetries", this.route53BindRebindRetries)
1164 .append("route53BindingRetryIntervalMs",
1165 this.route53BindingRetryIntervalMs)
1166 .append("route53DomainTTL", this.route53DomainTTL)
1167 .append("syncWhenTimestampDiffers", this.syncWhenTimestampDiffers)
1168 .append("useReadOnlyResponseCache", this.useReadOnlyResponseCache)
1169 .append("waitTimeInMsWhenSyncEmpty", this.waitTimeInMsWhenSyncEmpty)
1170 .append("xmlCodecName", this.xmlCodecName)
1171 .append("initialCapacityOfResponseCache",
1172 this.initialCapacityOfResponseCache)
1173 .append("expectedClientRenewalIntervalSeconds",
1174 this.expectedClientRenewalIntervalSeconds)
1175 .append("useAwsAsgApi", this.useAwsAsgApi).append("myUrl", this.myUrl)
1176 .toString();
1177 }
1178
1179 }
View Code
③ Eureka Server 初始化
还可以看到 EurekaServerAutoConfiguration 导入了 EurekaServerInitializerConfiguration 的初始化配置类，它启动了一个后台线程来初始化 eurekaServerBootstrap，进入可以看到跟 EurekaBootStrap 的初始化是类似的，只不过是简化了些，就不在展示了。
1 public void start() {
2 new Thread(() -> {
3 try {
4 // TODO: is this class even needed now?
5 eurekaServerBootstrap.contextInitialized(
6 EurekaServerInitializerConfiguration.this.servletContext);
7 log.info("Started Eureka Server");
8
9 publish(new EurekaRegistryAvailableEvent(getEurekaServerConfig()));
10 EurekaServerInitializerConfiguration.this.running = true;
11 publish(new EurekaServerStartedEvent(getEurekaServerConfig()));
12 }
13 catch (Exception ex) {
14 // Help!
15 log.error("Could not initialize Eureka servlet context", ex);
16 }
17 }).start();
18 }
View Code
2、spring-cloud-starter-netflix-eureka-client
① Eureka Client 自动化配置
Eureka Client 自动化配置类是 EurekaClientAutoConfiguration（@EnableEurekaClient 注解感觉没啥用），这里初始化类里主要初始化了 ApplicationInfoManager、EurekaClientConfigBean、EurekaInstanceConfigBean、EurekaClient 等。
需要注意，在 springcloud 中，EurekaClient 组件默认是 CloudEurekaClient；
1 @Configuration(proxyBeanMethods = false)
2 @EnableConfigurationProperties
3 @ConditionalOnClass(EurekaClientConfig.class)
4 @ConditionalOnProperty(value = "eureka.client.enabled", matchIfMissing = true)
5 @ConditionalOnDiscoveryEnabled
6 @AutoConfigureBefore({ NoopDiscoveryClientAutoConfiguration.class,
7 CommonsClientAutoConfiguration.class, ServiceRegistryAutoConfiguration.class })
8 @AutoConfigureAfter(name = {
9 "org.springframework.cloud.netflix.eureka.config.DiscoveryClientOptionalArgsConfiguration",
10 "org.springframework.cloud.autoconfigure.RefreshAutoConfiguration",
11 "org.springframework.cloud.netflix.eureka.EurekaDiscoveryClientConfiguration",
12 "org.springframework.cloud.client.serviceregistry.AutoServiceRegistrationAutoConfiguration" })
13 public class EurekaClientAutoConfiguration {
14
15 private ConfigurableEnvironment env;
16
17 public EurekaClientAutoConfiguration(ConfigurableEnvironment env) {
18 this.env = env;
19 }
20
21 @Bean
22 public HasFeatures eurekaFeature() {
23 return HasFeatures.namedFeature("Eureka Client", EurekaClient.class);
24 }
25
26 @Bean
27 @ConditionalOnMissingBean(value = EurekaClientConfig.class,
28 search = SearchStrategy.CURRENT)
29 public EurekaClientConfigBean eurekaClientConfigBean(ConfigurableEnvironment env) {
30 return new EurekaClientConfigBean();
31 }
32
33 @Bean
34 @ConditionalOnMissingBean
35 public ManagementMetadataProvider serviceManagementMetadataProvider() {
36 return new DefaultManagementMetadataProvider();
37 }
38
39 private String getProperty(String property) {
40 return this.env.containsProperty(property) ? this.env.getProperty(property) : "";
41 }
42
43 @Bean
44 @ConditionalOnMissingBean(value = EurekaInstanceConfig.class,
45 search = SearchStrategy.CURRENT)
46 public EurekaInstanceConfigBean eurekaInstanceConfigBean(InetUtils inetUtils,
47 ManagementMetadataProvider managementMetadataProvider) {
48 String hostname = getProperty("eureka.instance.hostname");
49 boolean preferIpAddress = Boolean
50 .parseBoolean(getProperty("eureka.instance.prefer-ip-address"));
51 String ipAddress = getProperty("eureka.instance.ip-address");
52 boolean isSecurePortEnabled = Boolean
53 .parseBoolean(getProperty("eureka.instance.secure-port-enabled"));
54
55 String serverContextPath = env.getProperty("server.servlet.context-path", "/");
56 int serverPort = Integer.parseInt(
57 env.getProperty("server.port", env.getProperty("port", "8080")));
58
59 Integer managementPort = env.getProperty("management.server.port", Integer.class);
60 String managementContextPath = env
61 .getProperty("management.server.servlet.context-path");
62 Integer jmxPort = env.getProperty("com.sun.management.jmxremote.port",
63 Integer.class);
64 EurekaInstanceConfigBean instance = new EurekaInstanceConfigBean(inetUtils);
65
66 instance.setNonSecurePort(serverPort);
67 instance.setInstanceId(getDefaultInstanceId(env));
68 instance.setPreferIpAddress(preferIpAddress);
69 instance.setSecurePortEnabled(isSecurePortEnabled);
70 if (StringUtils.hasText(ipAddress)) {
71 instance.setIpAddress(ipAddress);
72 }
73
74 if (isSecurePortEnabled) {
75 instance.setSecurePort(serverPort);
76 }
77
78 if (StringUtils.hasText(hostname)) {
79 instance.setHostname(hostname);
80 }
81 String statusPageUrlPath = getProperty("eureka.instance.status-page-url-path");
82 String healthCheckUrlPath = getProperty("eureka.instance.health-check-url-path");
83
84 if (StringUtils.hasText(statusPageUrlPath)) {
85 instance.setStatusPageUrlPath(statusPageUrlPath);
86 }
87 if (StringUtils.hasText(healthCheckUrlPath)) {
88 instance.setHealthCheckUrlPath(healthCheckUrlPath);
89 }
90
91 ManagementMetadata metadata = managementMetadataProvider.get(instance, serverPort,
92 serverContextPath, managementContextPath, managementPort);
93
94 if (metadata != null) {
95 instance.setStatusPageUrl(metadata.getStatusPageUrl());
96 instance.setHealthCheckUrl(metadata.getHealthCheckUrl());
97 if (instance.isSecurePortEnabled()) {
98 instance.setSecureHealthCheckUrl(metadata.getSecureHealthCheckUrl());
99 }
100 Map<String, String> metadataMap = instance.getMetadataMap();
101 metadataMap.computeIfAbsent("management.port",
102 k -> String.valueOf(metadata.getManagementPort()));
103 }
104 else {
105 // without the metadata the status and health check URLs will not be set
106 // and the status page and health check url paths will not include the
107 // context path so set them here
108 if (StringUtils.hasText(managementContextPath)) {
109 instance.setHealthCheckUrlPath(
110 managementContextPath + instance.getHealthCheckUrlPath());
111 instance.setStatusPageUrlPath(
112 managementContextPath + instance.getStatusPageUrlPath());
113 }
114 }
115
116 setupJmxPort(instance, jmxPort);
117 return instance;
118 }
119
120 private void setupJmxPort(EurekaInstanceConfigBean instance, Integer jmxPort) {
121 Map<String, String> metadataMap = instance.getMetadataMap();
122 if (metadataMap.get("jmx.port") == null && jmxPort != null) {
123 metadataMap.put("jmx.port", String.valueOf(jmxPort));
124 }
125 }
126
127 @Bean
128 public EurekaServiceRegistry eurekaServiceRegistry() {
129 return new EurekaServiceRegistry();
130 }
131
132 // @Bean
133 // @ConditionalOnBean(AutoServiceRegistrationProperties.class)
134 // @ConditionalOnProperty(value =
135 // "spring.cloud.service-registry.auto-registration.enabled", matchIfMissing = true)
136 // public EurekaRegistration eurekaRegistration(EurekaClient eurekaClient,
137 // CloudEurekaInstanceConfig instanceConfig, ApplicationInfoManager
138 // applicationInfoManager, ObjectProvider<HealthCheckHandler> healthCheckHandler) {
139 // return EurekaRegistration.builder(instanceConfig)
140 // .with(applicationInfoManager)
141 // .with(eurekaClient)
142 // .with(healthCheckHandler)
143 // .build();
144 // }
145
146 @Bean
147 @ConditionalOnBean(AutoServiceRegistrationProperties.class)
148 @ConditionalOnProperty(
149 value = "spring.cloud.service-registry.auto-registration.enabled",
150 matchIfMissing = true)
151 public EurekaAutoServiceRegistration eurekaAutoServiceRegistration(
152 ApplicationContext context, EurekaServiceRegistry registry,
153 EurekaRegistration registration) {
154 return new EurekaAutoServiceRegistration(context, registry, registration);
155 }
156
157 @Configuration(proxyBeanMethods = false)
158 @ConditionalOnMissingRefreshScope
159 protected static class EurekaClientConfiguration {
160
161 @Autowired
162 private ApplicationContext context;
163
164 @Autowired
165 private AbstractDiscoveryClientOptionalArgs<?> optionalArgs;
166
167 @Bean(destroyMethod = "shutdown")
168 @ConditionalOnMissingBean(value = EurekaClient.class,
169 search = SearchStrategy.CURRENT)
170 public EurekaClient eurekaClient(ApplicationInfoManager manager,
171 EurekaClientConfig config) {
172 return new CloudEurekaClient(manager, config, this.optionalArgs,
173 this.context);
174 }
175
176 @Bean
177 @ConditionalOnMissingBean(value = ApplicationInfoManager.class,
178 search = SearchStrategy.CURRENT)
179 public ApplicationInfoManager eurekaApplicationInfoManager(
180 EurekaInstanceConfig config) {
181 InstanceInfo instanceInfo = new InstanceInfoFactory().create(config);
182 return new ApplicationInfoManager(config, instanceInfo);
183 }
184
185 @Bean
186 @ConditionalOnBean(AutoServiceRegistrationProperties.class)
187 @ConditionalOnProperty(
188 value = "spring.cloud.service-registry.auto-registration.enabled",
189 matchIfMissing = true)
190 public EurekaRegistration eurekaRegistration(EurekaClient eurekaClient,
191 CloudEurekaInstanceConfig instanceConfig,
192 ApplicationInfoManager applicationInfoManager, @Autowired(
193 required = false) ObjectProvider<HealthCheckHandler> healthCheckHandler) {
194 return EurekaRegistration.builder(instanceConfig).with(applicationInfoManager)
195 .with(eurekaClient).with(healthCheckHandler).build();
196 }
197
198 }
199
200 @Configuration(proxyBeanMethods = false)
201 @ConditionalOnRefreshScope
202 protected static class RefreshableEurekaClientConfiguration {
203
204 @Autowired
205 private ApplicationContext context;
206
207 @Autowired
208 private AbstractDiscoveryClientOptionalArgs<?> optionalArgs;
209
210 @Bean(destroyMethod = "shutdown")
211 @ConditionalOnMissingBean(value = EurekaClient.class,
212 search = SearchStrategy.CURRENT)
213 @org.springframework.cloud.context.config.annotation.RefreshScope
214 @Lazy
215 public EurekaClient eurekaClient(ApplicationInfoManager manager,
216 EurekaClientConfig config, EurekaInstanceConfig instance,
217 @Autowired(required = false) HealthCheckHandler healthCheckHandler) {
218 // If we use the proxy of the ApplicationInfoManager we could run into a
219 // problem
220 // when shutdown is called on the CloudEurekaClient where the
221 // ApplicationInfoManager bean is
222 // requested but wont be allowed because we are shutting down. To avoid this
223 // we use the
224 // object directly.
225 ApplicationInfoManager appManager;
226 if (AopUtils.isAopProxy(manager)) {
227 appManager = ProxyUtils.getTargetObject(manager);
228 }
229 else {
230 appManager = manager;
231 }
232 CloudEurekaClient cloudEurekaClient = new CloudEurekaClient(appManager,
233 config, this.optionalArgs, this.context);
234 cloudEurekaClient.registerHealthCheck(healthCheckHandler);
235 return cloudEurekaClient;
236 }
237
238 @Bean
239 @ConditionalOnMissingBean(value = ApplicationInfoManager.class,
240 search = SearchStrategy.CURRENT)
241 @org.springframework.cloud.context.config.annotation.RefreshScope
242 @Lazy
243 public ApplicationInfoManager eurekaApplicationInfoManager(
244 EurekaInstanceConfig config) {
245 InstanceInfo instanceInfo = new InstanceInfoFactory().create(config);
246 return new ApplicationInfoManager(config, instanceInfo);
247 }
248
249 @Bean
250 @org.springframework.cloud.context.config.annotation.RefreshScope
251 @ConditionalOnBean(AutoServiceRegistrationProperties.class)
252 @ConditionalOnProperty(
253 value = "spring.cloud.service-registry.auto-registration.enabled",
254 matchIfMissing = true)
255 public EurekaRegistration eurekaRegistration(EurekaClient eurekaClient,
256 CloudEurekaInstanceConfig instanceConfig,
257 ApplicationInfoManager applicationInfoManager, @Autowired(
258 required = false) ObjectProvider<HealthCheckHandler> healthCheckHandler) {
259 return EurekaRegistration.builder(instanceConfig).with(applicationInfoManager)
260 .with(eurekaClient).with(healthCheckHandler).build();
261 }
262
263 }
264
265 @Target({ ElementType.TYPE, ElementType.METHOD })
266 @Retention(RetentionPolicy.RUNTIME)
267 @Documented
268 @Conditional(OnMissingRefreshScopeCondition.class)
269 @interface ConditionalOnMissingRefreshScope {
270
271 }
272
273 @Target({ ElementType.TYPE, ElementType.METHOD })
274 @Retention(RetentionPolicy.RUNTIME)
275 @Documented
276 @ConditionalOnClass(RefreshScope.class)
277 @ConditionalOnBean(RefreshAutoConfiguration.class)
278 @ConditionalOnProperty(value = "eureka.client.refresh.enable", havingValue = "true",
279 matchIfMissing = true)
280 @interface ConditionalOnRefreshScope {
281
282 }
283
284 private static class OnMissingRefreshScopeCondition extends AnyNestedCondition {
285
286 OnMissingRefreshScopeCondition() {
287 super(ConfigurationPhase.REGISTER_BEAN);
288 }
289
290 @ConditionalOnMissingClass("org.springframework.cloud.context.scope.refresh.RefreshScope")
291 static class MissingClass {
292
293 }
294
295 @ConditionalOnMissingBean(RefreshAutoConfiguration.class)
296 static class MissingScope {
297
298 }
299
300 @ConditionalOnProperty(value = "eureka.client.refresh.enable",
301 havingValue = "false")
302 static class OnPropertyDisabled {
303
304 }
305
306 }
307
308 @Configuration(proxyBeanMethods = false)
309 @ConditionalOnClass(Health.class)
310 protected static class EurekaHealthIndicatorConfiguration {
311
312 @Bean
313 @ConditionalOnMissingBean
314 @ConditionalOnEnabledHealthIndicator("eureka")
315 public EurekaHealthIndicator eurekaHealthIndicator(EurekaClient eurekaClient,
316 EurekaInstanceConfig instanceConfig, EurekaClientConfig clientConfig) {
317 return new EurekaHealthIndicator(eurekaClient, instanceConfig, clientConfig);
318 }
319
320 }
321
322 }
View Code
② Eureka Client 注册
Netflix 中服务注册的逻辑是在 InstanceInfoReplicator，springcloud 则封装到了 EurekaAutoServiceRegistration，InstanceInfoReplicator 启动之后要延迟40秒才会注册到注册中心，而这里的自动化配置在服务启动时就会注册到注册中心。
它这里调用了 serviceRegistry 来注册，进去可以发现它就是调用了 ApplicationInfoManager 的 setInstanceStatus 方法，进而触发了那个状态变更器 StatusChangeListener，然后向注册中心注册。
1 public void start() {
2 // only set the port if the nonSecurePort or securePort is 0 and this.port != 0
3 if (this.port.get() != 0) {
4 if (this.registration.getNonSecurePort() == 0) {
5 this.registration.setNonSecurePort(this.port.get());
6 }
7
8 if (this.registration.getSecurePort() == 0 && this.registration.isSecure()) {
9 this.registration.setSecurePort(this.port.get());
10 }
11 }
12
13 // only initialize if nonSecurePort is greater than 0 and it isn't already running
14 // because of containerPortInitializer below
15 if (!this.running.get() && this.registration.getNonSecurePort() > 0) {
16
17 this.serviceRegistry.register(this.registration);
18
19 this.context.publishEvent(new InstanceRegisteredEvent<>(this,
20 this.registration.getInstanceConfig()));
21 this.running.set(true);
22 }
23 }
View Code
十三、Eureka 总结
这一节来对 eureka 的学习做个总结，注意下面的一些截图是来自《重新定义Spring Cloud实战 》，具体可以参考原著。
1、Eureka Server 提供的 API
大部分的 API 在前面的源码分析中都已经接触过了，这里看下 eureka 提供的 API列表。注意 eureka 整合到 springcloud 之后，api 前缀固定为 /eureka，在 netflix 中是 /{version} 的形式。
2、Eureka Client 核心参数
Eureka Client 的参数可以分为基本参数、定时任务参数、http参数三大类。
① 基本参数
② 定时任务参数
③ http 参数
3、Eureka Server 核心参数
Eureka Server 的参数可以分为 基本参数、多级缓存参数、集群相关参数、http参数 四大类。
① 基本参数
② 多级缓存参数
③ 集群参数
④ http 参数
4、Eureka 核心功能
① 服务注册和发现：eureka 分客户端（Eureka Client）和服务端（Eureka Server），服务端即为注册中心，提供服务注册和发现的功能。所有客户端将自己注册到注册中心上，服务端使用 Map 结构基于内存保存所有客户端信息（IP、端口、续约等信息）。客户端定时从注册中心拉取注册表到本地，就可以通过负载均衡的方式进行服务间的调用。
② 服务注册（Register）：Eureka Client 启动时向 Eureka Server 注册，并提供自身的元数据、IP地址、端口、状态等信息。
③ 服务续约（Renew）：Eureka Client 默认每隔30秒向 Eureka Server 发送一次心跳进行服务续约，通过续约告知 Eureka Server 自己是正常的。如果 Eureka Server 180 秒没有收到客户端的续约，就会认为客户端故障，并将其剔除。
④ 抓取注册表（Fetch Registry）：Eureka Client 启动时会向 Eureka Server 全量抓取一次注册表到本地，之后会每隔30秒增量抓取注册表合并到本地注册表。如果合并后的本地注册表与 Eureka Server 端的注册表不一致（hash 比对），就全量抓取注册表覆盖本地的注册表。
⑤ 服务下线（Cancel）：Eureka Client 程序正常关闭时，会向 Eureka Server 发送下线请求，之后 Eureka Server 将这个实例从注册表中剔除。
⑥ 故障剔除（Eviction）：默认情况下，Eureka Client 连续180秒没有向 Eureka Server 发送续约请求，就会被认为实例故障，然后从注册表剔除。
⑦ Eureka Server 集群：Eureka Server 采用对等复制模式（Peer to Peer）来进行副本之间的数据同步，集群中每个 Server 节点都可以接收写操作和读操作。Server 节点接收到写操作后（注册、续约、下线、状态更新）会通过后台任务打包成批量任务发送到集群其它 Server 节点进行数据同步。Eureka Server 集群副本之间的数据会有短暂的不一致性，它是满足 CAP 中的 AP，即 高可用性和分区容错性。
5、Eureka 整体架构
最后用一张图来总结下 Eureka 的整体架构、运行流程以及核心机制。
https://www.cnblogs.com/chiangchou/p/eureka-1.html
Recommend
About Joyk
Aggregate valuable and interesting links.
Joyk means Joy of geeK