Spring Cloud Eureka源代码解析(2) EurekaServer 重要缓存解析

2017-12-16 10:45:58来源:oschina作者:zhxhash人点击

分享

我们从EurekaServer的缓存说起,因为缓存是EurekaServer的一切存储形式,并且我们通过对缓存的分析可以搞清楚一些对于EurekaServer的误解。


服务实例向EurekaServer注册,注册信息是放在缓存中。从EurekaServer中获取服务实例列表的时候,也是从缓存获取;但是这个缓存结构比较复杂,并且还有很多定时刷新和定时失效的机制,我们需要仔细分析


首先,从核心的服务注册信息存储的地方说起,并简单介绍下其中的注册,取消机制。


其核心逻辑位于AbstractInstanceRegistry这个类,这个类有如下几个重要的存储:


private final ConcurrentHashMap>> registry
= new ConcurrentHashMap>>();
private ConcurrentLinkedQueue recentlyChangedQueue = new ConcurrentLinkedQueue();

并由如下几个重要的锁控制:


private final ReentrantReadWriteLock readWriteLock = new ReentrantReadWriteLock();
private final Lock read = readWriteLock.readLock();
private final Lock write = readWriteLock.writeLock();

其他的field和一些监控还有自我保护机制相关,我们先不管。


有服务实例注册时,会调用register方法,简化后的代码为:


public void register(InstanceInfo registrant, int leaseDuration, boolean isReplication) {
try {
//register虽然看上去好像是修改,但是这里用的是读锁,后面会解释
read.lock();
//从registry中查看这个app是否存在
Map> gMap = registry.get(registrant.getAppName());
//不存在就创建
if (gMap == null) {
final ConcurrentHashMap> gNewMap = new ConcurrentHashMap>();
gMap = registry.putIfAbsent(registrant.getAppName(), gNewMap);
if (gMap == null) {
gMap = gNewMap;
}
}
//查看这个app的这个实例是否已存在
Lease existingLease = gMap.get(registrant.getId());
if (existingLease != null && (existingLease.getHolder() != null)) {
//如果已存在,对比时间戳,保留比较新的实例信息......
} else {
// 如果不存在,证明是一个新的实例
//更新自我保护监控变量的值的代码.....
}
Lease lease = new Lease(registrant, leaseDuration);
if (existingLease != null) {
lease.setServiceUpTimestamp(existingLease.getServiceUpTimestamp());
}
//放入registry
gMap.put(registrant.getId(), lease);
//加入最近修改的记录队列
recentlyChangedQueue.add(new RecentlyChangedItem(lease));
//初始化状态,记录时间等相关代码......
//主动让Response缓存失效
invalidateCache(registrant.getAppName(), registrant.getVIPAddress(), registrant.getSecureVipAddress());
} finally {
read.unlock();
}
}

总结起来,就是主要三件事:1.将实例注册信息放入或者更新registry2.将实例注册信息加入最近修改的记录队列3.主动让Response缓存失效


这个Response缓存,我们稍后就会介绍


当服务实例取消注册时,会调用cancel方法,cacel直接调用internalCancel,这个internalCancel被抽离出来是因为Eureka主动检查evict机制也会调用这个方法。简化后的internalCancel代码为:


protected boolean internalCancel(String appName, String id, boolean isReplication) {
try {
//cancel虽然看上去好像是修改,但是这里用的是读锁,后面会解释
read.lock();
//从registry中剔除这个实例
Map> gMap = registry.get(appName);
Lease leaseToCancel = null;
if (gMap != null) {
leaseToCancel = gMap.remove(id);
}
if (leaseToCancel == null) {
logger.warn("DS: Registry: cancel failed because Lease is not registered for: {}/{}", appName, id);
return false;
} else {
//改变状态,记录状态修改时间等相关代码......
if (instanceInfo != null) {
instanceInfo.setActionType(ActionType.DELETED);
//加入最近修改的记录队列
recentlyChangedQueue.add(new RecentlyChangedItem(leaseToCancel));
}
//主动让Response缓存失效
invalidateCache(appName, vip, svip);
logger.info("Cancelled instance {}/{} (replication={})", appName, id, isReplication);
return true;
}
} finally {
read.unlock();
}
}

