【源码系列】Eureka源码分析

对于服务注册中心、服务提供者、服务消费者这个三个主要元素来说，服务提供者和服务消费者（即Eureka客户端）在整个运行机制中是大部分通信行为的主动发起者（服务注册、续约、下线等），而注册中心主要是处理请求的接收者。所以，我们从Eureka的客户端为入口分析它是如何完成这些主动通信的。
一般情况下，我们将一个SpringBoot应用注册到 Eureka Server 或者从 Eureka Server 获取服务器列表时，就做了两件事：

在应用启动类添加注解 @EnableDiscoveryClient
在 application.properties 文件上用 eureka.client.service-url.defaultZone 参数指定注册中心的地址

我们先看看 @EnableDiscoveryClient 这个注解的源码，如下：

/**
 * Annotation to enable a DiscoveryClient implementation.
 * @author Spencer Gibb
 */
@Target(ElementType.TYPE)
@Retention(RetentionPolicy.RUNTIME)
@Documented
@Inherited
@Import(EnableDiscoveryClientImportSelector.class)
public @interface EnableDiscoveryClient {

    /**
     * If true, the ServiceRegistry will automatically register the local server.
     */
    boolean autoRegister() default true;
}

通过注释可以知道，该注解可以开启 DiscoveryClient 实例，然后我们搜索 DiscoveryClient 会发现一个类和一个接口，它们的关系如图。

enter description here

右边的org.springframework.cloud.client.discovery.DiscoveryClient 是SpringCloud的接口，体现了面向接口编程的思想，定义了用来发现服务的常用抽象方法。org.springframework.cloud.netflix.eureka.EurekaDiscoveryClient是该接口的实现，是对Eureka发现服务的封装，内部依赖了一个EurekaClient接口，所以真正实现发现服务的是com.netflix.discovery.DiscoveryClient类。
查看类注释的内容：

/**
 * The class that is instrumental for interactions with <tt>Eureka Server</tt>.
 *
 * <p>
 * <tt>Eureka Client</tt> is responsible for a) <em>Registering</em> the
 * instance with <tt>Eureka Server</tt> b) <em>Renewal</em>of the lease with
 * <tt>Eureka Server</tt> c) <em>Cancellation</em> of the lease from
 * <tt>Eureka Server</tt> during shutdown
 * <p>
 * d) <em>Querying</em> the list of services/instances registered with
 * <tt>Eureka Server</tt>
 * <p>
 *
 * <p>
 * <tt>Eureka Client</tt> needs a configured list of <tt>Eureka Server</tt>
 * {@link java.net.URL}s to talk to.These {@link java.net.URL}s are typically amazon elastic eips
 * which do not change. All of the functions defined above fail-over to other
 * {@link java.net.URL}s specified in the list in the case of failure.
 * </p>
 *
 * @author Karthik Ranganathan, Greg Kim
 * @author Spencer Gibb
 *
 */
@Singleton
public class DiscoveryClient implements EurekaClient {
    ...
}

这个类用于帮助与 Eureka Server 相互协作
Eureka Client客户端负责以下内容：

向Eureka Server 注册服务实例
向 Eureka Server 服务续约
服务关闭时取消租约
查询注册在 Eureka Server 上的服务或实例列表
Eureka Client 还需要配置一个 Eureka Server 的服务列表。

哪里对Eureka Server的URL列表配置？

根据我们配置的属性名eureka.client.serviceUrl.defaultZone,通过serviceUrl可以找到该属性相关的加载属性，就是DiscoveryClient里有个getEurekaServiceUrlsFromConfig（）方法但是弃用了，改用EndpointUtils这个工具类，代码如下：

/** 

     * Get the list of all eureka service urls from properties file for the eureka client to talk to. 

     * 

     * @param clientConfig the clientConfig to use 

     * @param instanceZone The zone in which the client resides 

     * @param preferSameZone true if we have to prefer the same zone as the client, false otherwise 

     * @return an (ordered) map of zone -> list of urls mappings, with the preferred zone first in iteration order 

     */ 

    public static Map<String, List<String>> getServiceUrlsMapFromConfig(EurekaClientConfig clientConfig, String instanceZone, boolean preferSameZone) { 

        Map<String, List<String>> orderedUrls = new LinkedHashMap<>(); 

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

        String zone = availZones[myZoneOffset]; 

        List<String> serviceUrls = clientConfig.getEurekaServerServiceUrls(zone); 

        if (serviceUrls != null) { 

            orderedUrls.put(zone, serviceUrls); 

        } 

        int currentOffset = myZoneOffset == (availZones.length – 1) ? 0 : (myZoneOffset + 1); 

        while (currentOffset != myZoneOffset) { 

            zone = availZones[currentOffset]; 

            serviceUrls = clientConfig.getEurekaServerServiceUrls(zone); 

            if (serviceUrls != null) { 

                orderedUrls.put(zone, serviceUrls); 

            } 

            if (currentOffset == (availZones.length – 1)) { 

                currentOffset = 0; 

            } else { 

                currentOffset++; 

            } 

        } 

        if (orderedUrls.size() < 1) { 

            throw new IllegalArgumentException(“DiscoveryClient: invalid serviceUrl specified!”); 

        } 

        return orderedUrls; 

    }

