SpringCloud--Eureka--原理及源码解析

一、Eureka的基础架构及服务治理机制

　　Eureka服务治理的基础架构包含三个核心：服务注册中心、服务提供者、服务消费者。其中服务注册中心，即Eureka提供的服务端，提供服务注册和发现的功能；服务提供者，即将自己的服务注册到注册中心；服务的消费者，从注册中心获取服务列表，从而使消费者知道到何处调用服务，服务消费可以使用Ribbon、Feign等。

　　1、服务提供者：

　　　　（1）服务注册：服务提供者在项目启动时，会通过发送REST请求的方式将自己注册到eureka server上，同时带上一些自己的元数据，Eureka Server收到请求后，将元数据存储在一个双层map中，第一层的key是服务名称，第二层的key是具体的服务实例。

　　　　（2）服务同步：如果A项目将服务注册到了M注册中心，B项目将服务注册到N注册中心，但是如果M项目和N项目开启了可以注册为服务的配置，那么当A项目将服务注册到M注册中心时，M注册中心会将请求转发到N注册中心，以保证两个注册中心副本中服务同步。

　　　　（3）服务续约：在注册完服务后，服务提供者会维护一个心跳来持续告诉注册中心其还活着，以防止注册中心的剔除任务将该服务实例从服务列表中删除。

　　　　关于心跳频率与剔除任务认为服务失效时间的配置参数如下所示（配置值均为默认值）：

eureka:
  instance:
    # 心跳检测频率
    lease-renewal-interval-in-seconds: 30 # 服务失效时间 lease-expiration-duration-in-seconds: 90

　　2、服务消费者：

　　　　（1）获取服务：当启动服务消费者项目时，会向注册中心发送一个REST请求来获取注册中心上注册的服务清单。为了性能的考虑，注册中心自己维护了一份只读的注册服务清单，每30秒更新一次，要调整注册中心中注册服务清单更新频率，可以使用如下参数进行设置（下面示例为默认值），同时，由于获取服务是服务消费的基础，因此需要保证eureka.client.fetch-registry为true

eureka:
  client:
    registry-fetch-interval-seconds: 30 fetch-registry: true

　　　　（2）服务调用：服务消费者在获取到服务提供清单后，会根据服务名获得具体的实例名和该实例的元数据，然后客户端可以根据自己需要，选择调用哪个实例，在上述代码样例中，我们使用的是Ribbon来做负载均衡，而ribbon默认采用轮询的方式进行调用，从而实现客户端的负载。对于访问实例的选择，Eureka中有Region和Zone的概念，一个Region中可以包含多个Zone，一个客户端会被注册到一个Zone中，所以一个客户端只对应一个Zone和一个Region，在服务调用时，优先访问处于同一个Zone中的服务提供者，若访问不到，再访问其他Zone中的服务提供者。

　　　　（3）服务下线：当客户端实例进行正常的关闭操作时，它会触发一个服务下线的REST请求给注册中心，告诉注册中心其要下线，注册中心收到请求后，将该服务状态置为下线，并把该事件传播出去。

　　3、服务注册中心

　　　　（1）失效剔除：有时服务实例并不会正常下线，可能是由于内存溢出、网络故障等原因使得服务不能正常运行，所以注册中心并未收到服务下线的请求。为了剔除该类不可用服务提供者实例，Eureka Server在启动时，会创建一个定时任务，每隔一段时间（默认60秒）将当前清单中超时（默认90秒）没有续约的服务剔除出去。

　　　　（2）自我保护：前面提到过，服务提供者启动后，会维护一个心跳，定时向注册中心发送心跳，告诉注册中心自己还活着。注册中心的运行期间，会统计心跳失败的比例在15分钟内是否低于85%，如果低于85%，注册中心会将该服务的实例保护起来，不让其过期，但是由于在本地测试，所以这个情况非常容易满足（而线上则主要是由于网络不稳定等导致），这就导致在保护期间内，如果服务提供者实例出现问题，那么客户端就会拿到有问题的实例，将会出现调用失败的情况，因此客户端必须要有容错机制，比如说请求重试、断路器等机制。如果我们想关闭自我保护机制，可以使用如下参数。

eureka:
  server:
    enable-self-preservation: false