总结起来,也是主要三件事:1.从registry中剔除这个实例2.将实例注册信息加入最近修改的记录队列3.主动让Response缓存失效


同时,这个类还会启动两个定时任务,一个是主动失效evict过期应用实例的服务,这里我们先不讨论;另一个就是定时清理最近修改的记录队列的任务:


Iterator it = recentlyChangedQueue.iterator();
while (it.hasNext()) {
if (it.next().getLastUpdateTime() <
System.currentTimeMillis() - serverConfig.getRetentionTimeInMSInDeltaQueue()) {
it.remove();
} else {
break;
}
}

这个RetentionTimeInMSInDeltaQueue默认是180s,可以看出这个队列是一个长度为180s的滑动窗口,保存最近180s以内的应用实例信息修改,后面我们会看到,客户端调用获取增量信息,实际上就是从这个queue中读取,所以可能一段时间内读取到的信息都是一样的。


可以看出registry使我们所有应用所有实例注册信息保存的地方;但是在客户端从EurekaServer获取实例信息的时候,并不是直接读取registry,而是从Response缓存中获取。


Response缓存的实现类是ResponseCacheImpl,主要包括如下缓存field:


private final ConcurrentMap readOnlyCacheMap = new ConcurrentHashMap();
private final LoadingCache readWriteCacheMap;

一个是guava的loadingcache,一个是普通的ConcurrentHashMap


这个loadingcache的初始化:


this.readWriteCacheMap = CacheBuilder.newBuilder().initialCapacity(1000)
.expireAfterWrite(serverConfig.getResponseCacheAutoExpirationInSeconds(), TimeUnit.SECONDS)
.removalListener(new RemovalListener() {
@Override
public void onRemoval(RemovalNotification notification) {
Key removedKey = notification.getKey();
if (removedKey.hasRegions()) {
Key cloneWithNoRegions = removedKey.cloneWithoutRegions();
regionSpecificKeys.remove(cloneWithNoRegions, removedKey);
}
}
})
.build(new CacheLoader() {
@Override
public Value load(Key key) throws Exception {
if (key.hasRegions()) {
Key cloneWithNoRegions = key.cloneWithoutRegions();
regionSpecificKeys.put(cloneWithNoRegions, key);
}
Value value = generatePayload(key);
return value;
}
});

对于每个不存在的Key,会首先初始化,主要是调用generatePayload这个方法:


private Value generatePayload(Key key) {
Stopwatch tracer = null;
try {
String payload;
switch (key.getEntityType()) {
case Application:
boolean isRemoteRegionRequested = key.hasRegions();
if (ALL_APPS.equals(key.getName())) {
//获取所有应用信息
if (isRemoteRegionRequested) {
tracer = serializeAllAppsWithRemoteRegionTimer.start();
payload = getPayLoad(key, registry.getApplicationsFromMultipleRegions(key.getRegions()));
} else {
tracer = serializeAllAppsTimer.start();
payload = getPayLoad(key, registry.getApplications());
}
} else if (ALL_APPS_DELTA.equals(key.getName())) {
//获取所有应用增量信息
if (isRemoteRegionRequested) {
tracer = serializeDeltaAppsWithRemoteRegionTimer.start();
versionDeltaWithRegions.incrementAndGet();
versionDeltaWithRegionsLegacy.incrementAndGet();
payload = getPayLoad(key,
registry.getApplicationDeltasFromMultipleRegions(key.getRegions()));
} else {
tracer = serializeDeltaAppsTimer.start();
versionDelta.incrementAndGet();
versionDeltaLegacy.incrementAndGet();
payload = getPayLoad(key, registry.getApplicationDeltas());
}
} else {
//获取单个应用信息
tracer = serializeOneApptimer.start();
payload = getPayLoad(key, registry.getApplication(key.getName()));
}
break;
//其他类型我们不关心,先忽略掉相关代码
}
return new Value(payload);
} finally {
if (tracer != null) {
tracer.stop();
}
}
}