Region,Zone

getRegion()方法可以看出一个微服务应用只可以属于一个Region，如果没配置则为default，可以通过eureka.client.region属性来定义。

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

getAvailabilityZones()方法可以看出Region与Zone的关系，一个Region可以有多个Zone，设置时可以用逗号来分隔。默认采用defaultZone。

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

        return value.split(",");
    }

在获取Region和Zone的信息后，根据传入的参数按一定的算法确定加载位于哪一个Zone的serviceUrls。

enter description here

getEurekaServerServiceUrls方法是EurekaClientConfigBean的实现类，该方法用来获取一个Zone下配置的所以serviceUrl，通过标注出来的地方可以知道，eureka.client.serviceUrl.defaultZone属性可以配置多个，用逗号来分隔。

enter description here

注意： Ribbon具有区域亲和特性，Ribbon的默认策略会优先访问同客户端处于同一个Zone中的实例。所以通过Zone属性的定义，配置实际部署的物理结构，我们就可以有效地设计出对区域性故障的容错集群。

服务注册

前面说了多个服务注册中心信息的加载，这里再看看 DiscoveryClient 类是如何实现服务注册的。通过查看该类的构造函数，发现它调用了以下方法。

 /**
     * Initializes all scheduled tasks.
     */
    private void initScheduledTasks() {
        ...

        if (clientConfig.shouldRegisterWithEureka()) {
           ...

            // Heartbeat timer
            scheduler.schedule(
                    new TimedSupervisorTask(
                            "heartbeat",
                            scheduler,
                            heartbeatExecutor,
                            renewalIntervalInSecs,
                            TimeUnit.SECONDS,
                            expBackOffBound,
                            new HeartbeatThread()
                    ),
                    renewalIntervalInSecs, TimeUnit.SECONDS);

            // InstanceInfo replicator
            instanceInfoReplicator = new InstanceInfoReplicator(
                    this,
                    instanceInfo,
                    clientConfig.getInstanceInfoReplicationIntervalSeconds(),
                    2); // burstSize

            ...
            instanceInfoReplicator.start(clientConfig.getInitialInstanceInfoReplicationIntervalSeconds());
        } else {
            logger.info("Not registering with Eureka server per configuration");
        }
    }

这里先根据配置判断是不是要注册到 Eureka，然后创建心跳检测任务，获取 instanceInfoReplicator。InstanceInfoReplicator类实现 Runnable接口，instanceInfoReplicator实例会执行一个定时任务，这个定时任务的内容可以查看该类的run()方法。

enter description here

这里定时刷新实例信息，discoveryClient.register()这里触发了服务注册，register()的内容如下：

服务注册的方法

通过注释也能看出来，这里是通过发送REST请求的方式进行的，com.netflix.appinfo.InstanceInfo就是注册时客户端给服务端的元数据。

服务获取与服务续约

上面说到的 initScheduledTasks() 方法还有两个定时任务，分别是服务获取和服务续约。

private void initScheduledTasks() {
        if (clientConfig.shouldFetchRegistry()) {
            // registry cache refresh timer
            int registryFetchIntervalSeconds = clientConfig.getRegistryFetchIntervalSeconds();
            int expBackOffBound = clientConfig.getCacheRefreshExecutorExponentialBackOffBound();
            scheduler.schedule(
                    new TimedSupervisorTask(
                            "cacheRefresh",
                            scheduler,
                            cacheRefreshExecutor,
                            registryFetchIntervalSeconds,
                            TimeUnit.SECONDS,
                            expBackOffBound,
                            new CacheRefreshThread()
                    ),
                    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
            scheduler.schedule(
                    new TimedSupervisorTask(
                            "heartbeat",
                            scheduler,
                            heartbeatExecutor,
                            renewalIntervalInSecs,
                            TimeUnit.SECONDS,
                            expBackOffBound,
                            new HeartbeatThread()
                    ),
                    renewalIntervalInSecs, TimeUnit.SECONDS);

           	...
            instanceInfoReplicator.start(clientConfig.getInitialInstanceInfoReplicationIntervalSeconds());
        } else {
            logger.info("Not registering with Eureka server per configuration");
        }
    }

clientConfig.shouldFetchRegistry()这里其实是通过eureka.client.fetch-registry参数来判断的，默认为true，它可以定期更新客户端的服务清单，从而客户端能访问到健康的服务实例。
服务续约也是发送REST请求实现的。

 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() == 404) {
                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() == 200;
        } catch (Throwable e) {
            logger.error(PREFIX + "{} - was unable to send heartbeat!", appPathIdentifier, e);
            return false;
        }
    }