在我们没有关闭自我保护之前，当我们在之前访问注册中心时：http://localhost:1112/，会看到红色警告（警告内容如下图所示），这就是触发了Eureka Server的自我保护机制。

　　　　 

二、Eureka源码分析

　　1、服务注册中心的加载

　　首先从服务提供者开始看，例如eureka-client项目，我们主要是在主函数上添加了@EnableDiscoveryClient注解，以及在配置文件中添加了注册中心地址等配置信息，那么源码入口就可以从@EnableDiscoveryClient注解开始看，从该注解我们可以了解，主要是开启DiscoveryClient，那么全局搜索DiscoveryClient，可以发现有两个，一个是springCloud提供的接口org.springframework.cloud.client.discovery.DiscoveryClient，另外一个是netflix的实现com.netflix.discovery.DiscoveryClient，可以看下类的依赖关系，SpringCloud提供的接口类依赖关系及 netflix的实现类的类依赖关系分别如以下左右两张图：

 

 

 而org.springframework.cloud.netflix.eureka.EurekaDiscoveryClient又使用了com.netflix.discovery.EurekaClient，所以，总的依赖关系如下：

org.springframework.cloud.client.discovery.DiscoveryClient提供了Springcloud服务注册相关的接口，而org.springframework.cloud.netflix.eureka.EurekaDiscoveryClient是netflix公司对于该接口的实现，而该实现，是包装了netflix公司开源项目中的com.netflix.discovery.EurekaClient接口及实现com.netflix.discovery.DiscoveryClient。

　　可以看调用链，先调用了com.netflix.discovery.DiscoveryClient#getEurekaServiceUrlsFromConfig方法

    /**
     * @deprecated use {@link #getServiceUrlsFromConfig(String, boolean)} instead.
     */
    @Deprecated
    public static List<String> getEurekaServiceUrlsFromConfig(String instanceZone, boolean preferSameZone) { return EndpointUtils.getServiceUrlsFromConfig(staticClientConfig, instanceZone, preferSameZone); }

　　可以看到该方法已经过期，被使用@link到了替代方法com.netflix.discovery.DiscoveryClient#getServiceUrlsFromConfig

    @Deprecated
    @Override
    public List<String> getServiceUrlsFromConfig(String instanceZone, boolean preferSameZone) { return EndpointUtils.getServiceUrlsFromConfig(clientConfig, instanceZone, preferSameZone); }

　　然后，调用到了com.netflix.discovery.endpoint.EndpointUtils#getServiceUrlsFromConfig方法

    public static List<String> getServiceUrlsFromConfig(EurekaClientConfig clientConfig, String instanceZone, boolean preferSameZone) {
        List<String> orderedUrls = new ArrayList<String>(); String region = getRegion(clientConfig); String[] availZones = clientConfig.getAvailabilityZones(clientConfig.getRegion()); if (availZones == null || availZones.length == 0) { availZones = new String[1]; availZones[0] = DEFAULT_ZONE; } logger.debug("The availability zone for the given region {} are {}", region, availZones); int myZoneOffset = getZoneOffset(instanceZone, preferSameZone, availZones); List<String> serviceUrls = clientConfig.getEurekaServerServiceUrls(availZones[myZoneOffset]); if (serviceUrls != null) { orderedUrls.addAll(serviceUrls); } int currentOffset = myZoneOffset == (availZones.length - 1) ? 0 : (myZoneOffset + 1); while (currentOffset != myZoneOffset) { serviceUrls = clientConfig.getEurekaServerServiceUrls(availZones[currentOffset]); if (serviceUrls != null) { orderedUrls.addAll(serviceUrls); } if (currentOffset == (availZones.length - 1)) { currentOffset = 0; } else { currentOffset++; } } if (orderedUrls.size() < 1) { throw new IllegalArgumentException("DiscoveryClient: invalid serviceUrl specified!"); } return orderedUrls; }

　　该方法依次操作：获取项目对应的Region、获取项目对应的Zones数组（如果没有，则取默认Zone）、计算项目对应Zones数组的偏移量、获取注册中心、计算当前偏移量、根据当前偏移量获取注册中心地址并将地址添加在地址集合中、返回地址集合

　　接下来每个步骤单独说明。

