Spring Cloud Eureka源码分析之三级缓存的设计原理及源码分析

Eureka Server 为了提供响应效率,提供了三层的缓存结构,将 Eureka Client 所需要的注册信息,直接存储在缓存结构中,实现原理如下图所示。
Spring Cloud Eureka源码分析之三级缓存的设计原理及源码分析-LMLPHP

1、三级缓存分别是什么?



1.1、第一层缓存

registry 注册表 没什么特殊的。

1.2、第二层缓存

readWriteCacheMap,本质上是 Guava 缓存。

写入缓存:readWriteCacheMap 的数据主要同步于存储层。当获取缓存时判断缓存中是否有数据,如果不存在此数据,则通过 CacheLoader 的 load 方法去加载,加载成功之后将数据放入缓存,同时返回数据。

清楚缓存:readWriteCacheMap 缓存过期时间,默认为 180 秒,当服务下线、过期、注册、状态变更,都会来清除此缓存中的数据。

Eureka Client 获取全量或者增量的数据时,会先从三级缓存中获取;如果三级缓存中不存在,再从二级缓存中获取;如果二级缓存也不存在,这时候先将存储层的数据同步到二级缓存中,再从缓存中获取。

通过 Eureka Server 的多层缓存机制,可以非常有效地提升 Eureka Server 的响应时间,通过数据存储层和缓存层的数据切割,根据使用场景来提供不同的数据支持

1.3、第三层缓存

readOnlyCacheMap,本质上是 ConcurrentHashMap,依赖定时从 readWriteCacheMap 同步数据,默认时间为 30 秒。

readOnlyCacheMap : 是一个 CurrentHashMap 只读缓存,这个主要是为了供客户端获取注册信息时使用,其缓存更新,依赖于定时器的更新,通过和 readWriteCacheMap 的值做对比,如果数据不一致,则以 readWriteCacheMap 的数据为准。

2、多级缓存的意义

这里为什么要设计多级缓存呢?原因很简单,就是当存在大规模的服务注册和更新时,如果只是修改一个ConcurrentHashMap数据,那么势必因为锁的存在导致竞争,影响性能。

而Eureka又是AP模型,只需要满足最终可用就行。所以它在这里用到多级缓存来实现读写分离。注册方法写的时候直接写内存注册表(一级缓存),写完之后主动失效读写缓存(二级缓存)。

2.1、注册一个服务实例

向注册表中写入服务实例信息,并使得二级缓存失效

2.2、寻找一个服务

从三级缓存中找,如果有则返回,如果没有则去二级缓存拿并更新三级缓存,如果二级缓存已经失效,触发guava的回调函数从注册表中同步, 同步之后将数据放入二级缓存,同时返回数据。

获取注册信息接口先从 readOnlyCacheMap 只读缓存取,只读缓存没有再去 readWriteCacheMap 读写缓存取,读写缓存没有再去内存注册表里取(不只是取,此处较复杂)。并且,读写缓存会更新回写只读缓存

  • responseCacheUpdateIntervalMs : readOnlyCacheMap 缓存更新的定时器时间间隔,默认为30秒
  • responseCacheAutoExpirationInSeconds : readWriteCacheMap 缓存过期时间,默认为 180 秒 。

2.3、为什么只满足AP?

  • 寻找服务的逻辑是这样的:读取一个配置useReadOnlyCache,默认为true,先从readOnlyCashMap缓存中读取相应的值,如果这个值为空,则去readWriteCashMap中取;但如果这个配置为false,直接去readWriteCashMap中取。如果二级缓存中也没有,则触发guava的回调函数从注册表中同步, 同步之后将数据放入二级缓存,同时返回数据。 再加上readWriteCashMap 与 readOnlyCashMap里面的数据是30秒同步一次,所以Eureka 保证了可用性,但不满足强一致性。
  • (同时这也是一个可以优化的点:我们可以在yml配置文件直接配置useReadOnlyCache 为false,直接从readWriteCashMap中读取注册信息)

2.4 、怎么实现读写分离的?