服务获取的过程省略。

服务下线

服务端根据实例Id和appName执行remove操作。

 void unregister() {
        // It can be null if shouldRegisterWithEureka == false
        if(eurekaTransport != null && eurekaTransport.registrationClient != null) {
            try {
                logger.info("Unregistering ...");
                EurekaHttpResponse<Void> httpResponse = eurekaTransport.registrationClient.cancel(instanceInfo.getAppName(), instanceInfo.getId());
                logger.info(PREFIX + "{} - deregister  status: {}", appPathIdentifier, httpResponse.getStatusCode());
            } catch (Exception e) {
                logger.error(PREFIX + "{} - de-registration failed{}", appPathIdentifier, e.getMessage(), e);
            }
        }
    }

注册中心处理

前面的分析都是从客户端出发的，现在看看 Eureka Server是如何处理各种Rest请求的。这种请求的定义都在com.netflix.eureka.resources包下。
以服务注册为例：
调用 ApplicationResource 类下的 addInstance()方法。

@POST
    @Consumes({"application/json", "application/xml"})
    public Response addInstance(InstanceInfo info, @HeaderParam("x-netflix-discovery-replication") String isReplication) {
        logger.debug("Registering instance {} (replication={})", info.getId(), isReplication);
        if(this.isBlank(info.getId())) {
            return Response.status(400).entity("Missing instanceId").build();
        } else if(this.isBlank(info.getHostName())) {
            return Response.status(400).entity("Missing hostname").build();
        } else if(this.isBlank(info.getIPAddr())) {
            return Response.status(400).entity("Missing ip address").build();
        } else if(this.isBlank(info.getAppName())) {
            return Response.status(400).entity("Missing appName").build();
        } else if(!this.appName.equals(info.getAppName())) {
            return Response.status(400).entity("Mismatched appName, expecting " + this.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();
        } else {
            DataCenterInfo dataCenterInfo = info.getDataCenterInfo();
            if(dataCenterInfo instanceof UniqueIdentifier) {
                String dataCenterInfoId = ((UniqueIdentifier)dataCenterInfo).getId();
                if(this.isBlank(dataCenterInfoId)) {
                    boolean experimental = "true".equalsIgnoreCase(this.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();
                    }

                    if(dataCenterInfo instanceof AmazonInfo) {
                        AmazonInfo amazonInfo = (AmazonInfo)dataCenterInfo;
                        String effectiveId = amazonInfo.get(MetaDataKey.instanceId);
                        if(effectiveId == null) {
                            amazonInfo.getMetadata().put(MetaDataKey.instanceId.getName(), info.getId());
                        }
                    } else {
                        logger.warn("Registering DataCenterInfo of type {} without an appropriate id", dataCenterInfo.getClass());
                    }
                }
            }

            this.registry.register(info, "true".equals(isReplication));
            return Response.status(204).build();
        }
    }

在对注册信息进行校验后，会调用org.springframework.cloud.netflix.eureka.server.InstanceRegistry的register(InstanceInfo info, int leaseDuration, boolean isReplication)方法。

enter description here

首先会把新服务注册事件传播出去，然后调用父类com.netflix.eureka.registry.AbstractInstanceRegistry中的实现。

 public void register(InstanceInfo registrant, int leaseDuration, boolean isReplication) {
        try {
            this.read.lock();
            Map<String, Lease<InstanceInfo>> gMap = (Map)this.registry.get(registrant.getAppName());
            ...
        } finally {
            this.read.unlock();
        }

    }

保存实例信息的双层Map

InstanceInfo的元数据信息保存在一个ConcurrentHashMap中，它是一个双层的Map结构，第一层的key是服务名（即InstanceInfo的appName属性），第二层的key是实例名（即InstanceInfo的InstanceId属性）。
ApplicationResource中的其他方法可以自行研究。

文章由极客之音整理，本文链接：https://www.bmabk.com/index.php/post/13222.html