Spring Cloud Eureka-Server 源码分析二
1 对外暴露服务接口
上一步完成了Eureka Server 初始化的分析，现在来分析下他是怎么对外暴露服务接口的；上一步简单带过了，现在我们来看下具体的实现细节。
在Eureka Server启动过程中主配置类注册了Jersey框架（是⼀个发布restful⻛格接⼝的框架，类似于我们熟知的SpringMVC。
进入自动装配的配置文件：/META-INF/spring.factories
image-20200708140142440.png
org.springframework.boot.autoconfigure.EnableAutoConfiguration=\
org.springframework.cloud.netflix.eureka.server.EurekaServerAutoConfiguration
进入 核心自动配置类EurekaServerAutoConfiguration，主要关注如下Bean的初始化。
@Configuration
@Import({EurekaServerInitializerConfiguration.class})
@ConditionalOnBean({Marker.class})
@EnableConfigurationProperties({EurekaDashboardProperties.class, InstanceRegistryProperties.class})
@PropertySource({"classpath:/eureka/server.properties"})
public class EurekaServerAutoConfiguration extends WebMvcConfigurerAdapter {
@Bean
public FilterRegistrationBean jerseyFilterRegistration(Application eurekaJerseyApp) {
FilterRegistrationBean bean = new FilterRegistrationBean();
bean.setFilter(new ServletContainer(eurekaJerseyApp));
bean.setOrder(2147483647);
bean.setUrlPatterns(Collections.singletonList("/eureka/*"));
return bean;
}
@Bean
public Application jerseyApplication(Environment environment, ResourceLoader resourceLoader) {
ClassPathScanningCandidateComponentProvider provider = new ClassPathScanningCandidateComponentProvider(false, environment);
//配置Jersey注解；@Path 类似于 SpringMVC中的 @RequestMapping注解
provider.addIncludeFilter(new AnnotationTypeFilter(Path.class));
provider.addIncludeFilter(new AnnotationTypeFilter(Provider.class));
Set<Class<?>> classes = new HashSet();
//注解扫描： 指定如下要扫描的包；操作类似于Spring @Component-Scan 操作
//private static final String[] EUREKA_PACKAGES = new String[]{"com.netflix.discovery", "com.netflix.eureka"};
String[] var5 = EUREKA_PACKAGES;
int var6 = var5.length;
for(int var7 = 0; var7 < var6; ++var7) {
String basePackage = var5[var7];
Set<BeanDefinition> beans = provider.findCandidateComponents(basePackage);
Iterator var10 = beans.iterator();
while(var10.hasNext()) {
BeanDefinition bd = (BeanDefinition)var10.next();
Class<?> cls = ClassUtils.resolveClassName(bd.getBeanClassName(), resourceLoader.getClassLoader());
classes.add(cls);
}
}
Map<String, Object> propsAndFeatures = new HashMap();
propsAndFeatures.put("com.sun.jersey.config.property.WebPageContentRegex", "/eureka/(fonts|images|css|js)/.*");
DefaultResourceConfig rc = new DefaultResourceConfig(classes);
rc.setPropertiesAndFeatures(propsAndFeatures);
return rc;
}
}
Jersey对外提供的接口服务被称为资源Resource，在SpringMVC中称为控制器Controller；
看下Jersey扫描的包：com.netflix.eureka.resources包下所有的Resource就是对外发布的服务接口，有服务列表拉取、服务注册、心跳续约等接口。
image-20200708140054936.png
2 服务注册接口（接收客户端的注册服务）
Restful风格，POST代表提交注册；
@Produces({"application/xml", "application/json"})
public class ApplicationResource {
@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);
//注册信息验证，必传的信息：instanceId hostname ipAddress appName dataCenterInfo
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(); //注册成功，响应204
}
}
}
com.netflix.eureka.registry.PeerAwareInstanceRegistryImpl#register - 注册服务信息并同步到其它Eureka节点；
public void register(InstanceInfo info, boolean isReplication) {
int leaseDuration = 90; //默认失效间隔，如果客户端自己配置了，则以客户端为准
if (info.getLeaseInfo() != null && info.getLeaseInfo().getDurationInSecs() > 0) {
leaseDuration = info.getLeaseInfo().getDurationInSecs();
}
//调用父类的注册实例
super.register(info, leaseDuration, isReplication);
//server把当前该实例信息注册到其他的对等server节点中
this.replicateToPeers(PeerAwareInstanceRegistryImpl.Action.Register, info.getAppName(), info.getId(), info, (InstanceStatus)null, isReplication);
}
2.1 register 注册实例
AbstractInstanceRegistry#register()：注册，实例信息存储到注册表是⼀个ConcurrentHashMap
public abstract class AbstractInstanceRegistry implements InstanceRegistry {
public void register(InstanceInfo registrant, int leaseDuration, boolean isReplication) {
try {
this.read.lock(); //读锁
//private final ConcurrentHashMap<String, Map<String, Lease<InstanceInfo>>> registry = new ConcurrentHashMap();
Map<String, Lease<InstanceInfo>> gMap = (Map)this.registry.get(registrant.getAppName());
EurekaMonitors.REGISTER.increment(isReplication);//注册统计+1
// 如果当前appName实例信息为空，新建Map
if (gMap == null) {
ConcurrentHashMap<String, Lease<InstanceInfo>> gNewMap = new ConcurrentHashMap();
gMap = (Map)this.registry.putIfAbsent(registrant.getAppName(), gNewMap);
if (gMap == null) {
gMap = gNewMap;
}
}
// 获取实例的Lease租约信息
Lease<InstanceInfo> existingLease = (Lease)((Map)gMap).get(registrant.getId());
// 如果已有租约，则比较两个租约信息，选择最近的为准
if (existingLease != null && existingLease.getHolder() != null) {
Long existingLastDirtyTimestamp = ((InstanceInfo)existingLease.getHolder()).getLastDirtyTimestamp();
Long registrationLastDirtyTimestamp = registrant.getLastDirtyTimestamp();
logger.debug("Existing lease found (existing={}, provided={}", existingLastDirtyTimestamp, registrationLastDirtyTimestamp);
//⽐较当前请求实例租约 和 已有租约 的LastDirtyTimestamp，选择靠后的
if (existingLastDirtyTimestamp > registrationLastDirtyTimestamp) {
logger.warn("There is an existing lease and the existing lease's dirty timestamp {} is greater than the one that is being registered {}", existingLastDirtyTimestamp, registrationLastDirtyTimestamp);
logger.warn("Using the existing instanceInfo instead of the new instanceInfo as the registrant");
registrant = (InstanceInfo)existingLease.getHolder();
}
} else {
//如果没有租约信息，说明是新的实例
synchronized(this.lock) {
if (this.expectedNumberOfClientsSendingRenews > 0) {
//期待的每分钟续约数+1（因为30s⼀个）
++this.expectedNumberOfClientsSendingRenews;
//更新numberOfRenewsPerMinThreshold每分钟续约阀值（85%）
this.updateRenewsPerMinThreshold();
}
}
logger.debug("No previous lease information found; it is new registration");
}
Lease<InstanceInfo> lease = new Lease(registrant, leaseDuration);
if (existingLease != null) {
lease.setServiceUpTimestamp(existingLease.getServiceUpTimestamp());
}
// 将当前实例信息放入缓存中
((Map)gMap).put(registrant.getId(), lease);
// 同步最近维护的注册队列
synchronized(this.recentRegisteredQueue) {
this.recentRegisteredQueue.add(new Pair(System.currentTimeMillis(), registrant.getAppName() + "(" + registrant.getId() + ")"));
}
//如果当前实例已经维护了OverriddenStatus，将其也放到此Eureka Server的overriddenInstanceStatusMap中
if (!InstanceStatus.UNKNOWN.equals(registrant.getOverriddenStatus())) {
logger.debug("Found overridden status {} for instance {}. Checking to see if needs to be add to the overrides", registrant.getOverriddenStatus(), registrant.getId());
if (!this.overriddenInstanceStatusMap.containsKey(registrant.getId())) {
logger.info("Not found overridden id {} and hence adding it", registrant.getId());
this.overriddenInstanceStatusMap.put(registrant.getId(), registrant.getOverriddenStatus());
}
}
InstanceStatus overriddenStatusFromMap = (InstanceStatus)this.overriddenInstanceStatusMap.get(registrant.getId());
if (overriddenStatusFromMap != null) {
logger.info("Storing overridden status {} from map", overriddenStatusFromMap);
registrant.setOverriddenStatus(overriddenStatusFromMap);
}
InstanceStatus overriddenInstanceStatus = this.getOverriddenInstanceStatus(registrant, existingLease, isReplication);
registrant.setStatusWithoutDirty(overriddenInstanceStatus);
//如果租约以UP状态注册，设置租赁服务时间戳
if (InstanceStatus.UP.equals(registrant.getStatus())) {
lease.serviceUp();
}
registrant.setActionType(ActionType.ADDED);//这是动作类型为添加
this.recentlyChangedQueue.add(new AbstractInstanceRegistry.RecentlyChangedItem(lease));
registrant.setLastUpdatedTimestamp();
// 使当前应⽤的ResponseCache失效
this.invalidateCache(registrant.getAppName(), registrant.getVIPAddress(), registrant.getSecureVipAddress());
logger.info("Registered instance {}/{} with status {} (replication={})", new Object[]{registrant.getAppName(), registrant.getId(), registrant.getStatus(), isReplication});
} finally {
this.read.unlock();//释放读锁
}
}
}
2.2 replicateToPeers 同步对等节点
PeerAwareInstanceRegistryImpl#replicateToPeers() ：复制到Eureka对等节点
private void replicateToPeers(PeerAwareInstanceRegistryImpl.Action action, String appName, String id, InstanceInfo info, InstanceStatus newStatus, boolean isReplication) {
Stopwatch tracer = action.getTimer().start();
try {
// 如果是复制操作（针对当前节点，false）
if (isReplication) {
this.numberOfReplicationsLastMin.increment();
}
// 如果它已经是复制，请不要再次复制，直接return
if (this.peerEurekaNodes == Collections.EMPTY_LIST || isReplication) {
return;
}
Iterator var8 = this.peerEurekaNodes.getPeerEurekaNodes().iterator();
// 遍历集群所有节点（除当前节点外）
while(var8.hasNext()) {
PeerEurekaNode node = (PeerEurekaNode)var8.next();
if (!this.peerEurekaNodes.isThisMyUrl(node.getServiceUrl())) {
// 复制Instance实例操作到某个node节点
this.replicateInstanceActionsToPeers(action, appName, id, info, newStatus, node);
}
}
} finally {
tracer.stop();
}
}
PeerAwareInstanceRegistryImpl#replicateInstanceActionsToPeers 复制实例操作到某个node节点
private void replicateInstanceActionsToPeers(PeerAwareInstanceRegistryImpl.Action action, String appName, String id, InstanceInfo info, InstanceStatus newStatus, PeerEurekaNode node) {
try {
InstanceInfo infoFromRegistry = null;
CurrentRequestVersion.set(Version.V2);
switch(action) {
case Cancel: // 服务下架动作
node.cancel(appName, id);
break;
case Heartbeat: // 服务心跳预约
InstanceStatus overriddenStatus = (InstanceStatus)this.overriddenInstanceStatusMap.get(id);
infoFromRegistry = this.getInstanceByAppAndId(appName, id, false);
node.heartbeat(appName, id, infoFromRegistry, overriddenStatus, false);
break;
case Register: // 服务注册
node.register(info);
break;
case StatusUpdate: //服务状态更新
infoFromRegistry = this.getInstanceByAppAndId(appName, id, false);
node.statusUpdate(appName, id, newStatus, infoFromRegistry);
break;
case DeleteStatusOverride: // 状态删除覆盖
infoFromRegistry = this.getInstanceByAppAndId(appName, id, false);
node.deleteStatusOverride(appName, id, infoFromRegistry);
}
} catch (Throwable var9) {
logger.error("Cannot replicate information to {} for action {}", new Object[]{node.getServiceUrl(), action.name(), var9});
}
}
3 服务心跳续约接口（客户端调用续约）
InstanceResource的renewLease⽅法中完成客户端的⼼跳（续约）处理，关键代码：registry.renew(app.getName(), id, isFromReplicaNode);
@Produces({"application/xml", "application/json"})
public class InstanceResource {
@PUT
public Response renewLease(@HeaderParam("x-netflix-discovery-replication") String isReplication, @QueryParam("overriddenstatus") String overriddenStatus, @QueryParam("status") String status, @QueryParam("lastDirtyTimestamp") String lastDirtyTimestamp) {
boolean isFromReplicaNode = "true".equals(isReplication);
//调用续约
boolean isSuccess = this.registry.renew(this.app.getName(), this.id, isFromReplicaNode);
if (!isSuccess) {
logger.warn("Not Found (Renew): {} - {}", this.app.getName(), this.id);
return Response.status(Status.NOT_FOUND).build();
} else {
Response response;
if (lastDirtyTimestamp != null && this.serverConfig.shouldSyncWhenTimestampDiffers()) {
response = this.validateDirtyTimestamp(Long.valueOf(lastDirtyTimestamp), isFromReplicaNode);
if (response.getStatus() == Status.NOT_FOUND.getStatusCode() && overriddenStatus != null && !InstanceStatus.UNKNOWN.name().equals(overriddenStatus) && isFromReplicaNode) {
this.registry.storeOverriddenStatusIfRequired(this.app.getAppName(), this.id, InstanceStatus.valueOf(overriddenStatus));
}
} else {
response = Response.ok().build();
}
logger.debug("Found (Renew): {} - {}; reply status={}", new Object[]{this.app.getName(), this.id, response.getStatus()});
return response;
}
}
}
com.netflflix.eureka.registry.PeerAwareInstanceRegistryImpl#renew
public boolean renew(String appName, String id, boolean isReplication) {
//调用父类的renew操作
if (super.renew(appName, id, isReplication)) {
// 操作同步到其他节点: 具体细节见上方 2.2
this.replicateToPeers(PeerAwareInstanceRegistryImpl.Action.Heartbeat, appName, id, (InstanceInfo)null, (InstanceStatus)null, isReplication);
return true;
} else {
return false;
}
}
com.netflix.eureka.registry.AbstractInstanceRegistry#renew
public boolean renew(String appName, String id, boolean isReplication) {
EurekaMonitors.RENEW.increment(isReplication);
Map<String, Lease<InstanceInfo>> gMap = (Map)this.registry.get(appName);
Lease<InstanceInfo> leaseToRenew = null;
if (gMap != null) {
leaseToRenew = (Lease)gMap.get(id);
}
if (leaseToRenew == null) {
EurekaMonitors.RENEW_NOT_FOUND.increment(isReplication);
logger.warn("DS: Registry: lease doesn't exist, registering resource: {} - {}", appName, id);
return false;
} else {
InstanceInfo instanceInfo = (InstanceInfo)leaseToRenew.getHolder();
if (instanceInfo != null) {
InstanceStatus overriddenInstanceStatus = this.getOverriddenInstanceStatus(instanceInfo, leaseToRenew, isReplication);
if (overriddenInstanceStatus == InstanceStatus.UNKNOWN) {
logger.info("Instance status UNKNOWN possibly due to deleted override for instance {}; re-register required", instanceInfo.getId());
EurekaMonitors.RENEW_NOT_FOUND.increment(isReplication);
return false;
}
if (!instanceInfo.getStatus().equals(overriddenInstanceStatus)) {
logger.info("The instance status {} is different from overridden instance status {} for instance {}. Hence setting the status to overridden status", new Object[]{instanceInfo.getStatus().name(), instanceInfo.getOverriddenStatus().name(), instanceInfo.getId()});
instanceInfo.setStatusWithoutDirty(overriddenInstanceStatus);
}
}
this.renewsLastMin.increment();
//对最后更新时间戳进⾏更新
leaseToRenew.renew();
return true;
}
}
com.netflix.eureka.lease.Lease#renew：更新最后更新时间戳
public void renew() {
this.lastUpdateTimestamp = System.currentTimeMillis() + this.duration;
}
