Eureka Server缓存机制
Eureka Server的缓存机制依赖于谷歌的gauva cache , 在Eureka中通过
com.netflix.eureka.registry.ResponseCacheImpl , 这个操作类来实现缓存的机制。
ResponseCacheImpl
ResponseCacheImpl(EurekaServerConfig serverConfig, ServerCodecs serverCodecs, AbstractInstanceRegistry registry) {
this.serverConfig = serverConfig;
this.serverCodecs = serverCodecs;
// 是否使用只读缓存
this.shouldUseReadOnlyResponseCache = serverConfig.shouldUseReadOnlyResponseCache();
this.registry = registry;
// 缓存更新的时间间隔
long responseCacheUpdateIntervalMs = serverConfig.getResponseCacheUpdateIntervalMs();
// 构建读写缓存
this.readWriteCacheMap =
CacheBuilder.newBuilder().initialCapacity(1000)
.expireAfterWrite(serverConfig.getResponseCacheAutoExpirationInSeconds(), TimeUnit.SECONDS)
.removalListener(new RemovalListener<Key, Value>() {
@Override
public void onRemoval(RemovalNotification<Key, Value> notification) {
Key removedKey = notification.getKey();
if (removedKey.hasRegions()) {
Key cloneWithNoRegions = removedKey.cloneWithoutRegions();
regionSpecificKeys.remove(cloneWithNoRegions, removedKey);
}
}
})
// 缓存加载器，当缓存不存在时，会自动执行load方法，进行缓存加载。同时返回缓存数据
.build(new CacheLoader<Key, Value>() {
@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;
}
});
// 是否使用只读缓存，如果使用，此处则启动一个定时器，用来复制readWriteCacheMap 的数据至readOnlyCacheMap
if (shouldUseReadOnlyResponseCache) {
timer.schedule(getCacheUpdateTask(),
new Date(((System.currentTimeMillis() / responseCacheUpdateIntervalMs) * responseCacheUpdateIntervalMs)
+ responseCacheUpdateIntervalMs),
responseCacheUpdateIntervalMs);
}
try {
Monitors.registerObject(this);
} catch (Throwable e) {
logger.warn("Cannot register the JMX monitor for the InstanceRegistry", e);
}
}
通过上面可以很简单的看出， Eureka Server的缓存是通过一个只读，一个读写缓存来实现的。
readWriteCacheMap ： 此处存放的是最终的缓存， 当服务下线，过期，注册，状态变更，都会来清除这个缓存里面的数据。 然后通过CacheLoader进行缓存加载，在进行readWriteCacheMap.get(key)的时候，首先看这个缓存里面有没有该数据，如果没有则通过CacheLoader的load方法去加载，加载成功之后将数据放入缓存，同时返回数据
readOnlyCacheMap ： 这是一个JVM的CurrentHashMap只读缓存，这个主要是为了供客户端获取注册信息时使用，其缓存更新，依赖于定时器的更新，通过和readWriteCacheMap 的值做对比，如果数据不一致，则以readWriteCacheMap 的数据为准。
responseCacheUpdateIntervalMs ： readOnlyCacheMap 缓存更新的定时器时间间隔，默认为30秒
responseCacheAutoExpirationInSeconds : readWriteCacheMap 缓存过期时间，默认为 180 秒 。
CacheUpdateTask
readOnlyCacheMap 定时器的任务执行类。
private TimerTask getCacheUpdateTask() {
return new TimerTask() {
@Override
public void run() {
logger.debug("Updating the client cache from response cache");
// 循环readOnlyCacheMap里面的KEY
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());
// 从readWriteCacheMap获取数据
Value cacheValue = readWriteCacheMap.get(key);
// 当前的只读数据
Value currentCacheValue = readOnlyCacheMap.get(key);
// 判断数据是否一致
if (cacheValue != currentCacheValue) {
// 如果不一致，覆盖只读缓存里面的数据，以readWriteCacheMap为准
readOnlyCacheMap.put(key, cacheValue);
}
} catch (Throwable th) {
logger.error("Error while updating the client cache from response cache", th);
}
}
}
};
}
invalidate缓存过期
public void invalidate(Key... keys) {
// 循环传入的key一次调用API进行清除
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<Key> 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这个缓存， readOnlyCacheMap 只读 缓存并没有更新，也就说当客户端的信息发生变化之后， 只读缓存不是第一时间感知到的。 只读缓存的更新只能依赖那个30秒的定时任务来更新。
GET获取缓存
@VisibleForTesting
String get(final Key key, boolean useReadOnlyCache) {
// 主要看这个getValue
Value payload = getValue(key, useReadOnlyCache);
if (payload == null || payload.getPayload().equals(EMPTY_PAYLOAD)) {
return null;
} else {
return payload.getPayload();
}
}
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;
}
useReadOnlyCache ： shouldUseReadOnlyResponseCache ，可以配置是否使用只读缓存，默认是true
readWriteCacheMap.get(key) : 这个使用的是gauva 的缓存机制，如果当前的缓存里面这个key没有，那么
会直接调用CacheLoader.load()方法，从最上面的代码可以看到， load方法，主要是执行了generatePayload()
方法。
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();
//调用registry.getApplications() 获取应用信息。同时调用getPayLoad进行编码
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();
// 调用registry.getApplicationDeltas() 获取增量注册信息
payload = getPayLoad(key, registry.getApplicationDeltas());
}
} else {
// 根据key直接获取注册信息
tracer = serializeOneApptimer.start();
payload = getPayLoad(key, registry.getApplication(key.getName()));
}
break;
// 根据VIP获取
case VIP:
case SVIP:
tracer = serializeViptimer.start();
payload = getPayLoad(key, getApplicationsForVip(key, registry));
break;
default:
logger.error("Unidentified entity type: " + key.getEntityType() + " found in the cache key.");
payload = "";
break;
}
return new Value(payload);
} finally {
if (tracer != null) {
tracer.stop();
}
}
}
// 编码类
private String getPayLoad(Key key, Applications apps) {
EncoderWrapper encoderWrapper = serverCodecs.getEncoder(key.getType(), key.getEurekaAccept());
String result;
try {
result = encoderWrapper.encode(apps);
} catch (Exception e) {
logger.error("Failed to encode the payload for all apps", e);
return "";
}
if(logger.isDebugEnabled()) {
logger.debug("New application cache entry {} with apps hashcode {}", key.toStringCompact(), apps.getAppsHashCode());
}
return result;
}
entityType : 分为三种，Application, VIP, SVIP ， 客户端获取注册信息的话，传入的主要是Application类型的，另外两种类型此处不做考虑 。
根据KEY_NAME 的不同，判断是全量获取信息，还是增量获取信息，由此执行不同的逻辑，想看全量获取和增量获取是怎么实现的可以看7. Eureka 获取注册信息（七）
Eureka Client缓存机制
Eureka Client缓存机制很简单，设置了一个每30秒执行一次的定时任务，定时去服务端获取注册信息。获取之后，存入本地内存。
具体的获取规则可以看7. Eureka 获取注册信息（七）
网友评论