Eureka Server集群数据同步_eureka数据同步 发布订阅实现

一、数据同步

Eureka Server之间会互相进行注册，构建Eureka Server集群，不同Eureka Server之间会进行服务同步，用来保证服务信息的一致性。当服务提供者发送注册请求到一个服务注册中心时， 它会将该请求转发给集群中相连的其他注册中心， 从而实现注册中心之间的服务同步。通过服务同步，两个服务提供者的服务信息就可以通过这两台服务注册中心中的任意一台获取到。

二、源码解析

Eureka Server 集群不区分主从节点，所有节点相同角色（也就是没有角色），完全对等。

提供集群功能的包路径：com.netflix.eureka.cluster

以下提到的同步，准确来说，就是复制（Replication）。

1、集群节点初始化

Eureka Server 封装了一个集群节点管理的类 PeerEurekaNodes。

/**
 *	eureka server 集群节点集合类，帮助管理维护集群节点
 */
@Singleton
public class PeerEurekaNodes {
	//应用实例注册表
	protected final PeerAwareInstanceRegistry registry;
	//Eureka Server 配置
    protected final EurekaServerConfig serverConfig;
    //Eureka Client 配置
    protected final EurekaClientConfig clientConfig;
    //Eureka Server 编解码
    protected final ServerCodecs serverCodecs;
    //应用实例信息管理器
    private final ApplicationInfoManager applicationInfoManager;
	
	//集群节点集合
    private volatile List<PeerEurekaNode> peerEurekaNodes = Collections.emptyList();
    //集群节点URL集合
    private volatile Set<String> peerEurekaNodeUrls = Collections.emptySet();
	//定时任务线程池
    private ScheduledExecutorService taskExecutor;