　　（1）获取项目对应的Region

    public static String getRegion(EurekaClientConfig clientConfig) {
        String region = clientConfig.getRegion(); if (region == null) { region = DEFAULT_REGION; } region = region.trim().toLowerCase(); return region; }

　　这个没什么可说的，就是从配置文件中获取Region的配置，如果没有，则取默认值，最终将Region转换成大写返回，这里可以使用如下参数设置Region

eureka:
  client:
    region: test-1

　　（2）获取项目对应的Zones数组

    public String[] getAvailabilityZones(String region) {
        String value = this.availabilityZones.get(region); if (value == null) { value = DEFAULT_ZONE; } return value.split(","); }

　　根据region获取配置文件中zone数组，如果没有配置，则取默认值（上面使用的eureka.client.service-url.defaultZone即是默认配置），若要指定Zone，可以使用如下配置：

 

eureka:
  client:
    #client所在zone为availabilityZones的第一个zone,如果未配置,则为defaultZone
    prefer-same-zone-eureka: true region: region1 availability-zones: region1: zone1,zone2,zone3 region2: zone4,zone5,zone6 service-url: zone1: http://localhost:1111/eureka/ zone2: http://localhost:1111/eureka/ zone3: http://localhost:1111/eureka/ zone4: http://localhost:1112/eureka/ zone5: http://localhost:1112/eureka/ zone6: http://localhost:1112/eureka/

 

　　（3）获取注册中心，计算zone数组下标，获得zone，然后根据zone获取注册中心

    public List<String> getEurekaServerServiceUrls(String myZone) {
        String serviceUrls = this.serviceUrl.get(myZone); if (serviceUrls == null || serviceUrls.isEmpty()) { serviceUrls = this.serviceUrl.get(DEFAULT_ZONE); } if (!StringUtils.isEmpty(serviceUrls)) { final String[] serviceUrlsSplit = StringUtils .commaDelimitedListToStringArray(serviceUrls); List<String> eurekaServiceUrls = new ArrayList<>(serviceUrlsSplit.length); for (String eurekaServiceUrl : serviceUrlsSplit) { if (!endsWithSlash(eurekaServiceUrl)) { eurekaServiceUrl += "/"; } eurekaServiceUrls.add(eurekaServiceUrl.trim()); } return eurekaServiceUrls; } return new ArrayList<>(); }

　　获取zone配置的注册中心，如果没有，则取默认的注册中心，然后使用逗号切割后组装成集合返回。

　　当我们在使用Ribbon来实现服务调用时，对于zone的设置可以实现区域亲和性，Ribbon会优先访问属于同一Zone中的服务实例，只有当同一zone中没有可用实例后，才会访问其他zone中的实例。所以通过zone属性的定义，配合实际部署的物理结构，我们就可以有效的设计出针对区域性故障的容错集群。

　　2、服务注册

