Spring Cloud Netflix - Eureka Server源码阅读

Understanding Eureka Peer to Peer communication

Posted by Chris on July 16, 2017

Eureka是一个基于REST并应用于AWS云服务的服务注册中心,并且经历过了Netflix公司的生产考验,绝对是我们值得细心研读的中间件。虽然我们可能并未接触过AWS,但在阅读Eureka之前应该简单地了解一下Amazon EC2中的某些概念,如地区和可用区域,这样能更好地理解Eureka中某些特定的术语。

在阅读本文之前,希望大家都已经对Eureka有简单的认识:

Eureka REST operations

Understanding eureka client server communication

Understanding Eureka Peer to Peer Communication

Register - 注册

Eureka Instance注册的REST入口在com.netflix.eureka.resources.ApplicationResource#addInstance

    /**
     * Registers information about a particular instance for an
     * {@link com.netflix.discovery.shared.Application}.
     *
     * @param info
     *            {@link InstanceInfo} information of the instance.
     * @param isReplication
     *            a header parameter containing information whether this is
     *            replicated from other nodes.
     */
    @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.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();
        }
      
        // handle cases where clients may be registering with bad DataCenterInfo with missing data
        DataCenterInfo dataCenterInfo = info.getDataCenterInfo();
      	// 仅当DataCenterInfo为AmazonInfo实例的时候,其父类有可能是UniqueIdentifier
        if (dataCenterInfo instanceof UniqueIdentifier) {
			// ......
        }
		// *********** 字段校验 END ************
        registry.register(info, "true".equals(isReplication));	// (1)
        return Response.status(204).build();  // 204 to be backwards compatible
    }

真正的注册操作在(1)处,需要注意的是isReplication变量取决于HTTP头x-netflix-discovery-replication的值。继续追踪(1)的调用栈,发现执行注册操作的方法是是com.netflix.eureka.registry.PeerAwareInstanceRegistryImpl#register

注意该方法的javadoc,他告诉了我们一个比较重要的讯息:将InstanceInfo实例信息注册到Eureka并且复制该信息到其他peer。如果当前收到的注册信息是来自其他peer的复制事件,那么将不会将这个注册信息继续复制到其他peer,这个标志位就是上面所述的isReplication

    /**
     * Registers the information about the {@link InstanceInfo} and replicates
     * this information to all peer eureka nodes. If this is replication event
     * from other replica nodes then it is not replicated.
     *
     * @param info
     *            the {@link InstanceInfo} to be registered and replicated.
     * @param isReplication
     *            true if this is a replication event from other replica nodes,
     *            false otherwise.
     */
    @Override
    public void register(final InstanceInfo info, final boolean isReplication) {
      	// 默认租约有效时长为90s
        int leaseDuration = Lease.DEFAULT_DURATION_IN_SECS;
      	// 注册信息里包含则依照注册信息的租约时长
        if (info.getLeaseInfo() != null && info.getLeaseInfo().getDurationInSecs() > 0) {
            leaseDuration = info.getLeaseInfo().getDurationInSecs();
        }
      	// super为AbstractInstanceRegistry
        super.register(info, leaseDuration, isReplication);
      	// 复制到其他peer
        replicateToPeers(Action.Register, info.getAppName(), info.getId(), info, null, isReplication);
    }

我们看到是先获取到租约的有效时长,然后才是真真正正地委托给super执行注册操作super.register(...)并将注册信息复制到其他peer。register方法非常长,我们重点观察一下他的注册表的结构:

private final ConcurrentHashMap<String, Map<String, Lease<InstanceInfo>>> registry

