【源码系列】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 timerscheduler.schedule(new TimedSupervisorTask("heartbeat",scheduler,heartbeatExecutor,renewalIntervalInSecs,TimeUnit.SECONDS,expBackOffBound,new HeartbeatThread()),renewalIntervalInSecs, TimeUnit.SECONDS);// InstanceInfo replicatorinstanceInfoReplicator = 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 timerint 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 timerscheduler.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 == falseif(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.cnblogs.com/2YSP/p/11072255.html