	//构造方法
	@Inject
    public PeerEurekaNodes(
            PeerAwareInstanceRegistry registry,
            EurekaServerConfig serverConfig,
            EurekaClientConfig clientConfig,
            ServerCodecs serverCodecs,
            ApplicationInfoManager applicationInfoManager) {
        this.registry = registry;
        this.serverConfig = serverConfig;
        this.clientConfig = clientConfig;
        this.serverCodecs = serverCodecs;
        this.applicationInfoManager = applicationInfoManager;
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37

peerEurekaNodes、peerEurekaNodeUrls、taskExecutor 属性，在构造方法中未设置和初始化，而是在 PeerEurekaNodes#start() 方法中，设置和初始化，接下来我们看 start() 方法的实现。

PeerEurekaNodes#start() 集群节点启动方法，主要完成以下几件事：

• 初始化定时任务线程池
• 初始化集群节点信息 updatePeerEurekaNodes 方法
• 初始化固定周期（默认10分钟，可配置）更新集群节点信息的任务的线程
• 通过定时任务，线程池定时执行更新集群节点线程

public void start() {
	//1、初始化定时任务线程池
    taskExecutor = Executors.newSingleThreadScheduledExecutor(
            new ThreadFactory() {
                @Override
                public Thread newThread(Runnable r) {
                    Thread thread = new Thread(r, "Eureka-PeerNodesUpdater");
                    thread.setDaemon(true);
                    return thread;
                }
            }
    );
    try {
    	//2、初始化集群节点信息
        updatePeerEurekaNodes(resolvePeerUrls());
        //3、初始化固定周期更新集群节点信息的任务（节点更新任务线程）
        Runnable peersUpdateTask = new Runnable() {
            @Override
            public void run() {
                try {
                    updatePeerEurekaNodes(resolvePeerUrls());
                } catch (Throwable e) {
                    logger.error("Cannot update the replica Nodes", e);
                }

            }
        };
        //4、创建并执行一个在给定初始延迟后首次启用的定期操作，随后，在每一次执行终止
        //和下一次执行开始之间都存在给定的延迟。如果任务的任一执行遇到异常，就会取
        //消后续执行，否则，只能通过执行程序的取消或终止方法来终止该任务。
        //参数：command（第一个参数）-要执行的任务，initialdelay（第二个参数）-首次执行的延迟时间
        //delay（第三个参数）-一次执行终止和下一次执行开始之间的延迟，
        //unit（第四个参数）-initialdelay和delay参数的时间单位
        taskExecutor.scheduleWithFixedDelay(
                peersUpdateTask,
                serverConfig.getPeerEurekaNodesUpdateIntervalMs(),
                serverConfig.getPeerEurekaNodesUpdateIntervalMs(),
                TimeUnit.MILLISECONDS
        );
    } catch (Exception e) {
        throw new IllegalStateException(e);
    }
    for (PeerEurekaNode node : peerEurekaNodes) {
        logger.info("Replica node URL:  {}", node.getServiceUrl());
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46

2、更新集群节点信息

通过 start() 方法可以看出，Eureka Server 是通过一个定时线程定时去更新集群的节点信息达到对集群节点的动态发现和感知，在上面我们可以看到更新操作主要由 updatePeerEurekaNodes(resolvePeerUrls()) 方法完成，下面查看此方法的实现。

protected void updatePeerEurekaNodes(List<String> newPeerUrls) {
    if (newPeerUrls.isEmpty()) {
        logger.warn("The replica size seems to be empty. Check the route 53 DNS Registry");
        return;
    }

	//计算要删除的集群节点地址（从以前的地址中删除最新的地址信息，剩下的就是不可用的地址）
    Set<String> toShutdown = new HashSet<>(peerEurekaNodeUrls);
    toShutdown.removeAll(newPeerUrls);
    //计算要新增的集群节点地址（从最新的地址中删除以前的地址信息，剩下的就是新增的地址）
    Set<String> toAdd = new HashSet<>(newPeerUrls);
    toAdd.removeAll(peerEurekaNodeUrls);
    
	//如果这两个集合都为空，说明前后地址信息一致，既没有新增也没有删除，不需要更新直接返回
    if (toShutdown.isEmpty() && toAdd.isEmpty()) { // No change
        return;
    }

    //这是以前的所有服务节点
    List<PeerEurekaNode> newNodeList = new ArrayList<>(peerEurekaNodes);

	//移除旧集合中不可用的节点信息
    if (!toShutdown.isEmpty()) {
        logger.info("Removing no longer available peer nodes {}", toShutdown);
        int i = 0;
        while (i < newNodeList.size()) {
            PeerEurekaNode eurekaNode = newNodeList.get(i);
            if (toShutdown.contains(eurekaNode.getServiceUrl())) {
            	//删除不可用节点，并关闭
                newNodeList.remove(i);
                eurekaNode.shutDown();
            } else {
                i++;
            }
        }
    }

    //添加新增加的节点信息
    if (!toAdd.isEmpty()) {
        logger.info("Adding new peer nodes {}", toAdd);
        for (String peerUrl : toAdd) {
            newNodeList.add(createPeerEurekaNode(peerUrl));
        }
    }
	//重新赋值：集群节点集合、集群节点URL集合
    this.peerEurekaNodes = newNodeList;
    this.peerEurekaNodeUrls = new HashSet<>(newPeerUrls);
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48

updatePeerEurekaNodes(resolvePeerUrls()) 根据传入的新集群的 URL 集合完成节点的更新

• 校验传入的 URL 集合是否需要更新
• 移除新 URL 集合中没有的旧节点并关闭节点
• 创建旧节点集合中没有的新 URL 节点，通过 createPeerEurekaNode(peerUrl) 方法
• 重新赋值节点集合以及节点URL集合完成节点的更新

3、创建节点信息

updatePeerEurekaNodes(resolvePeerUrls()) 方法传入的新 URL 集合，是通过resolvePeerUrls() 方法获取，这个方法实际上就是解析配置文件中的 eureka.serviceUrl 前缀的配置，并动态监听配置的更新。下面我们看创建节点的方法：

protected PeerEurekaNode createPeerEurekaNode(String peerEurekaNodeUrl) {
	//创建一个连接远程节点的客户端
    HttpReplicationClient replicationClient = JerseyReplicationClient.createReplicationClient(serverConfig, serverCodecs, peerEurekaNodeUrl);
    //获取新节点host信息
    String targetHost = hostFromUrl(peerEurekaNodeUrl);
    if (targetHost == null) {
        targetHost = "host";
    }
    //创建新节点
    return new PeerEurekaNode(registry, targetHost, peerEurekaNodeUrl, replicationClient, serverConfig);
}
1
2
3
4
5
6
7
8
9
10
11

4、集群节点数据同步

我们来看看创建新节点方法 PeerEurekaNode，集群节点间都有哪些数据需要同步

PeerEurekaNode(PeerAwareInstanceRegistry registry, String targetHost, String serviceUrl,
                                 HttpReplicationClient replicationClient, EurekaServerConfig config,
                                 int batchSize, long maxBatchingDelayMs,
                                 long retrySleepTimeMs, long serverUnavailableSleepTimeMs) {
    this.registry = registry;
    this.targetHost = targetHost;
    this.replicationClient = replicationClient;

    this.serviceUrl = serviceUrl;
    this.config = config;
    this.maxProcessingDelayMs = config.getMaxTimeForReplication();

    String batcherName = getBatcherName();
    //初始化：集群复制任务处理器
    ReplicationTaskProcessor taskProcessor = new ReplicationTaskProcessor(targetHost, replicationClient);
    //初始化：批量任务分发器
    this.batchingDispatcher = TaskDispatchers.createBatchingTaskDispatcher(
            batcherName,
            config.getMaxElementsInPeerReplicationPool(),
            batchSize,
            config.getMaxThreadsForPeerReplication(),
            maxBatchingDelayMs,
            serverUnavailableSleepTimeMs,
            retrySleepTimeMs,
            taskProcessor
    );
    //初始化：单任务分发器
    this.nonBatchingDispatcher = TaskDispatchers.createNonBatchingTaskDispatcher(
            targetHost,
            config.getMaxElementsInStatusReplicationPool(),
            config.getMaxThreadsForStatusReplication(),
            maxBatchingDelayMs,
            serverUnavailableSleepTimeMs,
            retrySleepTimeMs,
            taskProcessor
    );
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37

PeerEurekaNode 完成以下几件事

• 创建数据同步的任务处理器 ReplicationTaskProcessor
• 创建批处理任务分发器
• 创建单任务分发器

说明：eureka 将节点间的数据同步工作包装成一个个细微的任务 ReplicationTask，每一个数据操作代表一个任务，将任务发送给任务调度器 TaskDispatcher 去异步处理。

接下来我们看看 PeerEurekaNode 都可以创建哪些同步任务：

• register：当 Eureka Server 注册新服务时，同时创建一个定时任务将新服务同步到集群其他节点

public void register(final InstanceInfo info) throws Exception {
    long expiryTime = System.currentTimeMillis() + getLeaseRenewalOf(info);
    //任务调度器中添加一个请求类型为注册register新服务的同步任务
    batchingDispatcher.process(
            taskId("register", info),
            new InstanceReplicationTask(targetHost, Action.Register, info, null, true) {
                public EurekaHttpResponse<Void> execute() {
                    return replicationClient.register(info);
                }
            },
            expiryTime
    );
}
1
2
3
4
5
6
7
8
9
10
11
12
13

• cancel：取消服务注册任务，当前节点有服务取消注册，将信息同步到集群远程节点

public void cancel(final String appName, final String id) throws Exception {
    long expiryTime = System.currentTimeMillis() + maxProcessingDelayMs;
    //任务调度器中添加一个请求类型为取消cancel服务的同步任务
    batchingDispatcher.process(
            taskId("cancel", appName, id),
            new InstanceReplicationTask(targetHost, Action.Cancel, appName, id) {
                @Override
                public EurekaHttpResponse<Void> execute() {
                    return replicationClient.cancel(appName, id);
                }

                @Override
                public void handleFailure(int statusCode, Object responseEntity) throws Throwable {
                    super.handleFailure(statusCode, responseEntity);
                    if (statusCode == 404) {
                        logger.warn("{}: missing entry.", getTaskName());
                    }
                }
            },
            expiryTime
    );
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

• heartbeat：心跳同步任务，当前节点有服务发送心跳续租，将信息同步到集群远程节点

public void heartbeat(final String appName, final String id,
                      final InstanceInfo info, final InstanceStatus overriddenStatus,
                      boolean primeConnection) throws Throwable {
    if (primeConnection) {
        // We do not care about the result for priming request.
        replicationClient.sendHeartBeat(appName, id, info, overriddenStatus);
        return;
    }
    ReplicationTask replicationTask = new InstanceReplicationTask(targetHost, Action.Heartbeat, info, overriddenStatus, false) {
        @Override
        public EurekaHttpResponse<InstanceInfo> execute() throws Throwable {
            return replicationClient.sendHeartBeat(appName, id, info, overriddenStatus);
        }

        @Override
        public void handleFailure(int statusCode, Object responseEntity) throws Throwable {
            super.handleFailure(statusCode, responseEntity);
            if (statusCode == 404) {
                logger.warn("{}: missing entry.", getTaskName());
                if (info != null) {
                    logger.warn("{}: cannot find instance id {} and hence replicating the instance with status {}",
                            getTaskName(), info.getId(), info.getStatus());
                    register(info);
                }
            } else if (config.shouldSyncWhenTimestampDiffers()) {
                InstanceInfo peerInstanceInfo = (InstanceInfo) responseEntity;
                if (peerInstanceInfo != null) {
                    syncInstancesIfTimestampDiffers(appName, id, info, peerInstanceInfo);
                }
            }
        }
    };
    long expiryTime = System.currentTimeMillis() + getLeaseRenewalOf(info);
    //任务调度器中添加一个请求类型为heartbeat服务的同步任务
    batchingDispatcher.process(taskId("heartbeat", info), replicationTask, expiryTime);
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36

接下来还有StatusUpdate、DeleteStatusOverride，就不一一列举。

5、集群节点数据同步任务处理

在 PeerEurekaNode 的构造函数中可以看到同步任务处理由 ReplicationTaskProcessor 完成，下面看此类源码

/**
 * 单个处理 ReplicationTask任务
 */
@Override
public ProcessingResult process(ReplicationTask task) {
    try {
    	//调用任务execute方法，完成任务的执行
        EurekaHttpResponse<?> httpResponse = task.execute();
        int statusCode = httpResponse.getStatusCode();
        Object entity = httpResponse.getEntity();
        if (logger.isDebugEnabled()) {
            logger.debug("Replication task {} completed with status {}, (includes entity {})", task.getTaskName(), statusCode, entity != null);
        }
        //判断任务返回结果
        if (isSuccess(statusCode)) {
            task.handleSuccess();
        } else if (statusCode == 503) {
            logger.debug("Server busy (503) reply for task {}", task.getTaskName());
            return ProcessingResult.Congestion;
        } else {
            task.handleFailure(statusCode, entity);
            return ProcessingResult.PermanentError;
        }
    } catch (Throwable e) {
    	if (maybeReadTimeOut(e)) {
            logger.error("It seems to be a socket read timeout exception, it will retry later. if it continues to happen and some eureka node occupied all the cpu time, you should set property 'eureka.server.peer-node-read-timeout-ms' to a bigger value", e);
        	//read timeout exception is more Congestion then TransientError, return Congestion for longer delay 
            return ProcessingResult.Congestion;
        } else if (isNetworkConnectException(e)) {
            logNetworkErrorSample(task, e);
            return ProcessingResult.TransientError;
        } else {
            logger.error("{}: {} Not re-trying this exception because it does not seem to be a network exception",
                    peerId, task.getTaskName(), e);
            return ProcessingResult.PermanentError;
        }
    }
    return ProcessingResult.Success;
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39

单任务处理

• 调用任务 task 的 execute 完成远程数据同步
• 分析远程返回结果

/**
 * 批量处理ReplicationTask任务
 */
@Override
public ProcessingResult process(List<ReplicationTask> tasks) {
	//根据task集合创建ReplicationList
    ReplicationList list = createReplicationListOf(tasks);
    try {
    	//调用批量同步接口 将同步集合发送到远端节点同步数据
        EurekaHttpResponse<ReplicationListResponse> response = replicationClient.submitBatchUpdates(list);
        //判断同步返回结果
        int statusCode = response.getStatusCode();
        if (!isSuccess(statusCode)) {
            if (statusCode == 503) {
                logger.warn("Server busy (503) HTTP status code received from the peer {}; rescheduling tasks after delay", peerId);
                return ProcessingResult.Congestion;
            } else {
                // Unexpected error returned from the server. This should ideally never happen.
                logger.error("Batch update failure with HTTP status code {}; discarding {} replication tasks", statusCode, tasks.size());
                return ProcessingResult.PermanentError;
            }
        } else {
            handleBatchResponse(tasks, response.getEntity().getResponseList());
        }
    } catch (Throwable e) {
        if (maybeReadTimeOut(e)) {
            logger.error("It seems to be a socket read timeout exception, it will retry later. if it continues to happen and some eureka node occupied all the cpu time, you should set property 'eureka.server.peer-node-read-timeout-ms' to a bigger value", e);
        	//read timeout exception is more Congestion then TransientError, return Congestion for longer delay 
            return ProcessingResult.Congestion;
        } else if (isNetworkConnectException(e)) {
            logNetworkErrorSample(null, e);
            return ProcessingResult.TransientError;
        } else {
            logger.error("Not re-trying this exception because it does not seem to be a network exception", e);
            return ProcessingResult.PermanentError;
        }
    }
    return ProcessingResult.Success;
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39

批处理任务，将一组任务一次性发送到远程进行处理

• 根据task集合创建 ReplicationList
• 调用批量同步接口将同步集合发送到远端节点同步数据，即调用 rest API
• 分析远程返回结果

三、总结

数据同步功能主要由 PeerEurekaNodes 与 PeerEurekaNode 类实现。

1、集群节点初始化：PeerEurekaNodes#start() 方法

• 初始化定时任务线程池
• 初始化集群节点信息 updatePeerEurekaNodes 方法
• 初始化固定周期（默认10分钟，可配置）更新集群节点信息的任务的线程
• 通过定时任务，线程池定时执行更新集群节点线程

2、更新集群节点信息：updatePeerEurekaNodes(resolvePeerUrls()) 方法

• 校验传入的 URL 集合是否需要更新
• 移除新 URL 集合中没有的旧节点并关闭节点
• 创建旧节点集合中没有的新 URL 节点，通过 createPeerEurekaNode(peerUrl) 方法
• 重新赋值节点集合以及节点URL集合完成节点的更新

resolvePeerUrls() 方法，实际上就是解析配置文件中的 eureka.serviceUrl 前缀的配置，并动态监听配置的更新。

3、创建节点信息：createPeerEurekaNode(String peerEurekaNodeUrl) 方法

4、集群节点数据同步：PeerEurekaNode 方法

• 创建数据同步的任务处理器 ReplicationTaskProcessor
• 创建批处理任务分发器
• 创建单任务分发器

PeerEurekaNode 可以创建的同步任务：register、cancel、heartbeat、statusUpdate、deleteStatusOverride

5、集群节点数据同步任务处理：ReplicationTaskProcessor.process() 完成

• 单任务处理
• 批量任务处理