EurekaServer集群同步源码解析

EurekaServer集群同步源码解析

本文主要基于源码分析server集群间的数据同步，和部分server提供给client端续约、注册信息刷新等REST的接口进行简要分析。

集群同步

Eureka-Server 集群不区分主从节点或者 Primary & Secondary 节点，所有节点相同角色( 也就是没有角色 )，完全对等。

Eureka-Client 可以向任意 Eureka-Client 发起任意读写 *** 作，Eureka-Server 将 *** 作复制到另外的 Eureka-Server 以达到最终一致性。

通过上篇文章对EurekaServer 的分析，可以知道server初始化时，会全量获取注册信息，并启动定时任务，以一定间隔时间(默认30秒)向其他server发起增量实例获取，来保证当前server的注册信息是完整的。

那么当服务向某一个server发起注册、续约、下线等请求时，server间是如何进行数据同步的呢？下面根据server的注册接口来看一下：

client向server发起注册请求，注册接口首先是必要的参数校验，当实例的数据信息是UniqueIdentifier的实现是再校验下实例数据信息，然后通过registry.register(info, "true".equals(isReplication));去注册client实例。

@POST
    @Consumes({"application/json", "application/xml"})
    public Response addInstance(InstanceInfo info,
                                @HeaderParam(PeerEurekaNode.HEADER_REPLICATION) String isReplication) {
        logger.debug("Registering instance {} (replication={})", info.getId(), isReplication);
        // validate that the instanceinfo contains all the necessary required fields
        // 参数校验
        if (isBlank(info.getId())) {
            return Response.status(400).entity("Missing instanceId").build();
        } else if (isBlank(info.getHostName())) {
            return Response.status(400).entity("Missing hostname").build();
        } else if (isBlank(info.getIPAddr())) {
            return Response.status(400).entity("Missing ip address").build();
        } else if (isBlank(info.getAppName())) {
            return Response.status(400).entity("Missing appName").build();
        } else if (!appName.equals(info.getAppName())) {
            return Response.status(400).entity("Mismatched appName, expecting " + appName + " but was " + info.getAppName()).build();
        } else if (info.getDataCenterInfo() == null) {
            return Response.status(400).entity("Missing dataCenterInfo").build();
        } else if (info.getDataCenterInfo().getName() == null) {
            return Response.status(400).entity("Missing dataCenterInfo Name").build();
        }
        // 处理客户可能注册坏的DataCenterInfo缺少数据的情况(没有ID的情况)
        // handle cases where clients may be registering with bad DataCenterInfo with missing data
        DataCenterInfo dataCenterInfo = info.getDataCenterInfo();
        if (dataCenterInfo instanceof UniqueIdentifier) {
            String dataCenterInfoId = ((UniqueIdentifier) dataCenterInfo).getId();
            if (isBlank(dataCenterInfoId)) {
                boolean experimental = "true".equalsIgnoreCase(serverConfig.getExperimental("registration.validation.dataCenterInfoId"));
                if (experimental) {
                    String entity = "DataCenterInfo of type " + dataCenterInfo.getClass() + " must contain a valid id";
                    return Response.status(400).entity(entity).build();
                } else if (dataCenterInfo instanceof AmazonInfo) {
                    AmazonInfo amazonInfo = (AmazonInfo) dataCenterInfo;
                    String effectiveId = amazonInfo.get(AmazonInfo.metaDataKey.instanceId);
                    if (effectiveId == null) {
                        amazonInfo.getmetadata().put(AmazonInfo.metaDataKey.instanceId.getName(), info.getId());
                    }
                } else {
                    logger.warn("Registering DataCenterInfo of type {} without an appropriate id", dataCenterInfo.getClass());
                }
            }
        }

        registry.register(info, "true".equals(isReplication));
        return Response.status(204).build();  // 204 to be backwards compatible
    }

当实例的续约信息存在，且大于0时，使用实例的过期清除时间，否则使用默认的90秒清除(通常client这个属性也是会使用默认的90秒)。 接着是通过父类的方法去注册client实例，然后根据当前是否是复制请求来决定是否对当前请求进行集群同步(复制请求是指：server集群同步时发起的请求，实例数据不变，只是传入的isReplication为true，告诉其他server当前请求是集群同步，无需再次对其他server发起同步 *** 作)。super.register()这篇已经分析过，感兴趣的可以看下。 EurekaServer基于源码的启动流程，下面主要看下集群同步的代码 