该注册表是一个以app name为key(在Spring Cloud里就是spring.application.name),嵌套Map为value的ConcurrentHashMap结构。其嵌套Map是以Instance ID为key,Lease对象为value的键值结构。这个registry注册表在Eureka Server或SpringBoot Admin的监控面板上以Eureka Service这个角色出现。

    /**
     * Registers a new instance with a given duration.
     *
     * @see com.netflix.eureka.lease.LeaseManager#register(java.lang.Object, int, boolean)
     */
    public void register(InstanceInfo registrant, int leaseDuration, boolean isReplication) {
        try {
            read.lock();
          	// 可以看出registry是一个以info的app name为key的Map结构, 也就是以spring.application.name的大写串为key
            Map<String, Lease<InstanceInfo>> gMap = registry.get(registrant.getAppName());
            REGISTER.increment(isReplication);
            if (gMap == null) {
                final ConcurrentHashMap<String, Lease<InstanceInfo>> gNewMap = new ConcurrentHashMap<String, Lease<InstanceInfo>>();
                gMap = registry.putIfAbsent(registrant.getAppName(), gNewMap);
                if (gMap == null) {
                    gMap = gNewMap;
                }
            }
          	// registry的value的Map结构是以info的id为key,这里的id就是Eureka文档上的Instance ID,给你个例子你就想起是什么东西了:10.8.88.233:config-server:10888
            Lease<InstanceInfo> existingLease = gMap.get(registrant.getId());
            // .......
            Lease<InstanceInfo> lease = new Lease<InstanceInfo>(registrant, leaseDuration);
            if (existingLease != null) {
                lease.setServiceUpTimestamp(existingLease.getServiceUpTimestamp());
            }
            gMap.put(registrant.getId(), lease);
			// .......
        } finally {
            read.unlock();
        }
    }

上面是register(...)中关于registry的大致操作,其中有相当一部分的操作被略去了,如果感兴趣的话可以细致地研究一下。

Renew and Cancel Lease - 续约与取消租约

续约的REST入口在com.netflix.eureka.resources.InstanceResource#renewLease

而取消租约的REST入口在com.netflix.eureka.resources.InstanceResource#cancelLease

两者的基本思想相似,经由InstanceRegistry -> AbstractInstanceRegistry -> PeerAwareInstanceRegistryImpl,其中PeerAwareInstanceRegistryImpl装饰了添加复制信息到其他节点的功能。其中register、renew、cancel、statusUpdate和deleteStatusOverride都会将其信息复制到其他节点。

Fetch Registry - 获取注册信息

获取所有Eureka Instance的注册信息,com.netflix.eureka.resources.ApplicationsResource#getContainers,其注册信息由ResponseCacheImpl缓存,缓存的过期时间在其构造函数中由EurekaServerConfig.getResponseCacheUpdateIntervalMs()所控制,默认缓存时间为30s。而差量注册信息在Server端会保存得更为长一些(大约3分钟),因此获取的差量可能会重复返回相同的实例。Eureka Client会自动处理这些重复信息。

Evcition

Eureke Server定期进行失效节点的清理,执行该任务的定时器被定义在com.netflix.eureka.registry.AbstractInstanceRegistry#evictionTimer,真正的任务是由他的内部类AbstractInstanceRegistry#EvictionTask所执行,默认为每60s执行一次清理任务,其执行间隔由EurekaServerConfig#getEvictionIntervalTimerInMs[eureka.server.eviction-interval-timer-in-ms]所决定。

回顾一下上面刚说完的注册流程,在PeerAwareInstanceRegistryImpl#register里面特别指出了默认的租约时长为90s[eureka.Instance.lease-expiration-duration-in-seconds],即如果90s后都没有收到特定的Eureka Instance的Heartbeats,则会认为这个Instance已经失效(Instance在正常情况下默认每隔30s发送一个Heartbeats[eureka.Instance.lease-renewal-interval-in-seconds],对以上两个默认值有疑问的可以翻阅LeaseInfo),EvictionTask则会把这个Instance纳入清理的范围。我们看看EvictionTask的清理代码是怎么写的。

    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.
      	// (2) 下面的for循环就是把registry中所有的Lease提取到局部变量expiredLeases
        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());	// (3)
        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);
            }
        }
    }

(2)中把本地的registry中的租约信息全部提取出来,并在(3)通过serverConfig.getRenewalPercentThreshold()[eureka.server.renewal-percent-threshold,默认85%]计算出一个最大可剔除的阈值evictionLimit

新增Peer Node时的初始化

在有多个Eureka Server的情况下,每个Eureka Server之间是如何发现对方的呢?

通过调试之后,我们根据调用链从下往上追溯,其初始入口为org.springframework.cloud.netflix.eureka.server.EurekaServerBootstrap#contextInitialized

	public void contextInitialized(ServletContext context) {
		try {
			initEurekaEnvironment();
			initEurekaServerContext();	// (4)

			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);
		}
	}

由下个入口(4)最终可以定位到方法com.netflix.eureka.registry.PeerAwareInstanceRegistryImpl#syncUp,从对应的javadoc上我们可以知道该方法从peer eureka节点往自己填充注册表信息。 如果操作失败则此同步操作将failover到其他节点,直到遍历完列表(service urls)为止。该方法与普通的Eureka Client注册到Eureka Server不同的一点是,其标志位isReplication为true,如果不记得这是什么作用的话可以翻阅到上面的Register - 注册小节。

