1. 首页
  2. 文章中心
  3. 面试题

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

77 阅读 0 评论
我是靠谱客的博主 风中季节，最近开发中收集的这篇文章主要介绍Spring Cloud Eureka源码分析之三级缓存的设计原理及源码分析，觉得挺不错的，现在分享给大家，希望可以做个参考。

概述

Eureka源码

Eureka Server 为了提供响应效率，提供了两层的缓存结构，将 Eureka Client 所需要的注册信息，直接存储在缓存结构中，实现原理如下图所示。

image-20211121172504080

第一层缓存：readOnlyCacheMap，本质上是 ConcurrentHashMap，依赖定时从 readWriteCacheMap 同步数据，默认时间为 30 秒。

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

第二层缓存：readWriteCacheMap，本质上是 Guava 缓存。

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

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

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

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

多级缓存的意义

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

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

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

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

缓存初始化

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_APPS 或 ALL_APPS_DELTA 。

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

缓存同步

在ResponseCacheImpl这个类的构造实现中，初始化了一个定时任务，这个定时任务每个

ResponseCacheImpl(EurekaServerConfig serverConfig, ServerCodecs serverCodecs, AbstractInstanceRegistry registry) {
    //省略...
    if (shouldUseReadOnlyResponseCache) {
        timer.schedule(getCacheUpdateTask(),
                       new Date(((System.currentTimeMillis() / responseCacheUpdateIntervalMs) * responseCacheUpdateIntervalMs)
                                + responseCacheUpdateIntervalMs),
                       responseCacheUpdateIntervalMs);
    }
}

默认每30s从readWriteCacheMap更新有差异的数据同步到readOnlyCacheMap中

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

缓存失效

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

public void register(InstanceInfo registrant, int leaseDuration, boolean isReplication) {
    //....
     invalidateCache(registrant.getAppName(), registrant.getVIPAddress(), registrant.getSecureVipAddress());
    //....
}

最终调用ResponseCacheImpl.invalidate方法，完成缓存的失效机制

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);
            }
        }
    }
}

版权声明：本博客所有文章除特别声明外，均采用 CC BY-NC-SA 4.0 许可协议。转载请注明来自 Mic带你学架构！
如果本篇文章对您有帮助，还请帮忙点个关注和赞，您的坚持是我不断创作的动力。欢迎关注「跟着Mic学架构」公众号公众号获取更多技术干货！

最后

以上就是风中季节为你收集整理的Spring Cloud Eureka源码分析之三级缓存的设计原理及源码分析的全部内容，希望文章能够帮你解决Spring Cloud Eureka源码分析之三级缓存的设计原理及源码分析所遇到的程序开发问题。

如果觉得靠谱客网站的内容还不错，欢迎将靠谱客网站推荐给程序员好友。

本图文内容来源于网友提供，作为学习参考使用，或来自网络收集整理，版权属于原作者所有。
点赞(51)

相关文章

Eureka 缓存机制详解Eureka Server 数据存储Eureka Server 缓存机制其它缓存设计Eureka 常用配置Eureka Server常用配置Eureka Client 常用配置Eureka Instance 常用配置总结
Eureka 缓存机制详解Eureka Server 数据存储Eureka Server 缓存机制其它缓存设计Eureka 常用配置Eureka Server常用配置Eureka Client 常用配置Eureka Instance 常用配置总结
Eureka缓存机制
Eureka缓存机制
Eureka的server端理解
Eureka的server端理解
聊聊eureka server的response cache
聊聊eureka server的response cache
Spring Cloud Eureka源码分析之三级缓存的设计原理及源码分析
Spring Cloud Eureka源码分析之三级缓存的设计原理及源码分析
eureka数据解构及配置优化
eureka数据解构及配置优化
Spring Cloud Eureka参数详解
Spring Cloud Eureka参数详解
【源码】eureka的 三级缓存之设计探索
【源码】eureka的 三级缓存之设计探索

评论列表共有 0 条评论

立即
投稿
返回
顶部