@Override
    public void register(final InstanceInfo info, final boolean isReplication) {
        // 获取过期清除client的时间间隔，默认值90秒
        int leaseDuration = Lease.DEFAULT_DURATION_IN_SECS;
        if (info.getLeaseInfo() != null && info.getLeaseInfo().getDurationInSecs() > 0) {
            leaseDuration = info.getLeaseInfo().getDurationInSecs();
        }
        // 注册应用实例信息
        super.register(info, leaseDuration, isReplication);
        // Eureka-Server 复制（集群同步）
        replicateToPeers(Action.Register, info.getAppName(), info.getId(), info, null, isReplication);
    }

1、若当前server的集群节点是空或者当前是server间同步的请求，则直接return，不对当前请求进行请求转发。

 2、遍历当前所有eureka server的节点，忽略当前server节点，其余节点丢到线程池中执行同步 *** 作。

private void replicateToPeers(Action action, String appName, String id,
                                  InstanceInfo info ,
                                  InstanceStatus newStatus , boolean isReplication) {
        Stopwatch tracer = action.getTimer().start();
        try {
            if (isReplication) {
                numberOfReplicationsLastMin.increment();
            }
            // If it is a replication already, do not replicate again as this will create a poison replication
            // 若集群为空或者是复制请求，则无需集群同步
            if (peerEurekaNodes == Collections.EMPTY_LIST || isReplication) {
                return;
            }

            for (final PeerEurekaNode node : peerEurekaNodes.getPeerEurekaNodes()) {
                // If the url represents this host, do not replicate to yourself.
                // 自身服务，忽略
                if (peerEurekaNodes.isThisMyUrl(node.getServiceUrl())) {
                    continue;
                }
                // 循环丢到线程池中执行同步 *** 作
                replicateInstanceActionsToPeers(action, appName, id, info, newStatus, node);
            }
        } finally {
            tracer.stop();
        }
    }

这里可以看到针对不同的client请求 *** 作 ，server节点需要进行相应的处理，接着看node.register(info)的流程。

private void replicateInstanceActionsToPeers(Action action, String appName,
                                                 String id, InstanceInfo info, InstanceStatus newStatus,
                                                 PeerEurekaNode node) {
        try {
            InstanceInfo infoFromRegistry;
            CurrentRequestVersion.set(Version.V2);
            switch (action) {
                // taskId：ActionType#appName/instanceId 相同应用实例的相同同步 *** 作使用相同任务编号
                case Cancel:
                    node.cancel(appName, id);
                    break;
                case Heartbeat:
                    InstanceStatus overriddenStatus = overriddenInstanceStatusMap.get(id);
                    infoFromRegistry = getInstanceByAppAndId(appName, id, false);
                    node.heartbeat(appName, id, infoFromRegistry, overriddenStatus, false);
                    break;
                case Register:
                    node.register(info);
                    break;
                case StatusUpdate:
                    infoFromRegistry = getInstanceByAppAndId(appName, id, false);
                    node.statusUpdate(appName, id, newStatus, infoFromRegistry);
                    break;
                case DeleteStatusOverride:
                    infoFromRegistry = getInstanceByAppAndId(appName, id, false);
                    node.deleteStatusOverride(appName, id, infoFromRegistry);
                    break;
            }
        } catch (Throwable t) {
            logger.error("Cannot replicate information to {} for action {}", node.getServiceUrl(), action.name(), t);
        } finally {
            CurrentRequestVersion.remove();
        }
    }

注册同步 

集群同步注册时，首先是创建同步的超时时间为当前时间+30秒，并根据实例信息创建task信息放入到批量任务分发器去 *** 作，其中taskId是根据 requestType + '#' + appName + '/' + id来生成任务ID。

TaskDispatcher的process方法是通过调用任务接收执行器(AcceptorExecutor)的process方法将任务ID、任务、过期时间放入接收队列中。

public void register(final InstanceInfo info) throws Exception {
        // 当前时间+30（续约间隔)，超时，则发送心跳时，同步
        long expiryTime = System.currentTimeMillis() + getLeaseRenewalOf(info);
        batchingDispatcher.process(
                taskId("register", info),
                new InstanceReplicationTask(targetHost, Action.Register, info, null, true) {
                    public EurekaHttpResponse execute() {
                        return replicationClient.register(info);
                    }
                },
                expiryTime
        );
    }
// batchingDispatcher
 return new TaskDispatcher() {
            
            @Override
            public void process(ID id, T task, long expiryTime) {
                acceptorExecutor.process(id, task, expiryTime);
            }

            @Override
            public void shutdown() {
                acceptorExecutor.shutdown();
                taskExecutor.shutdown();
            }
        };
