上一篇说到Eureka客户端,这篇分析服务器端是如何与客户端交互及实现了哪些功能
这里解释一些Eureka的专有名词
instanceInfo: 服务实例信息, instanceId(唯一),app, appGroupName, ipAddr, port, leaseInfo等
Lease:租约,客户端向服务器端注册相当于签下了一纸租约,存在租期的概念,到期了要是没动静就默认不租了,把实例剔除。非常形象生动
LeaseInfo:使用了内联构造器的设计模式,来标识实例的租约信息,例如服务器端:注册时间、最后一次续约时间、被剔除时间、标记为UP的时间,客户端:续约间隔时间、租约的有效时长
LeaseManager:望文生义,这个接口负责管理实例,那么要知道如何管理服务注册、服务下线、服务续约和服务剔除,查看这个接口的实现类就能明白。
所以,查看LeaseManager的类图,看到类的继承关系,大致就明白,AbstractInstanceRegistry就是LeaseManager的基本实现类,它还同时实现了LookupService
LookupService接口的实现类还有DiscoveryClient, 主要的作用是发现活跃的服务实例,那么可否认为Eureka Server也是一个Eureka Client呢,因为和DiscoveryClient功能实现很像。事实上是可以的,Eureka Server同时也是一个Eureka Client, 这就是在配置时要注意的,如果不想注册Eureka Server为Client时,registerWithEureka属性设置为true的原因。
registerWithEureka: false #先不向Eureka Server注册自己的信息
那么基本框架大致讲到这里,接下来就详细讲解AbstractInstanceRegistry这个Eureka Server的核心类,其中registry就是注册表,结构如下
ConcurrentHashMap<String/*app name*/, Map<String/*InstanceId*/, Lease<InstanceInfo>>> registry = new ConcurrentHashMap<String, Map<String, Lease<InstanceInfo>>>();
服务注册 AbstractInstanceRegistry#register
public void register(InstanceInfo registrant, int leaseDuration, boolean isReplication) {
try {
// 获取读锁
read.lock();
// 根据appName对服务实例集群进行分类
Map<String, Lease<InstanceInfo>> gMap = registry.get(registrant.getAppName());
REGISTER.increment(isReplication);
if (gMap == null) {
final ConcurrentHashMap<String, Lease<InstanceInfo>> gNewMap = new ConcurrentHashMap<String, Lease<InstanceInfo>>();
// 如果存在键值,直接返回已存在的值,否则添加该键值对
gMap = registry.putIfAbsent(registrant.getAppName(), gNewMap);
if (gMap == null) {
gMap = gNewMap;
}
}
// 根据instanceId获取实例的租约
Lease<InstanceInfo> existingLease = gMap.get(registrant.getId());
// Retain the last dirty timestamp without overwriting it, if there is already a lease
if (existingLease != null && (existingLease.getHolder() != null)) {
Long existingLastDirtyTimestamp = existingLease.getHolder().getLastDirtyTimestamp();
Long registrationLastDirtyTimestamp = registrant.getLastDirtyTimestamp();
logger.debug("Existing lease found (existing={}, provided={}", existingLastDirtyTimestamp, registrationLastDirtyTimestamp);
// this is a > instead of a >= because if the timestamps are equal, we still take the remote transmitted
// InstanceInfo instead of the server local copy.
// 如果该实例的租约已经存在,比较最后更新的时间戳,取最大值的注册信息为有效信息
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 = existingLease.getHolder();
}
} else {
// The lease does not exist and hence it is a new registration
// 如果租约不存在,创建新的注册实例
synchronized (lock) {
if (this.expectedNumberOfClientsSendingRenews > 0) {
// Since the client wants to register it, increase the number of clients sending renews
// 因为有新的注册实例,需要相应增加服务器端期待的Client发来续约的数量
this.expectedNumberOfClientsSendingRenews = this.expectedNumberOfClientsSendingRenews + 1;
// 同时更新每分钟续约的阈值
// this.numberOfRenewsPerMinThreshold = (int) (this.expectedNumberOfClientsSendingRenews * (60.0 / serverConfig.getExpectedClientRenewalIntervalSeconds()) * serverConfig.getRenewalPercentThreshold());
updateRenewsPerMinThreshold();
}
}
logger.debug("No previous lease information found; it is new registration");
}
// 创建新的租约
Lease<InstanceInfo> lease = new Lease<InstanceInfo>(registrant, leaseDuration);
if (existingLease != null) {
// 如果租约存在,继承租约的服务上线初始时间
lease.setServiceUpTimestamp(existingLease.getServiceUpTimestamp());
}
// 保存租约
gMap.put(registrant.getId(), lease);
synchronized (recentRegisteredQueue) {
// 添加最近注册的实例信息,用来做debug或者统计
// CircularQueues here for debugging/statistics purposes only
recentRegisteredQueue.add(new Pair<Long, String>(
System.currentTimeMillis(),
registrant.getAppName() + "(" + registrant.getId() + ")"));
}
// This is where the initial state transfer of overridden status happens
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 (!overriddenInstanceStatusMap.containsKey(registrant.getId())) {
logger.info("Not found overridden id {} and hence adding it", registrant.getId());
overriddenInstanceStatusMap.put(registrant.getId(), registrant.getOverriddenStatus());
}
}
InstanceStatus overriddenStatusFromMap = overriddenInstanceStatusMap.get(registrant.getId());
if (overriddenStatusFromMap != null) {
logger.info("Storing overridden status {} from map", overriddenStatusFromMap);
registrant.setOverriddenStatus(overriddenStatusFromMap);
}
// Set the status based on the overridden status rules
// 根据覆盖状态规则得到服务实例的最终状态,并设置服务实例的当前状态
InstanceStatus overriddenInstanceStatus = getOverriddenInstanceStatus(registrant, existingLease, isReplication);
registrant.setStatusWithoutDirty(overriddenInstanceStatus);
// If the lease is registered with UP status, set lease service up timestamp
// 如果服务实例状态为UP,设置租约的服务上线时间,只有第一次设置有效
if (InstanceStatus.UP.equals(registrant.getStatus())) {
lease.serviceUp();
}
// 表示ActionType为ADDED
registrant.setActionType(ActionType.ADDED);
// 添加最近租约变更记录队列,用于Eureka Client增量式获取注册表信息
recentlyChangedQueue.add(new RecentlyChangedItem(lease));
// 设置服务实例信息更新时间
registrant.setLastUpdatedTimestamp();
// 设置response缓存过期,将用于Eureka Client全量获取注册表信息,也就是说失效responseCache中对应的缓存
invalidateCache(registrant.getAppName(), registrant.getVIPAddress(), registrant.getSecureVipAddress());
logger.info("Registered instance {}/{} with status {} (replication={})",
registrant.getAppName(), registrant.getId(), registrant.getStatus(), isReplication);
} finally {
read.unlock();
}
}
接受服务心跳AbstractInstanceRegistry#renew
public boolean renew(String appName, String id, boolean isReplication) {
RENEW.increment(isReplication);
// 根据appName获得服务集群的租约组合
Map<String, Lease<InstanceInfo>> gMap = registry.get(appName);
Lease<InstanceInfo> leaseToRenew = null;
if (gMap != null) {
leaseToRenew = gMap.get(id);
}
// 租约不存在,直接返回false
if (leaseToRenew == null) {
RENEW_NOT_FOUND.increment(isReplication);
logger.warn("DS: Registry: lease doesn't exist, registering resource: {} - {}", appName, id);
return false;
} else {
InstanceInfo instanceInfo = leaseToRenew.getHolder();
if (instanceInfo != null) {
// touchASGCache(instanceInfo.getASGName());
// 根据覆盖状态规则得到服务实例的最终状态
InstanceStatus overriddenInstanceStatus = this.getOverriddenInstanceStatus(
instanceInfo, leaseToRenew, isReplication);
// 如果得到的服务实例最后状态是UNKNOWN,取消续约
if (overriddenInstanceStatus == InstanceStatus.UNKNOWN) {
logger.info("Instance status UNKNOWN possibly due to deleted override for instance {}"
+ "; re-register required", instanceInfo.getId());
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", instanceInfo.getStatus().name(),
instanceInfo.getOverriddenStatus().name(),
instanceInfo.getId());
instanceInfo.setStatusWithoutDirty(overriddenInstanceStatus);
}
}
// 统计每分钟续租的次数,用于自我保护
renewsLastMin.increment();
// 更新租约中的有效时间
leaseToRenew.renew();
return true;
}
}
服务剔除AbstractInstanceRegistry#evict(),服务剔除是一个定时的任务,EvictionTask继承了TimerTask,负责定期剔除过期服务,有个evictionTimer负责每过60s执行一次任务,可以通过参数evictionIntervalTimerInMs设置剔除服务运行时间间隔。
public void evict() {
evict(0l);
}
public void evict(long additionalLeaseMs) {
logger.debug("Running the evict task");
// 如果enableSelfPreservation的值为false,则不允许剔除服务。默认启动剔除服务功能
if (!isLeaseExpirationEnabled()) {
logger.debug("DS: lease expiration is currently disabled.");
return;
}
// We collect first all expired items, to evict them in random order. For large eviction sets,
// if we do not that, we might wipe out whole apps before self preservation kicks in. By randomizing it,
// the impact should be evenly distributed across all applications.
// 此处开发者本人解释了为何要随机地剔除服务,这是因为大量的服务在同一个应用上剔除会击垮该应用,导致程序崩溃。而随机剔除就能将影响均匀分布到各个应用上,保护程序不被击溃。
List<Lease<InstanceInfo>> expiredLeases = new ArrayList<>();
// 遍历注册表registry,一次性获得所有的过期租约
for (Entry<String, Map<String, Lease<InstanceInfo>>> groupEntry : registry.entrySet()) {
Map<String, Lease<InstanceInfo>> leaseMap = groupEntry.getValue();
if (leaseMap != null) {
for (Entry<String, Lease<InstanceInfo>> leaseEntry : leaseMap.entrySet()) {
Lease<InstanceInfo> lease = leaseEntry.getValue();
if (lease.isExpired(additionalLeaseMs) && lease.getHolder() != null) {
expiredLeases.add(lease);
}
}
}
}
// To compensate for GC pauses or drifting local time, we need to use current registry size as a base for
// triggering self-preservation. Without that we would wipe out full registry.
// 计算最大允许剔除的租约数,获取注册表租约的总数
int registrySize = (int) getLocalRegistrySize();
// 计算注册表租约的阀值
int registrySizeThreshold = (int) (registrySize * serverConfig.getRenewalPercentThreshold());
// 计算剔除租约的数量
int evictionLimit = registrySize - registrySizeThreshold;
int toEvict = Math.min(expiredLeases.size(), evictionLimit);
if (toEvict > 0) {
logger.info("Evicting {} items (expired={}, evictionLimit={})", toEvict, expiredLeases.size(), evictionLimit);
Random random = new Random(System.currentTimeMillis());
// 逐个随机剔除服务,让剔除均匀分布在所有应用中,防止程序崩溃
for (int i = 0; i < toEvict; i++) {
// Pick a random item (Knuth shuffle algorithm)
int next = i + random.nextInt(expiredLeases.size() - i);
Collections.swap(expiredLeases, i, next);
Lease<InstanceInfo> lease = expiredLeases.get(i);
String appName = lease.getHolder().getAppName();
String id = lease.getHolder().getId();
EXPIRED.increment();
logger.warn("DS: Registry: expired lease for {}/{}", appName, id);
// 最后进行服务下线
internalCancel(appName, id, false);
}
}
}
服务下线AbstractInstanceRegistry#internalCancel()
protected boolean internalCancel(String appName, String id, boolean isReplication) {
try {
// 获取读锁,防止被其它线程进行修改
read.lock();
CANCEL.increment(isReplication);
// 根据appName获取服务实例的集群
Map<String, Lease<InstanceInfo>> gMap = registry.get(appName);
Lease<InstanceInfo> leaseToCancel = null;
// 如果集群不为空,移除服务实例的租约
if (gMap != null) {
leaseToCancel = gMap.remove(id);
}
synchronized (recentCanceledQueue) {
// 将服务实例信息添加到最近下线服务实例统计队列
recentCanceledQueue.add(new Pair<Long, String>(System.currentTimeMillis(), appName + "(" + id + ")"));
}
InstanceStatus instanceStatus = overriddenInstanceStatusMap.remove(id);
if (instanceStatus != null) {
logger.debug("Removed instance id {} from the overridden map which has value {}", id, instanceStatus.name());
}
// 如果租约不存在,返回false
if (leaseToCancel == null) {
CANCEL_NOT_FOUND.increment(isReplication);
logger.warn("DS: Registry: cancel failed because Lease is not registered for: {}/{}", appName, id);
return false;
} else {
// Cancels the lease by updating the eviction time,
leaseToCancel.cancel();
InstanceInfo instanceInfo = leaseToCancel.getHolder();
String vip = null;
String svip = null;
if (instanceInfo != null) {
// 将instanceInfo中的ActionType设置为DELETED
instanceInfo.setActionType(ActionType.DELETED);
// 用于Eureka Client增量式获取注册表信息
recentlyChangedQueue.add(new RecentlyChangedItem(leaseToCancel));
instanceInfo.setLastUpdatedTimestamp();
vip = instanceInfo.getVIPAddress();
svip = instanceInfo.getSecureVipAddress();
}
// 使appName, vip, svip在缓存中失效
invalidateCache(appName, vip, svip);
logger.info("Cancelled instance {}/{} (replication={})", appName, id, isReplication);
// 下线成功
return true;
}
} finally {
// 释放锁
read.unlock();
}
}