Eureka是怎么work的
那eureka client如何将本地服务的注册信息发送到远端的注册服务器eureka server上。通过下面的源码分析，看出Eureka Client的定时任务调用Eureka Server的Reset接口，而Eureka接收到调用请求后会处理服务的注册以及Eureka Server中的数据同步的问题。
服务注册
源码分析，看出服务注册可以认为是Eureka client自己完成，不需要服务本身来关心。
Eureka Client的定时任务调用Eureka Server的提供接口
在com.netflix.discovery.DiscoveryClient启动的时候，会初始化一个定时任务，定时的把本地的服务配置信息，即需要注册到远端的服务信息自动刷新到注册服务器上。
首先看一下Eureka的代码，在spring-cloud-netflix-eureka-server工程中可以找到这个依赖eureka-client-1.4.11.jar查看代码可以看到，
com.netflix.discovery.DiscoveryClient.java中的1240行可以看到Initializes all scheduled tasks，在1277行，可以看到InstanceInfoReplicator定时任务
在DiscoveryClient中初始化一个InstanceInfoReplicator，其实里面封装了以定时任务。
/**
* Initializes all scheduled tasks.
*/
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);
// 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);
}
//点击可以查看start方法
instanceInfoReplicator.start(clientConfig.getInitialInstanceInfoReplicationIntervalSeconds());
} else {
logger.info("Not registering with Eureka server per configuration");
}
}
com.netflix.discovery.DiscoveryClient中的 register()方法，大概在811行。
/**
* Register with the eureka service by making the appropriate REST call.
*/
boolean register() throws Throwable {
logger.info(PREFIX + appPathIdentifier + ": registering service...");
EurekaHttpResponse<Void> httpResponse;
try {
//Eureka Client客户端，调用Eureka服务端的入口
httpResponse = eurekaTransport.registrationClient.register(instanceInfo);
} catch (Exception e) {
logger.warn("{} - registration failed {}", PREFIX + appPathIdentifier, e.getMessage(), e);
throw e;
}
if (logger.isInfoEnabled()) {
logger.info("{} - registration status: {}", PREFIX + appPathIdentifier, httpResponse.getStatusCode());
}
return httpResponse.getStatusCode() == 204;
}
Eureka server服务端请求入口
ApplicationResource.java文件中第183行，如下所示，可以看出Eureka是通过http post的方式去服务注册
@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.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 amazonInfo1 = "DataCenterInfo of type " + dataCenterInfo.getClass() + " must contain a valid id";
return Response.status(400).entity(amazonInfo1).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());
}
}
}
//进入到PeerAwareInstanceRegistryImpl中的register方法
this.registry.register(info, "true".equals(isReplication)); return Response.status(204).build();
}
}
InstanceRegistry.java文件中的52行，可以看到调用PeerAwareInstanceRegistryImpl中的278行register方法
public void register(InstanceInfo info, boolean isReplication) {
this.handleRegistration(info, this.resolveInstanceLeaseDuration(info), isReplication);
super.register(info, isReplication);
}
PeerAwareInstanceRegistryImpl中的278行register方法
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);
// 同步Eureka中的服务信息
this.replicateToPeers(PeerAwareInstanceRegistryImpl.Action.Register, info.getAppName(), info.getId(), info, (InstanceStatus)null, isReplication);
}
AbstractInstanceRegistry.java中151行，可以看到Eureka真正的服务注册实现的代码
public void register(InstanceInfo registrant, int leaseDuration, boolean isReplication) {
try {
this.read.lock();
Object gMap = (Map)this.registry.get(registrant.getAppName());
EurekaMonitors.REGISTER.increment(isReplication);
if(gMap == null) {
ConcurrentHashMap existingLease = new ConcurrentHashMap();
gMap = (Map)this.registry.putIfAbsent(registrant.getAppName(), existingLease);
if(gMap == null) {
gMap = existingLease;
}
}
Lease existingLease1 = (Lease)((Map)gMap).get(registrant.getId());
if(existingLease1 != null && existingLease1.getHolder() != null) {
Long lease1 = ((InstanceInfo)existingLease1.getHolder()).getLastDirtyTimestamp();
Long overriddenStatusFromMap = registrant.getLastDirtyTimestamp();
logger.debug("Existing lease found (existing={}, provided={}", lease1, overriddenStatusFromMap);
if(lease1.longValue() > overriddenStatusFromMap.longValue()) {
logger.warn("There is an existing lease and the existing lease\'s dirty timestamp {} is greater than the one that is being registered {}", lease1, overriddenStatusFromMap);
logger.warn("Using the existing instanceInfo instead of the new instanceInfo as the registrant");
registrant = (InstanceInfo)existingLease1.getHolder();
}
} else {
Object lease = this.lock;
synchronized(this.lock) {
if(this.expectedNumberOfRenewsPerMin > 0) {
this.expectedNumberOfRenewsPerMin += 2;
this.numberOfRenewsPerMinThreshold = (int)((double)this.expectedNumberOfRenewsPerMin * this.serverConfig.getRenewalPercentThreshold());
}
}
logger.debug("No previous lease information found; it is new registration");
}
Lease lease2 = new Lease(registrant, leaseDuration);
if(existingLease1 != null) {
lease2.setServiceUpTimestamp(existingLease1.getServiceUpTimestamp());
}
((Map)gMap).put(registrant.getId(), lease2);
AbstractInstanceRegistry.CircularQueue overriddenStatusFromMap1 = this.recentRegisteredQueue;
synchronized(this.recentRegisteredQueue) {
this.recentRegisteredQueue.add(new Pair(Long.valueOf(System.currentTimeMillis()), registrant.getAppName() + "(" + registrant.getId() + ")"));
}
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 overriddenStatusFromMap2 = (InstanceStatus)this.overriddenInstanceStatusMap.get(registrant.getId());
if(overriddenStatusFromMap2 != null) {
logger.info("Storing overridden status {} from map", overriddenStatusFromMap2);
registrant.setOverriddenStatus(overriddenStatusFromMap2);
}
InstanceStatus overriddenInstanceStatus = this.getOverriddenInstanceStatus(registrant, existingLease1, isReplication);
registrant.setStatusWithoutDirty(overriddenInstanceStatus);
if(InstanceStatus.UP.equals(registrant.getStatus())) {
lease2.serviceUp();
}
registrant.setActionType(ActionType.ADDED);
this.recentlyChangedQueue.add(new AbstractInstanceRegistry.RecentlyChangedItem(lease2));
registrant.setLastUpdatedTimestamp();
this.invalidateCache(registrant.getAppName(), registrant.getVIPAddress(), registrant.getSecureVipAddress());
logger.info("Registered instance {}/{} with status {} (replication={})", new Object[]{registrant.getAppName(), registrant.getId(), registrant.getStatus(), Boolean.valueOf(isReplication)});
} finally {
this.read.unlock();
}
}
说明:注册信息其实就是存储在一个 ConcurrentHashMap<string, map<string,="" lease<instanceinfo="">>> registry的结构中。
private final ConcurrentHashMap<String, Map<String, Lease<InstanceInfo>>> registry = new ConcurrentHashMap();
总结
ApplicationResource类接收Http服务请求，调用PeerAwareInstanceRegistryImpl的register方法，PeerAwareInstanceRegistryImpl完成服务注册后，
调用replicateToPeers向其它Eureka Server节点（Peer）做状态同步。如下图所示。
相关类
EurekaHttpClient：提供eureka 的客户端API方法 ，实现类JerseyReplicationClient ，eureka客户端发送http 请求类
ApplicationResource ：eureka 服务端接收 eureka 客户端发送请求的处理类
InstanceRegistry eureka 服务端注册表实例类
LookupService：：查询活动服务的实例接口
DefaultEurekaClientConfig：eureka 客户端默认配置类
EurekaClientConfigBean :eureka 客户端配置bean ，在配置文件中提示属性的类，配置前缀：eureka.client