获取所有应用信息,是从registry中直接拿registry.getApplications(),核心方法是getApplicationsFromMultipleRegions,看下简化过的源码:


public Applications getApplicationsFromMultipleRegions(String[] remoteRegions) {
boolean includeRemoteRegion = null != remoteRegions && remoteRegions.length != 0;
Applications apps = new Applications();
apps.setVersion(1L);
//将registry中的信息封装好放入Applications
for (Entry>> entry : registry.entrySet()) {
Application app = null;
if (entry.getValue() != null) {
for (Entry> stringLeaseEntry : entry.getValue().entrySet()) {
Lease lease = stringLeaseEntry.getValue();
if (app == null) {
app = new Application(lease.getHolder().getAppName());
}
app.addInstance(decorateInstanceInfo(lease));
}
}
if (app != null) {
apps.addApplication(app);
}
}
//读取其他Region的Apps信息,我们目前不关心,略过这部分代码......
//设置AppsHashCode,在之后的介绍中,我们会提到,客户端读取到之后会对比这个AppsHashCode
apps.setAppsHashCode(apps.getReconcileHashCode());
return apps;
}

获取所有应用增量信息,registry.getApplicationDeltas():


public Applications getApplicationDeltas() {
Applications apps = new Applications();
apps.setVersion(responseCache.getVersionDelta().get());
Map applicationInstancesMap = new HashMap();
try {
//这里读取用的是写锁,下面我们就会解释为何这么用
write.lock();
//遍历recentlyChangedQueue,获取所有增量信息
Iterator iter = this.recentlyChangedQueue.iterator();
logger.debug("The number of elements in the delta queue is :"
+ this.recentlyChangedQueue.size());
while (iter.hasNext()) {
Lease lease = iter.next().getLeaseInfo();
InstanceInfo instanceInfo = lease.getHolder();
Object[] args = {instanceInfo.getId(),
instanceInfo.getStatus().name(),
instanceInfo.getActionType().name()};
logger.debug(
"The instance id %s is found with status %s and actiontype %s",
args);
Application app = applicationInstancesMap.get(instanceInfo
.getAppName());
if (app == null) {
app = new Application(instanceInfo.getAppName());
applicationInstancesMap.put(instanceInfo.getAppName(), app);
apps.addApplication(app);
}
app.addInstance(decorateInstanceInfo(lease));
}
//读取其他Region的Apps信息,我们目前不关心,略过这部分代码......
Applications allApps = getApplications(!disableTransparentFallback);
//设置AppsHashCode,在之后的介绍中,我们会提到,客户端读取到之后更新好自己的Apps缓存之后会对比这个AppsHashCode,如果不一样,就会进行一次全量Apps信息请求
apps.setAppsHashCode(allApps.getReconcileHashCode());
return apps;
} finally {
write.unlock();
}
}

为何这里读写锁这么用,首先我们来分析下这个锁保护的对象是谁,可以很明显的看出,是recentlyChangedQueue这个队列。那么谁在修改这个队列,谁又在读取呢?每个服务实例注册,取消的时候,都会修改这个队列,这个队列是多线程修改的。但是读取,只有loadingcache的ALL_APPS_DELTAkey初始化线程会读取,而且在缓存失效前都不会再有线程读取。所以可以归纳为,多线程频繁修改,但是单线程不频繁读取。如果没有锁,那么recentlyChangedQueue在遍历读取时如果遇到修改,就会抛出并发修改异常。如果用writeLock锁住多线程修改,那么同一时间只有一个线程能修改,效率不好。所以。利用读锁锁住多线程修改,利用写锁锁住单线程读取正好符合这里的场景。


前面提到,EurekaClient的查询请求,都是从ResponseCache中获取(从ResponseCache本身缓存的就是请求)。ResponseCache还包括readOnlyCacheMap,这个默认时启用的,就是用户请求会先从readOnlyCacheMap读取,如果readOnlyCacheMap中不存在,则从上面介绍的readWriteCacheMap中获取,之后再放入readOnlyCacheMap。


