一、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方法将服务注册。