// acceptorExecutor
void process(ID id, T task, long expiryTime) {
        acceptorQueue.add(new TaskHolder(id, task, expiryTime));
        acceptedTasks++;
    }

放入队列后是如何 *** 作同步的呢？AcceptorExecutor又是在何时创建？

批量任务的定时任务的创建、执行时机

首先该类初始化时，将AcceptorRunner作为具体的实现，已经创建并执行acceptorThread的start()方法，这个类的构造器通过向上寻找可以发现是updatePeerEurekaNodes()时createPeerEurekaNode()，创建server节点时，就启动好该线程池，下面看下AcceptorRunner的run()方法做了些什么

AcceptorExecutor(String id,
                     int maxBufferSize,
                     int maxBatchingSize,
                     long maxBatchingDelay,
                     long congestionRetryDelayMs,
                     long networkFailureRetryMs) {
        this.id = id;
        this.maxBufferSize = maxBufferSize;
        this.maxBatchingSize = maxBatchingSize;
        this.maxBatchingDelay = maxBatchingDelay;
        // 创建网络通信整形器
        this.trafficShaper = new TrafficShaper(congestionRetryDelayMs, networkFailureRetryMs);

        
        ThreadGroup threadGroup = new ThreadGroup("eurekaTaskExecutors");
        this.acceptorThread = new Thread(threadGroup, new AcceptorRunner(), "TaskAcceptor-" + id);
        this.acceptorThread.setDaemon(true);
        this.acceptorThread.start();

        final double[] percentiles = {50.0, 95.0, 99.0, 99.5};
        final StatsConfig statsConfig = new StatsConfig.Builder()
                .withSampleSize(1000)
                .withPercentiles(percentiles)
                .withPublishStdDev(true)
                .build();
        final MonitorConfig config = MonitorConfig.builder(METRIC_REPLICATION_PREFIX + "batchSize").build();
        this.batchSizeMetric = new StatsTimer(config, statsConfig);
        try {
            Monitors.registerObject(id, this);
        } catch (Throwable e) {
            logger.warn("Cannot register servo monitor for this object", e);
        }
    }

只要当前线程未关闭(类似后台线程，死循环)，

1、先处理刚添加进的任务的队列

2、计算调度时间

3、调度任务，执行同步

