SpringCloud Eureka Server 源码

Myth丶恋晨 2022-05-27 06:59 214阅读 0赞

Server端，又可以称之为注册中心，其作用简单的说，可以总结为以下几点：

1.  接收客户端的注册信息，把这些信息存起来
2.  供其它客户端查询注册信息
3.  同步其它节点的注册信息
4.  剔除失效的客户端实例
5.  自我保护

可是具体是怎么实现的呢？  
第一、二点为对外（客户端）提供的接口，就不详细说了，看看剩下的3个

## 启动 ##

EurekaServerInitializerConfiguration实现了ServletContextAware和SmartLifecycle接口，通过SmartLifecycle接口，Spring容器初始化该bean时会调用相应生命周期方法start（）

@Override
        public void start() {
            new Thread(new Runnable() {
                @Override
                public void run() {
                    try {
                        //TODO: is this class even needed now?
                        eurekaServerBootstrap.contextInitialized(EurekaServerInitializerConfiguration.this.servletContext);
                        log.info("Started Eureka Server");
    
                        publish(new EurekaRegistryAvailableEvent(getEurekaServerConfig()));
                        EurekaServerInitializerConfiguration.this.running = true;
                        publish(new EurekaServerStartedEvent(getEurekaServerConfig()));
                    }
                    catch (Exception ex) {
                        // Help!
                        log.error("Could not initialize Eureka servlet context", ex);
                    }
                }
            }).start();
        }

1.  初始化上下文-eurekaServerBootstrap.contextInitialized（）
2.  发布了“可以注册事件”和“服务启动事件”

EurekaServerBootstrap的初始化方法contextInitialized：

public void contextInitialized(ServletContext context) {
            try {
                initEurekaEnvironment();
                initEurekaServerContext();
    
                context.setAttribute(EurekaServerContext.class.getName(), this.serverContext);
            }
            catch (Throwable e) {
                log.error("Cannot bootstrap eureka server :", e);
                throw new RuntimeException("Cannot bootstrap eureka server :", e);
            }
        }

第一行初始化环境，第二行初始化上下文，我们来看一下initEurekaServerContext。

protected void initEurekaServerContext() throws Exception {
            // For backward compatibility
            JsonXStream.getInstance().registerConverter(new V1AwareInstanceInfoConverter(),
                    XStream.PRIORITY_VERY_HIGH);
            XmlXStream.getInstance().registerConverter(new V1AwareInstanceInfoConverter(),
                    XStream.PRIORITY_VERY_HIGH);
    
            if (isAws(this.applicationInfoManager.getInfo())) {
                this.awsBinder = new AwsBinderDelegate(this.eurekaServerConfig,
                        this.eurekaClientConfig, this.registry, this.applicationInfoManager);
                this.awsBinder.start();
            }
    
            EurekaServerContextHolder.initialize(this.serverContext);
    
            log.info("Initialized server context");
    
            // Copy registry from neighboring eureka node
            int registryCount = this.registry.syncUp();
            this.registry.openForTraffic(this.applicationInfoManager, registryCount);
    
            // Register all monitoring statistics.
            EurekaMonitors.registerAllStats();
        }

这里做了2件事：

1.  向其它节点同步数据：this.registry.syncUp();
2.  注册监视器：EurekaMonitors.registerAllStats();

/** * Populates the registry information from a peer eureka node. This * operation fails over to other nodes until the list is exhausted if the * communication fails. */
        @Override
        public int syncUp() {
            // Copy entire entry from neighboring DS node
            int count = 0;
    
            for (int i = 0; ((i < serverConfig.getRegistrySyncRetries()) && (count == 0)); i++) {
                if (i > 0) {
                    try {
                        Thread.sleep(serverConfig.getRegistrySyncRetryWaitMs());
                    } catch (InterruptedException e) {
                        logger.warn("Interrupted during registry transfer..");
                        break;
                    }
                }
                Applications apps = eurekaClient.getApplications();
                for (Application app : apps.getRegisteredApplications()) {
                    for (InstanceInfo instance : app.getInstances()) {
                        try {
                            if (isRegisterable(instance)) {
                                register(instance, instance.getLeaseInfo().getDurationInSecs(), true);
                                count++;
                            }
                        } catch (Throwable t) {
                            logger.error("During DS init copy", t);
                        }
                    }
                }
            }
            return count;
        }
    
        @Override
        public void openForTraffic(ApplicationInfoManager applicationInfoManager, int count) {
            // Renewals happen every 30 seconds and for a minute it should be a factor of 2.
            this.expectedNumberOfRenewsPerMin = count * 2;
            this.numberOfRenewsPerMinThreshold =
                    (int) (this.expectedNumberOfRenewsPerMin * serverConfig.getRenewalPercentThreshold());
            logger.info("Got " + count + " instances from neighboring DS node");
            logger.info("Renew threshold is: " + numberOfRenewsPerMinThreshold);
            this.startupTime = System.currentTimeMillis();
            if (count > 0) {
                this.peerInstancesTransferEmptyOnStartup = false;
            }
            DataCenterInfo.Name selfName = applicationInfoManager.getInfo().getDataCenterInfo().getName();
            boolean isAws = Name.Amazon == selfName;
            if (isAws && serverConfig.shouldPrimeAwsReplicaConnections()) {
                logger.info("Priming AWS connections for all replicas..");
                primeAwsReplicas(applicationInfoManager);
            }
            logger.info("Changing status to UP");
            applicationInfoManager.setInstanceStatus(InstanceStatus.UP);
            super.postInit();
        }

其中syncUp()方法通过eurekaClient.getApplications();获取集群节点，之后轮询可注册的节点，将信息同步到其它节点。