Value getValue(final Key key, boolean useReadOnlyCache) {
Value payload = null;
try {
if (useReadOnlyCache) {
final Value currentPayload = readOnlyCacheMap.get(key);
if (currentPayload != null) {
payload = currentPayload;
} else {
payload = readWriteCacheMap.get(key);
readOnlyCacheMap.put(key, payload);
}
} else {
payload = readWriteCacheMap.get(key);
}
} catch (Throwable t) {
logger.error("Cannot get value for key :" + key, t);
}
return payload;
}

还有个定时任务:每隔只读缓存刷新时间将ReadWriteMap的信息复制到ReadOnlyMap上面:这个readOnlyCacheMap里面数据是定时从readWriteCacheMap中拷贝出来的:


private TimerTask getCacheUpdateTask() {
return new TimerTask() {
@Override
public void run() {
logger.debug("Updating the client cache from response cache");
for (Key key : readOnlyCacheMap.keySet()) {
if (logger.isDebugEnabled()) {
Object[] args = {key.getEntityType(), key.getName(), key.getVersion(), key.getType()};
logger.debug("Updating the client cache from response cache for key : {} {} {} {}", args);
}
try {
CurrentRequestVersion.set(key.getVersion());
Value cacheValue = readWriteCacheMap.get(key);
Value currentCacheValue = readOnlyCacheMap.get(key);
if (cacheValue != currentCacheValue) {
readOnlyCacheMap.put(key, cacheValue);
}
} catch (Throwable th) {
logger.error("Error while updating the client cache from response cache", th);
}
}
}
};
}

在本篇最开始的时候提到register和cancel都会主动失效对应的ResponseCache,这个主动失效的源代码是:


public void invalidate(String appName, @Nullable String vipAddress, @Nullable String secureVipAddress) {
for (Key.KeyType type : Key.KeyType.values()) { for (Version v : Version.values()) { //对于任意一个APP缓存失效,都要让对应的APP请求响应,全量APP信息请求响应,增量APP信息请求响应失效 invalidate( new Key(Key.EntityType.Application, appName, type, v, EurekaAccept.full), new Key(Key.EntityType.Application, appName, type, v, EurekaAccept.compact), new Key(Key.EntityType.Application, ALL_APPS, type, v, EurekaAccept.full), new Key(Key.EntityType.Application, ALL_APPS, type, v, EurekaAccept.compact), new Key(Key.EntityType.Application, ALL_APPS_DELTA, type, v, EurekaAccept.full), new Key(Key.EntityType.Application, ALL_APPS_DELTA, type, v, EurekaAccept.compact) ); if (null != vipAddress) { invalidate(new Key(Key.EntityType.VIP, vipAddress, type, v, EurekaAccept.full)); }
if (null != secureVipAddress) {
invalidate(new Key(Key.EntityType.SVIP, secureVipAddress, type, v, EurekaAccept.full));
}
}
}
}
public void invalidate(Key... keys) {
for (Key key : keys) {
logger.debug("Invalidating the response cache key : {} {} {} {}, {}",
key.getEntityType(), key.getName(), key.getVersion(), key.getType(), key.getEurekaAccept());
readWriteCacheMap.invalidate(key);
Collection keysWithRegions = regionSpecificKeys.get(key);
if (null != keysWithRegions && !keysWithRegions.isEmpty()) {
for (Key keysWithRegion : keysWithRegions) {
logger.debug("Invalidating the response cache key : {} {} {} {} {}",
key.getEntityType(), key.getName(), key.getVersion(), key.getType(), key.getEurekaAccept());
readWriteCacheMap.invalidate(keysWithRegion);
}
}
}
}

在readWriteCacheMap中使对应的APP请求响应,全量APP信息请求响应,增量APP信息请求响应失效后,下次请求,就会再读取registry生成。对于registry,新加入的应用或者实例会被读取到。对于cancel,退出的应用或者实例也会被去除掉


所以,总结起来,用下面这张图展示下EurekaServer 重要缓存和对应的请求:


image

微信扫一扫

第七城市微信公众平台