4、若待执行队列为空，休眠10毫秒，避免资源浪费。

    class AcceptorRunner implements Runnable {
        @Override
        public void run() {
            long scheduleTime = 0;
            // 只要没有关闭，就一直执行
            while (!isShutdown.get()) {
                try {
                    drainInputQueues();

                    int totalItems = processingOrder.size();

                    // 调度时间小于当前时间时，计算调度时间
                    long now = System.currentTimeMillis();
                    if (scheduleTime < now) {
                        scheduleTime = now + trafficShaper.transmissionDelay();
                    }

                    // 调度
                    if (scheduleTime <= now) {
                        
                        assignBatchWork();
                        
                        assignSingleItemWork();
                    }

                    // If no worker is requesting data or there is a delay injected by the traffic shaper,
                    // sleep for some time to avoid tight loop.
                    
                    if (totalItems == processingOrder.size()) {
                        Thread.sleep(10);
                    }
                } catch (InterruptedException ex) {
                    // Ignore
                } catch (Throwable e) {
                    // Safe-guard, so we never exit this loop in an uncontrolled way.
                    logger.warn("Discovery AcceptorThread error", e);
                }
            }
        }

处理队列中的任务

drainReprocessQueue()：若重试队列中不为空、且待执行队列未满，则优先d出最新加入到队列的任务，去执行流程，当该任务已过期，或者map中已经包含该任务Id就不做任何处理，否则加入到待处理队列、map中。

注：因为是优先处理新任务，所以即使出现taskId重复的情况，map记录的是最新的需要重试的task信息，其他的任务被队列d出后，进行数据的记录，便于被监控观察，无其他的处理。

若待执行队列达到最大值(默认值10000)，直接清空重试队列，防止任务积压过多。

private void drainInputQueues() throws InterruptedException {
            do {
                // 处理完重新执行任务
                drainReprocessQueue();
                // 处理完接收任务
                drainAcceptorQueue();

                if (isShutdown.get()) {
                    break;
                }

                
                // If all queues are empty, block for a while on the acceptor queue
                if (reprocessQueue.isEmpty() && acceptorQueue.isEmpty() && pendingTasks.isEmpty()) {
                    TaskHolder taskHolder = acceptorQueue.poll(10, TimeUnit.MILLISECONDS);
                    if (taskHolder != null) {
                        appendTaskHolder(taskHolder);
                    }
                }
            } while (!reprocessQueue.isEmpty() || !acceptorQueue.isEmpty() || pendingTasks.isEmpty());
        }


// drainReprocessQueue
private void drainReprocessQueue() {
            long now = System.currentTimeMillis();
            while (!reprocessQueue.isEmpty() && !isFull()) {
                // 优先拿较新的任务
                TaskHolder taskHolder = reprocessQueue.pollLast();
                ID id = taskHolder.getId();
                if (taskHolder.getExpiryTime() <= now) {// 过期
                    expiredTasks++;
                } else if (pendingTasks.containsKey(id)) {// 已存在
                    overriddenTasks++;
                } else {// 加入待执行任务映射，加入待处理队列的头部
                    pendingTasks.put(id, taskHolder);
                    processingOrder.addFirst(id);
                }
            }
            
            if (isFull()) {
                queueOverflows += reprocessQueue.size();
                reprocessQueue.clear();
            }
        }

接着处理接收队列中的任务，当不为空时，根据先进先出规则d出任务，若待执行队列已满，移除待处理队伍d出的数据，根据队列先进先出的规律，当待处理队列满了，待重试任务优先被删除，将接收队列的任务添加到待执行队列中。

 private void drainAcceptorQueue() {
            while (!acceptorQueue.isEmpty()) {// 循环，直到接收队列为空
                appendTaskHolder(acceptorQueue.poll());
            }
        }

private void appendTaskHolder(TaskHolder taskHolder) {
            // 如果待执行队列已满，移除待处理队列，放弃较早的任务
            if (isFull()) {
                pendingTasks.remove(processingOrder.poll());
                queueOverflows++;
            }
            // 添加到待执行队列
            TaskHolder previousTask = pendingTasks.put(taskHolder.getId(), taskHolder);
            if (previousTask == null) {
                processingOrder.add(taskHolder.getId());
            } else {
                overriddenTasks++;
            }
        }

若处理完任务后，发现待重试、接收、待执行队列都为空，休眠10毫秒，再扫描，避免资源浪费。

计算调度时间

处理完队列后，会计算调度时间，当拥挤、瞬时错误的时间戳时，会延迟一定时再对server发起请求。

当调用时间小于当前时间时，执行调用批量、单个任务（类似，只分析批量）

若队列大于等于最大值，或者队列最先添加的任务提交时间大于500毫秒，执行后续逻辑。依次d出任务数据并以list添加到批量执行队列中

void assignBatchWork() {
            if (hasEnoughTasksForNextBatch()) {
                
                if (batchWorkRequests.tryAcquire(1)) {
                    long now = System.currentTimeMillis();
                    int len = Math.min(maxBatchingSize, processingOrder.size());
                    // 从待执行队列中获取任务，并将待执行 map中该任务 移除，当未到达任务的过期时间放入
                    List> holders = new ArrayList<>(len);
                    while (holders.size() < len && !processingOrder.isEmpty()) {
                        ID id = processingOrder.poll();
                        TaskHolder holder = pendingTasks.remove(id);
                        if (holder.getExpiryTime() > now) {
                            holders.add(holder);
                        } else {
                            expiredTasks++;
                        }
                    }
                    if (holders.isEmpty()) {
                        // 未调度到批量任务，释放请求信号量
                        batchWorkRequests.release();
                    } else {
                        // 加入到批量执行队列中
                        batchSizeMetric.record(holders.size(), TimeUnit.MILLISECONDS);
                        batchWorkQueue.add(holders);
                    }
                }
            }
        }

TaskDispatchers初始化时，创建批量任务执行器、并启动，若线程未关闭，死循环获取批量任务队列中的一个批量任务，若不存在休眠一秒，得到批量任务后，发起批量请求，若是成功、永久错误不进行重试处理，当是限流、网络原因导致的拥挤、瞬时错误会添加到重试队列中，并更新上一次拥挤、瞬时的时间戳，

static class BatchWorkerRunnable extends WorkerRunnable {

        BatchWorkerRunnable(String workerName,
                            AtomicBoolean isShutdown,
                            TaskExecutorMetrics metrics,
                            TaskProcessor processor,
                            AcceptorExecutor acceptorExecutor) {
            super(workerName, isShutdown, metrics, processor, acceptorExecutor);
        }

        @Override
        public void run() {
            try {
                while (!isShutdown.get()) {
                    List> holders = getWork();

                    // 监控相关，无视
                    metrics.registerExpiryTimes(holders);
                    // 得到taskList，可用lamda表达式，有些low
                    List tasks = getTasksOf(holders);
                    // 调用处理器执行任务
                    ProcessingResult result = processor.process(tasks);
                    switch (result) {
                        case Success:
                            break;
                            // 拥挤、瞬时错误，放入重新执行队列，后续再次调用
                        case Congestion:
                        case TransientError:
                            taskDispatcher.reprocess(holders, result);
                            break;
                        case PermanentError:
                            logger.warn("Discarding {} tasks of {} due to permanent error", holders.size(), workerName);
                    }
                    // 监控，无视
                    metrics.registerTaskResult(result, tasks.size());
                }
            } catch (InterruptedException e) {
                // Ignore
            } catch (Throwable e) {
                // Safe-guard, so we never exit this loop in an uncontrolled way.
                logger.warn("Discovery WorkerThread error", e);
            }
        }

若是Congestion、TransientError异常，则批量任务记录到重试队列中，更新异常时间戳，在任务调用执行时，会进行一定延迟后，再次执行重试队列中的任务(拥挤、网络异常是远程server当前处于不稳定状态，多一段时间后再访问)。

void reprocess(List> holders, ProcessingResult processingResult) {
        // 添加到重新执行队列
        reprocessQueue.addAll(holders);
        replayedTasks += holders.size();
        // 注册失败时间、trafficShaper计算重新执行任务的再次执行延迟时间
        trafficShaper.registerFailure(processingResult);
    }

发起批量同步

下面看下processor.process(tasks)

1、根据任务list创建复制list，向自身服务发起peerreplication/batch/ 一次性将批量( 多个 )的同步 *** 作任务发起请求，这个接口会遍历请求将需同步的实例信息发送出去，并处理好所有请求的响应返回。

2、批量任务返回拥挤、异常错误码时，会重试，若是正常信息码返回，则一次匹配单个任务的执行结果，异常的输出日志，未做重试处理。只有当整个批量任务返回Congestion、TransientError时，才会在一定延迟后(默认30秒)重试。

public ProcessingResult process(List tasks) {
        // 创建批量提交同步 *** 作任务的请求对象
        ReplicationList list = createReplicationListOf(tasks);
        try {
            // 发起批量提交同步 *** 作任务请求
            EurekaHttpResponse response = replicationClient.submitBatchUpdates(list);
            // 响应状态
            int statusCode = response.getStatusCode();
            
            if (!isSuccess(statusCode)) {
                // 状态码 503 ，目前 Eureka-Server 返回 503 的原因是被限流
                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;
    }

场景定制

单个任务在失败时，打印日志，在大多数场景不会有问题，可是再续约请求中，若发生404的异常情况，client服务去续约，但是server注册信息中没有该实例存在，就应该去重新注册。所以在部分场景下，会重写ReplicationTask的handleFailure()。如：

取消实例请求，若404，多输出一行日志，便于分析

@Override
                    public void handleFailure(int statusCode, Object responseEntity) throws Throwable {
                        // 404时，再次输出日志
                        super.handleFailure(statusCode, responseEntity);
                        if (statusCode == 404) {
                            logger.warn("{}: missing entry.", getTaskName());
                        }
                    }

若是404，是因为远程server发现请求的该实例的数据更新时间大于本身缓存，所以让重新发送注册请求到远程服务器中，收到404后会发起注册请求，若同步请求时，远程server发现请求过来的实例的时间戳更新时间戳小于本身注册表的实例信息，则将本身缓存的服务注册信息返回给发起请求的server，发起请求的server得到后，覆盖自身已过期的实例数据。

@Override
            public void handleFailure(int statusCode, Object responseEntity) throws Throwable {
                super.handleFailure(statusCode, responseEntity);
                // 状态码404输出日志，被任务server当前不存在该实例，发起注册当前node的实例的请求（兜底 *** 作，防止网络波动，导致server不存在某个实例）
                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()) {
                    // node的lastDirtyTimestamp小于，则覆盖当前node的实例
                    InstanceInfo peerInstanceInfo = (InstanceInfo) responseEntity;
                    if (peerInstanceInfo != null) {
                        syncInstancesIfTimestampDiffers(appName, id, info, peerInstanceInfo);
                    }
                }
            }

最近看的Eureka的源码整理的差不多了，关于REST风格的接口，后续有时间的话，简单整理下，主要是些业务代码，暂未处理。