这里有个配置属性numberRegistrySyncRetries、registrySyncRetryWaitMs。前者为向其它节点注册的次数，后者是失败后再次注册需要等待的时间。默认重试次数为5次。此处保证了至少向一个节点成功同步数据。

openForTraffic()中，最终调用super.postInit()来开启定时任务，来清理过期得客户端。

protected void postInit() {
            renewsLastMin.start();
            if (evictionTaskRef.get() != null) {
                evictionTaskRef.get().cancel();
            }
            evictionTaskRef.set(new EvictionTask());
            evictionTimer.schedule(evictionTaskRef.get(),
                    serverConfig.getEvictionIntervalTimerInMs(),
                    serverConfig.getEvictionIntervalTimerInMs());
        }

renewsLastMin.start() ： 在每个Eureka-Server端都维护着，每分钟的续约数量，续约数量是有一个Long类型的变量  
来存储的，每过一分钟就需要对这个变量进行清0

清理任务是EvictionTask这个类实现的。

class EvictionTask extends TimerTask {
    
            private final AtomicLong lastExecutionNanosRef = new AtomicLong(0l);
    
            @Override
            public void run() {
                try {
                    long compensationTimeMs = getCompensationTimeMs();
                    logger.info("Running the evict task with compensationTime {}ms", compensationTimeMs);
                    evict(compensationTimeMs);
                } catch (Throwable e) {
                    logger.error("Could not run the evict task", e);
                }
            }
    
            /** * compute a compensation time defined as the actual time this task was executed since the prev iteration, * vs the configured amount of time for execution. This is useful for cases where changes in time (due to * clock skew or gc for example) causes the actual eviction task to execute later than the desired time * according to the configured cycle. */
            long getCompensationTimeMs() {
                long currNanos = getCurrentTimeNano();
                long lastNanos = lastExecutionNanosRef.getAndSet(currNanos);
                if (lastNanos == 0l) {
                    return 0l;
                }
    
                long elapsedMs = TimeUnit.NANOSECONDS.toMillis(currNanos - lastNanos);
                long compensationTime = elapsedMs - serverConfig.getEvictionIntervalTimerInMs();
                return compensationTime <= 0l ? 0l : compensationTime;
            }
    
            long getCurrentTimeNano() {  // for testing
                return System.nanoTime();
            }
    
        }
    
    
    public void evict(long additionalLeaseMs) {
            logger.debug("Running the evict task");
    
            if (!isLeaseExpirationEnabled()) {
                logger.debug("DS: lease expiration is currently disabled.");
                return;
            }
    
            // We collect first all expired items, to evict them in random order. For large eviction sets,
            // if we do not that, we might wipe out whole apps before self preservation kicks in. By randomizing it,
            // the impact should be evenly distributed across all applications.
            List<Lease<InstanceInfo>> expiredLeases = new ArrayList<>();
            for (Entry<String, Map<String, Lease<InstanceInfo>>> groupEntry : registry.entrySet()) {
                Map<String, Lease<InstanceInfo>> leaseMap = groupEntry.getValue();
                if (leaseMap != null) {
                    for (Entry<String, Lease<InstanceInfo>> leaseEntry : leaseMap.entrySet()) {
                        Lease<InstanceInfo> lease = leaseEntry.getValue();
                        if (lease.isExpired(additionalLeaseMs) && lease.getHolder() != null) {
                            expiredLeases.add(lease);
                        }
                    }
                }
            }
    
            // To compensate for GC pauses or drifting local time, we need to use current registry size as a base for
            // triggering self-preservation. Without that we would wipe out full registry.
            int registrySize = (int) getLocalRegistrySize();
            int registrySizeThreshold = (int) (registrySize * serverConfig.getRenewalPercentThreshold());
            int evictionLimit = registrySize - registrySizeThreshold;
    
            int toEvict = Math.min(expiredLeases.size(), evictionLimit);
            if (toEvict > 0) {
                logger.info("Evicting {} items (expired={}, evictionLimit={})", toEvict, expiredLeases.size(), evictionLimit);
    
                Random random = new Random(System.currentTimeMillis());
                for (int i = 0; i < toEvict; i++) {
                    // Pick a random item (Knuth shuffle algorithm)
                    int next = i + random.nextInt(expiredLeases.size() - i);
                    Collections.swap(expiredLeases, i, next);
                    Lease<InstanceInfo> lease = expiredLeases.get(i);
    
                    String appName = lease.getHolder().getAppName();
                    String id = lease.getHolder().getId();
                    EXPIRED.increment();
                    logger.warn("DS: Registry: expired lease for {}/{}", appName, id);
                    internalCancel(appName, id, false);
                }
            }
        }

其中isExpired判断续约是否过期。

public boolean isExpired(long additionalLeaseMs) {
            return (evictionTimestamp > 0 || System.currentTimeMillis() > (lastUpdateTimestamp + duration + additionalLeaseMs));
        }

evictionTimestamp： 为客户端请求的服务下线时间。  
lastUpdateTimestamp：上一次续约修改过的时间。  
duration：续约延长的时间，为客户端注册时的lease-expiration-duration-in-seconds

#### 分批过期机制 ####

从上面可以得知 , 这里有个分批过期的概念，每次最多过期15%的机器，超过15%则不会自动过期，这15%随机选择。

int registrySize = (int) getLocalRegistrySize();
    int registrySizeThreshold = (int) (registrySize * serverConfig.getRenewalPercentThreshold());
    int evictionLimit = registrySize - registrySizeThreshold;

serverConfig.getRenewalPercentThreshold()默认0.85。

#### 自我保护 ####

如果开启自我保护，则认为不允许过期，以上过期机制将无效。  
evict方法中代码如下：

if (!isLeaseExpirationEnabled()) {
        logger.debug("DS: lease expiration is currently disabled.");
        return;
    }