Peer Node信息的定时更新

首先我们看Eureka Server的上下文实体中的方法com.netflix.eureka.DefaultEurekaServerContext#initialize

    @PostConstruct
    @Override
    public void initialize() throws Exception {
        logger.info("Initializing ...");
        peerEurekaNodes.start();	// (5)
        registry.init(peerEurekaNodes);
        logger.info("Initialized");
    }

该方法明确指出这是一个Spring Bean,在构建Bean完成后执行此方法,继续追踪(5)

    public void start() {
        taskExecutor = Executors.newSingleThreadScheduledExecutor(
                new ThreadFactory() {
                    @Override
                    public Thread newThread(Runnable r) {
                        Thread thread = new Thread(r, "Eureka-PeerNodesUpdater");
                        thread.setDaemon(true);
                        return thread;
                    }
                }
        );
        try {
            updatePeerEurekaNodes(resolvePeerUrls());	// (6)
            Runnable peersUpdateTask = new Runnable() {	// (7)
                @Override
                public void run() {
                    try {
                        updatePeerEurekaNodes(resolvePeerUrls());	// (6)
                    } catch (Throwable e) {
                        logger.error("Cannot update the replica Nodes", e);
                    }

                }
            };
            taskExecutor.scheduleWithFixedDelay(
                    peersUpdateTask,	// (7)
                    serverConfig.getPeerEurekaNodesUpdateIntervalMs(),	// (8)
                    serverConfig.getPeerEurekaNodesUpdateIntervalMs(),
                    TimeUnit.MILLISECONDS
            );
        } catch (Exception e) {
            throw new IllegalStateException(e);
        }
        for (PeerEurekaNode node : peerEurekaNodes) {
            logger.info("Replica node URL:  " + node.getServiceUrl());
        }
    }

上面这段代码很清晰地告诉我们在启动Eureka Server的时候就会调用updatePeerEurekaNodes(...)更新peer的状态,并封装为一个Runnable进行周期性更新。这个定时时间由serverConfig.getPeerEurekaNodesUpdateIntervalMs()[eureka.server.peer-eureka-nodes-update-interval-ms]所控制,默认值为600s,即10min。一直经由EndpointUtils#getDiscoveryServiceUrlsEndpointUtils#getServiceUrlsFromConfigEurekaClientConfigBean#getEurekaServerServiceUrls获得对应zone的service urls,如有需要可以覆盖上述getEurekaServerServiceUrls方法以动态获取service urls,而不是选择Spring Cloud默认从properties文件读取。

Self Preservation - 自我保护

当新增Eureka Server时,他会先尝试从其他Peer上获取所有Eureka Instance的注册信息。如果在获取时出现问题,该Eureka Server会在放弃之前尝试在其他Peer上获取注册信息。如果这个Eureka Server成功获取到所有Instance的注册信息,那么他就会根据所获取到的注册信息设置应该接收到的续约阈值。如果在任何时候续约的阈值低于所设定的值(在15分钟[eureka.server.renewal-threshold-update-interval-ms]内低于85%[eureka.server.renewal-percent-threshold]),则该Eureka Server会出于保护当前注册列表的目的而停止将任何Instance进行过期处理。

在Netflix中上述保护措施被成为自我保护模式,主要是用于Eureka Server与Eureka Client存在网络分区情况下的场景。在这种情况下,Eureka Server尝试保护其已有的实例信息,但如果出现大规模的网络分区时,相应的Eureka Client会获取到大量无法响应的服务。所以,Eureka Client必须确保对于一些不存在或者无法响应的Eureka Instance具备更加弹性的应对策略,例如快速超时并尝试其他实例。

在网络分区出现时可能会发生以下几种情况:

  • Peer之间的心跳可能会失败,某Eureka Server检测到这种情况并为了保护当前的注册列表而进入了自我保护模式。新的注册可能发生在某些孤立的Eureka Server上,某些Eureka Client可能会拥有新的注册列表,而另外一些则可能没有(不同的实例视图)。
  • 当网络恢复到稳定状态后,Eureka Server会进行自我修复。当Peer能正常通信之后注册信息会被重新同步。
  • 最重要的一点是,在网络中断期间Eureka Server应该更距弹性,但在这段期间Eureka Client可能会有不同的实例视图。