　　在com.netflix.discovery.DiscoveryClient构造函数中调用了com.netflix.discovery.DiscoveryClient#initScheduledTasks方法

    private void initScheduledTasks() {
        if (clientConfig.shouldFetchRegistry()) { // registry cache refresh timer int registryFetchIntervalSeconds = clientConfig.getRegistryFetchIntervalSeconds(); int expBackOffBound = clientConfig.getCacheRefreshExecutorExponentialBackOffBound(); cacheRefreshTask = new TimedSupervisorTask( "cacheRefresh", scheduler, cacheRefreshExecutor, registryFetchIntervalSeconds, TimeUnit.SECONDS, expBackOffBound, new CacheRefreshThread() ); scheduler.schedule( cacheRefreshTask, registryFetchIntervalSeconds, TimeUnit.SECONDS); } if (clientConfig.shouldRegisterWithEureka()) { int renewalIntervalInSecs = instanceInfo.getLeaseInfo().getRenewalIntervalInSecs(); int expBackOffBound = clientConfig.getHeartbeatExecutorExponentialBackOffBound(); logger.info("Starting heartbeat executor: " + "renew interval is: {}", renewalIntervalInSecs); // Heartbeat timer heartbeatTask = new TimedSupervisorTask( "heartbeat", scheduler, heartbeatExecutor, renewalIntervalInSecs, TimeUnit.SECONDS, expBackOffBound, new HeartbeatThread() ); scheduler.schedule( heartbeatTask, renewalIntervalInSecs, TimeUnit.SECONDS); // InstanceInfo replicator instanceInfoReplicator = new InstanceInfoReplicator( this, instanceInfo, clientConfig.getInstanceInfoReplicationIntervalSeconds(), 2); // burstSize  statusChangeListener = new ApplicationInfoManager.StatusChangeListener() { @Override public String getId() { return "statusChangeListener"; } @Override public void notify(StatusChangeEvent statusChangeEvent) { if (InstanceStatus.DOWN == statusChangeEvent.getStatus() || InstanceStatus.DOWN == statusChangeEvent.getPreviousStatus()) { // log at warn level if DOWN was involved logger.warn("Saw local status change event {}", statusChangeEvent); } else { logger.info("Saw local status change event {}", statusChangeEvent); } instanceInfoReplicator.onDemandUpdate(); } }; if (clientConfig.shouldOnDemandUpdateStatusChange()) { applicationInfoManager.registerStatusChangeListener(statusChangeListener); } instanceInfoReplicator.start(clientConfig.getInitialInstanceInfoReplicationIntervalSeconds()); } else { logger.info("Not registering with Eureka server per configuration"); } }

　　对于该方法，依次错了如下操作：

　　　　判断是否开启服务获取，如果开启，创建一个定时任务，定时刷新服务列表

　　　　判断是否开启服务注册，如果开启，添加一个服务续租定时任务；异步注册服务。

　　接下来我们一一查看操作：

　　（1）判断是否开启服务注册，如果开启，添加一个服务续租定时任务；异步注册服务。

　　这个判断使用的参数是上述的eureka.client.register-with-eureka参数，如果配置为true，创建了一个服务续租的定时任务，还创建了一个异步注册的任务。

　　a、先看异步注册服务类InstanceInfoReplicator，该类实现了Rnnable接口，因此直接看run方法

    public void run() {
        try { discoveryClient.refreshInstanceInfo(); Long dirtyTimestamp = instanceInfo.isDirtyWithTime(); if (dirtyTimestamp != null) { discoveryClient.register(); instanceInfo.unsetIsDirty(dirtyTimestamp); } } catch (Throwable t) { logger.warn("There was a problem with the instance info replicator", t); } finally { Future next = scheduler.schedule(this, replicationIntervalSeconds, TimeUnit.SECONDS); scheduledPeriodicRef.set(next); } }

这里注册的，是discoveryClient.register();这一行代码，查看register方法

    boolean register() throws Throwable {
        logger.info(PREFIX + "{}: registering service...", appPathIdentifier); EurekaHttpResponse<Void> httpResponse; try { httpResponse = eurekaTransport.registrationClient.register(instanceInfo); } catch (Exception e) { logger.warn(PREFIX + "{} - registration failed {}", appPathIdentifier, e.getMessage(), e); throw e; } if (logger.isInfoEnabled()) { logger.info(PREFIX + "{} - registration status: {}", appPathIdentifier, httpResponse.getStatusCode()); } return httpResponse.getStatusCode() == Status.NO_CONTENT.getStatusCode(); }

可以看到，这里使用了http请求将元数据InstanceInfo请求到注册中心。

　　b、然后看服务续租定时任务

　　该任务的执行类是HeartbeatThread，直接看run方法，润方法中调用了com.netflix.discovery.DiscoveryClient#renew方法

    boolean renew() {
        EurekaHttpResponse<InstanceInfo> httpResponse; try { httpResponse = eurekaTransport.registrationClient.sendHeartBeat(instanceInfo.getAppName(), instanceInfo.getId(), instanceInfo, null); logger.debug(PREFIX + "{} - Heartbeat status: {}", appPathIdentifier, httpResponse.getStatusCode()); if (httpResponse.getStatusCode() == Status.NOT_FOUND.getStatusCode()) { REREGISTER_COUNTER.increment(); logger.info(PREFIX + "{} - Re-registering apps/{}", appPathIdentifier, instanceInfo.getAppName()); long timestamp = instanceInfo.setIsDirtyWithTime(); boolean success = register(); if (success) { instanceInfo.unsetIsDirty(timestamp); } return success; } return httpResponse.getStatusCode() == Status.OK.getStatusCode(); } catch (Throwable e) { logger.error(PREFIX + "{} - was unable to send heartbeat!", appPathIdentifier, e); return false; } }

　　可以发现该方法，也是使用http的请求，将appname等信息发送给注册中心。

　　（2）判断是否开启服务获取，如果开启，创建一个定时任务，定时刷新服务列表

　　这个判断对应上述使用的参数eureka.client.fetch-registry，如果开启则创建一个定时任务，该定时任务的执行频率等都是使用参数配置的（参数内容不再单独说明，看属性名称即可判断配置参数名称），然后主要看调用的实现类CacheRefreshThread，该实现类的run方法调用了com.netflix.discovery.DiscoveryClient#refreshRegistry方法：

    @VisibleForTesting
    void refreshRegistry() { try { boolean isFetchingRemoteRegionRegistries = isFetchingRemoteRegionRegistries(); boolean remoteRegionsModified = false; // This makes sure that a dynamic change to remote regions to fetch is honored. String latestRemoteRegions = clientConfig.fetchRegistryForRemoteRegions(); if (null != latestRemoteRegions) { String currentRemoteRegions = remoteRegionsToFetch.get(); if (!latestRemoteRegions.equals(currentRemoteRegions)) { // Both remoteRegionsToFetch and AzToRegionMapper.regionsToFetch need to be in sync  synchronized (instanceRegionChecker.getAzToRegionMapper()) { if (remoteRegionsToFetch.compareAndSet(currentRemoteRegions, latestRemoteRegions)) { String[] remoteRegions = latestRemoteRegions.split(","); remoteRegionsRef.set(remoteRegions); instanceRegionChecker.getAzToRegionMapper().setRegionsToFetch(remoteRegions); remoteRegionsModified = true; } else { logger.info("Remote regions to fetch modified concurrently," + " ignoring change from {} to {}", currentRemoteRegions, latestRemoteRegions); } } } else { // Just refresh mapping to reflect any DNS/Property change  instanceRegionChecker.getAzToRegionMapper().refreshMapping(); } } boolean success = fetchRegistry(remoteRegionsModified); if (success) { registrySize = localRegionApps.get().size(); lastSuccessfulRegistryFetchTimestamp = System.currentTimeMillis(); } if (logger.isDebugEnabled()) { StringBuilder allAppsHashCodes = new StringBuilder(); allAppsHashCodes.append("Local region apps hashcode: "); allAppsHashCodes.append(localRegionApps.get().getAppsHashCode()); allAppsHashCodes.append(", is fetching remote regions? "); allAppsHashCodes.append(isFetchingRemoteRegionRegistries); for (Map.Entry<String, Applications> entry : remoteRegionVsApps.entrySet()) { allAppsHashCodes.append(", Remote region: "); allAppsHashCodes.append(entry.getKey()); allAppsHashCodes.append(" , apps hashcode: "); allAppsHashCodes.append(entry.getValue().getAppsHashCode()); } logger.debug("Completed cache refresh task for discovery. All Apps hash code is {} ", allAppsHashCodes); } } catch (Throwable e) { logger.error("Cannot fetch registry from server", e); } }

　　前面的一大堆都是校验，后面的一大堆都是输出，唯一有用的是：boolean success = fetchRegistry(remoteRegionsModified);里面会根据是否是第一发起服务获取请求做不同的请求处理。

　　3、注册中心

　　Eureka server对于各类rest请求的定义都位于com.netflix.eureka.resources包下，以服务注册方法为例：com.netflix.eureka.resources.ApplicationResource#addInstance

    @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(); } // 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 }

　　上面一通校验，对业务逻辑有影响的只有registry.register(info, "true".equals(isReplication));最终调用到org.springframework.cloud.netflix.eureka.server.InstanceRegistry#register(com.netflix.appinfo.InstanceInfo, boolean)方法

    public void register(final InstanceInfo info, final boolean isReplication) {
        handleRegistration(info, resolveInstanceLeaseDuration(info), isReplication);
        super.register(info, isReplication);
    }

　　可以看到，先调用了handleRegistration方法将注册事件通知出去，然后调用了父类的register方法将服务注册。