来看一个问题:为什么默认要一开始从readOnlyCashMap缓存中读取?

  • 为了保证在高并发的情况下 高可用
  • 如果有大量的服务进来,不影响从readOnlyCashMap缓存中读取配置信息,而进来的服务是往registry里写的,减少了读写冲突。

3、缓存初始化

readWriteCacheMap使用的是LoadingCache对象,它是guava中提供的用来实现内存缓存的一个api。创建方式如下

LoadingCache<Long, String> cache = CacheBuilder.newBuilder()
    //缓存池大小,在缓存项接近该大小时, Guava开始回收旧的缓存项
    .maximumSize(10000)
    //设置时间对象没有被读/写访问则对象从内存中删除(在另外的线程里面不定期维护)
    .expireAfterAccess(10, TimeUnit.MINUTES)
    //移除监听器,缓存项被移除时会触发
    .removalListener(new RemovalListener <Long, String>() {
        @Override
        public void onRemoval(RemovalNotification<Long, String> rn) {
            //执行逻辑操作
        }
    })
    .recordStats()//开启Guava Cache的统计功能
    .build(new CacheLoader<String, Object>() {
        @Override
        public Object load(String key) {
            //从 SQL或者NoSql 获取对象
        }
    });//CacheLoader类 实现自动加载

其中,CacheLoader是用来实现缓存自动加载的功能,当触发readWriteCacheMap.get(key)方法时,就会回调CacheLoader.load方法,根据key去服务注册信息中去查找实例数据进行缓存

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(serverConfig.getInitialCapacityOfResponseCache())
        .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);
                }
            }
        })
        .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;
            }
        });

而缓存的加载,是基于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;
            case VIP:
            case SVIP:
                tracer = serializeViptimer.start();
                payload = getPayLoad(key, getApplicationsForVip(key, registry));
                break;
            default:
                logger.error("Unidentified entity type: {} found in the cache key.", key.getEntityType());
                payload = "";
                break;
        }
        return new Value(payload);
    } finally {
        if (tracer != null) {
            tracer.stop();
        }
    }
}

此方法接受一个 Key 类型的参数,返回一个 Value 类型。 其中 Key 中重要的字段有:

  • KeyType ,表示payload文本格式,有 JSON和 XML两种值。
  • EntityType ,表示缓存的类型,有 Application , VIP , SVIP 三种值。
  • entityName ,表示缓存的名称,可能是单个应用名,也可能是 ALL_APPSALL_APPS_DELTA

Value 则有一个 String 类型的payload和一个 byte 数组,表示gzip压缩后的字节。

4、缓存同步

ResponseCacheImpl这个类的构造实现中,初始化了一个定时任务,这个定时任务默认每30s从readWriteCacheMap更新有差异的数据同步到readOnlyCacheMap中

ResponseCacheImpl(EurekaServerConfig serverConfig, ServerCodecs serverCodecs, AbstractInstanceRegistry registry) {
    //省略...
    if (shouldUseReadOnlyResponseCache) {
        timer.schedule(getCacheUpdateTask(),
                       new Date(((System.currentTimeMillis() / responseCacheUpdateIntervalMs) * responseCacheUpdateIntervalMs)
                                + responseCacheUpdateIntervalMs),
                       responseCacheUpdateIntervalMs);
    }
}
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()) {
                    logger.debug("Updating the client cache from response cache for key : {} {} {} {}",
                                 key.getEntityType(), key.getName(), key.getVersion(), key.getType());
                }
                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 for key {}", key.toStringCompact(), th);
                } finally {
                    CurrentRequestVersion.remove();
                }
            }
        }
    };
}

5、缓存失效

在AbstractInstanceRegistry.register这个方法中,当完成服务信息保存后,会调用invalidateCache失效缓存

public void register(InstanceInfo registrant, int leaseDuration, boolean isReplication) {
    //....
     invalidateCache(registrant.getAppName(), registrant.getVIPAddress(), registrant.getSecureVipAddress());
    //....
}
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<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);
            }
        }
    }
}
06-20 15:14