聊聊eureka server的response cache

本文主要研究一下eureka server的response cache

ApplicationResource

eureka-core-1.8.8-sources.jar!/com/netflix/eureka/resources/ApplicationResource.java

    /**
     * Gets information about a particular {@link com.netflix.discovery.shared.Application}.
     *
     * @param version
     *            the version of the request.
     * @param acceptHeader
     *            the accept header of the request to indicate whether to serve
     *            JSON or XML data.
     * @return the response containing information about a particular
     *         application.
     */
    @GET
    public Response getApplication(@PathParam("version") String version,
                                   @HeaderParam("Accept") final String acceptHeader,
                                   @HeaderParam(EurekaAccept.HTTP_X_EUREKA_ACCEPT) String eurekaAccept) {
        if (!registry.shouldAllowAccess(false)) {
            return Response.status(Status.FORBIDDEN).build();
        }

        EurekaMonitors.GET_APPLICATION.increment();

        CurrentRequestVersion.set(Version.toEnum(version));
        KeyType keyType = Key.KeyType.JSON;
        if (acceptHeader == null || !acceptHeader.contains("json")) {
            keyType = Key.KeyType.XML;
        }

        Key cacheKey = new Key(
                Key.EntityType.Application,
                appName,
                keyType,
                CurrentRequestVersion.get(),
                EurekaAccept.fromString(eurekaAccept)
        );

        String payLoad = responseCache.get(cacheKey);

        if (payLoad != null) {
            logger.debug("Found: {}", appName);
            return Response.ok(payLoad).build();
        } else {
            logger.debug("Not Found: {}", appName);
            return Response.status(Status.NOT_FOUND).build();
        }
    }

以ApplicationResource为例,可以看到这里是直接调用了responseCache.get(cacheKey)方法

ResponseCacheImpl.get

    /**
     * Get the cached information about applications.
     *
     * <p>
     * If the cached information is not available it is generated on the first
     * request. After the first request, the information is then updated
     * periodically by a background thread.
     * </p>
     *
     * @param key the key for which the cached information needs to be obtained.
     * @return payload which contains information about the applications.
     */
    public String get(final Key key) {
        return get(key, shouldUseReadOnlyResponseCache);
    }

    @VisibleForTesting
    String get(final Key key, boolean useReadOnlyCache) {
        Value payload = getValue(key, useReadOnlyCache);
        if (payload == null || payload.getPayload().equals(EMPTY_PAYLOAD)) {
            return null;
        } else {
            return payload.getPayload();
        }
    }

    @VisibleForTesting
    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;
    }

这里可以看到,有两个缓存,一个是readOnlyCacheMap,一个是readWriteCacheMap

CacheMap

eureka-core-1.8.8-sources.jar!/com/netflix/eureka/registry/ResponseCacheImpl.java

/**
 * The class that is responsible for caching registry information that will be
 * queried by the clients.
 *
 * <p>
 * The cache is maintained in compressed and non-compressed form for three
 * categories of requests - all applications, delta changes and for individual
 * applications. The compressed form is probably the most efficient in terms of
 * network traffic especially when querying all applications.
 *
 * The cache also maintains separate pay load for <em>JSON</em> and <em>XML</em>
 * formats and for multiple versions too.
 * </p>
 *
 * @author Karthik Ranganathan, Greg Kim
 */
public class ResponseCacheImpl implements ResponseCache {
    //......

    private final ConcurrentMap<Key, Value> readOnlyCacheMap = new ConcurrentHashMap<Key, Value>();

    private final LoadingCache<Key, Value> readWriteCacheMap;

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

        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);
        }
    }
}
  • 这里可以看到readOnlyCacheMap就是一个ConcurrentMap,而readWriteCacheMap是Guava Cache的LoadingCache类型,initialCapacity为1000,而expireAfterWrite为serverConfig.getResponseCacheAutoExpirationInSeconds(),然后CacheLoader的load方法是调用generatePayload方法
  • shouldUseReadOnlyResponseCache读取的是eureka.server.use-read-only-response-cache参数
  • responseCacheUpdateIntervalMs读取的是eureka.server.response-cache-update-interval-ms参数
  • responseCacheAutoExpirationInSeconds读取的是eureka.server.response-cache-update-interval-ms参数

generatePayload

eureka-core-1.8.8-sources.jar!/com/netflix/eureka/registry/ResponseCacheImpl.java

    /*
     * Generate pay load for the given key.
     */
    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();
            }
        }
    }

可以看到这个方法分类型,调用了registry进行查询,然后使用getPayLoad方法进行转码。

CacheUpdateTask

eureka-core-1.8.8-sources.jar!/com/netflix/eureka/registry/ResponseCacheImpl.java

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

这个定时任务主要是从readWriteCacheMap更新数据到readOnlyCacheMap

小结

eureka server对rest api提供了多级缓存,第一层是readOnlyCacheMap,然后是readWriteCacheMap,最后如果readWriteCacheMap读取不到,然后就从registry进行读取。其中readOnlyCacheMap会定时从readWriteCacheMap更新数据,而readWriteCacheMap有自己的过期时间,过期后自动从loader加载新数据。

doc

  • 聊聊eureka的delta配置

原文发布于微信公众号 - 码匠的流水账(geek_luandun)

原文发表时间:2018-05-27

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

发表于

我来说两句

0 条评论
登录 后参与评论

相关文章

来自专栏数据结构与算法

BZOJ1269: [AHOI2006]文本编辑器editor

Descriptio 这些日子,可可不和卡卡一起玩了,原来可可正废寝忘食的想做一个简单而高效的文本编辑器。你能帮助他吗? 为了明确任务目标,可可对“文本编辑器...

2757
来自专栏大内老A

集成EntLib实现ASP.NET MVC的异常处理[续篇]

在《集成EntLib实现ASP.NET MVC的异常处理》我们实现采用EntLib的Exception Handling Application Block(E...

1969
来自专栏Spark学习技巧

textFile构建RDD的分区及compute计算策略

1,textFile A),第一点,就是输入格式,key,value类型及并行度的意义。 def textFile( path: String, mi...

2357
来自专栏偏前端工程师的驿站

Java魔法堂:JUnit4使用详解

目录                                                                              ...

1805
来自专栏码匠的流水账

聊聊eureka server的RemoteRegionRegistry

本文主要研究下eureka server的RemoteRegionRegistry

791
来自专栏个人分享

MongoDB项目中常用方法

  前台通过DBCollection 根据名称获取相应的value 然后加入list中~

1353
来自专栏逆向与安全

菜鸟 学注册机编写之 “查表”

2.载入OD通过下MessageBoxA函数, F9运行程序, 随便输入用户名与注册码, 点ok后断下,如下图所示:

860
来自专栏码匠的流水账

聊聊sentinel的SimpleHttpCommandCenter

sentinel-transport-simple-http-0.1.1-sources.jar!/com/alibaba/csp/sentinel/trans...

801
来自专栏desperate633

设计模式之组合模式(Composite 模式)引入composite模式composite模式的具体实例composite模式小结

在计算机文件系统中,有文件夹的概念,文件夹里面既可以放入文件也可以放入文件夹,但是文件中却不能放入任何东西。文件夹和文件构成了一种递归结构和容器结构。 虽然文...

762
来自专栏wannshan(javaer,RPC)

dubbo通信消息解析过程分析(1)

由于rpc底层涉及网络编程接口,线程模型,网络数据结构,服务协议,细到字节的处理。牵涉内容较多,今天就先从一个点说起。 说说,dubbo通过netty框架做传...

4736

扫码关注云+社区