緩存框架 Caffeine 的可視化探索與實踐

作者：Wang Zhi 2024-07-25 14:04:16

開發

Caffeine 作為一個高性能的緩存框架而被大量使用。本文基于Caffeine已有的基礎進行定制化開發實現可視化功能。

一、背景

Caffeine緩存是一個高性能、可擴展、內存優化的 Java 緩存庫，基于 Google 的 Guava Cache演進而來并提供了接近最佳的命中率。

Caffeine 緩存包含以下特點：

高效快速：Caffeine 緩存使用近似算法和并發哈希表等優化技術，使得緩存的訪問速度非常快。
內存友好：Caffeine 緩存使用一種內存優化策略，能夠根據需要動態調整緩存的大小，有效地利用內存資源。
多種緩存策略：Caffeine 緩存支持多種緩存策略，如基于容量、時間、權重、手動移除、定時刷新等，并提供了豐富的配置選項，能夠適應不同的應用場景和需求。
支持異步加載和刷新：Caffeine 緩存支持異步加載和刷新緩存項，可以與 Spring 等框架無縫集成。
清理策略：Caffeine 使用 Window TinyLFU 清理策略，它提供了接近最佳的命中率。
支持自動加載和自動過期：Caffeine 緩存可以根據配置自動加載和過期緩存項，無需手動干預。
統計功能：Caffeine 緩存提供了豐富的統計功能，如緩存命中率、緩存項數量等，方便評估緩存的性能和效果。

正是因為Caffeine具備的上述特性，Caffeine作為項目中本地緩存的不二選擇，越來越多的項目集成了Caffeine的功能，進而衍生了一系列的業務視角的需求。

日常使用的需求之一希望能夠實時評估Caffeine實例的內存占用情況并能夠提供動態調整緩存參數的能力，但是已有的內存分析工具MAT需要基于dump的文件進行分析無法做到實時，這也是整個事情的起因之一。

二、業務的技術視角

能夠對項目中的Caffeine的緩存實例能夠做到近實時統計，實時查看緩存的實例個數。
能夠對Caffeine的每個實例的緩存配置參數、內存占用、緩存命中率做到實時查看，同時能夠支持單個實例的緩存過期時間，緩存條目等參數進行動態配置下發。
能夠對Caffeine的每個實例的緩存數據做到實時查看，并且能夠支持緩存數據的立即失效等功能。

基于上述的需求背景，結合caffeine的已有功能和定制的部分源碼開發，整體作為caffeine可視化的技術項目進行推進和落地。

三、可視化能力

Caffeine可視化項目目前已支持功能包括：

項目維度的全局緩存實例的管控。
單緩存實例配置信息可視化、內存占用可視化、命中率可視化。
單緩存實例的數據查詢、配置動態變更、緩存數據失效等功能。

3.1 緩存實例的全局管控

說明：

以應用維度+機器維度展示該應用下包含的緩存實例對象，每個實例包含緩存設置中的大小、過期策略、過期時間、內存占用、緩存命中率等信息。
單實例維度的內存占用和緩存命中率支持以趨勢圖進行展示。
單實例維度支持配置變更操作和緩存查詢操作。

3.2 內存占用趨勢

說明：

內存占用趨勢記錄該緩存實例對象近一段時間內存占用的趨勢變化。
時間周期目前支持展示近兩天的數據。

3.3 命中率趨勢

說明：

命中率趨勢記錄該緩存實例對象近一段時間緩存命中的變化情況。
時間周期目前支持展示近兩天的數據。

3.4 配置變更

說明：

配置變更目前支持緩存大小和過期時間的動態設置。
目前暫時支持單實例的設置，后續會支持全量生效功能。

3.5 緩存查詢

說明：

單實例維度支持緩存數據的查詢。
目前支持常見的緩存Key類型包括String類型、Long類型、Int類型。

四、原理實現

4.1 整體設計框架

Caffeine框架功能整合

說明：

沿用Caffeine的基礎功能包括Caffeine的緩存功能和Caffeine統計功能。
新增Caffeine內存占用預估功能，該功能主要是預估緩存實例對象占用的內存情況。
新增Caffeine實例命名功能，該功能是針對每個實例對象提供命名功能，是全局管控的基礎。
新增Caffeine實例全局管控功能，該功能主要維護項目運行中所有的緩存實例。

Caffeine可視化框架

說明：

【項目工程側】：Caffeine的可視化框架基于Caffeine框架功能整合的基礎上增加通信層進行數據數據上報和配置的下發。
【管控平臺側】：負責緩存數據上報的接收展示，配置變更命令的下發。
【通信層支持push和pull兩種模式】，push模式主要用于統計數據的實時上報，pull模式主要用于配置下發和緩存數據查詢。

4.2 源碼實現

業務層-緩存對象的管理

static Cache<String, List<String>> accountWhiteCache = Caffeine.newBuilder()
            .expireAfterWrite(VivoConfigManager.getInteger("trade.account.white.list.cache.ttl", 10), TimeUnit.MINUTES)
            .recordStats().maximumSize(VivoConfigManager.getInteger("trade.account.white.list.cache.size", 100)).build();
常規的Caffeine實例的創建方式
 
 
static Cache<String, List<String>> accountWhiteCache = Caffeine.newBuilder().applyName("accountWhiteCache")
            .expireAfterWrite(VivoConfigManager.getInteger("trade.account.white.list.cache.ttl", 10), TimeUnit.MINUTES)
            .recordStats().maximumSize(VivoConfigManager.getInteger("trade.account.white.list.cache.size", 100)).build();
支持實例命名的Caffeine實例的創建方式

說明：

在Caffeine實例創建的基礎上增加了緩存實例的命名功能，通過.applyName("accountWhiteCache")來定義緩存實例的命名。

public final class Caffeine<K, V> {
 
  /**
   * caffeine的實例名稱
   */
  String instanceName;
 
  /**
   * caffeine的實例維護的Map信息
   */
  static Map<String, Cache> cacheInstanceMap = new ConcurrentHashMap<>();
 
  @NonNull
  public <K1 extends K, V1 extends V> Cache<K1, V1> build() {
    requireWeightWithWeigher();
    requireNonLoadingCache();
 
    @SuppressWarnings("unchecked")
    Caffeine<K1, V1> self = (Caffeine<K1, V1>) this;
    Cache localCache =  isBounded() ? new BoundedLocalCache.BoundedLocalManualCache<>(self) : new UnboundedLocalCache.UnboundedLocalManualCache<>(self);
 
    if (null != localCache && StringUtils.isNotEmpty(localCache.getInstanceName())) {
      cacheInstanceMap.put(localCache.getInstanceName(), localCache);
    }
 
    return localCache;
  }
}

說明：

每個Caffeine都有一個實例名稱instanceName。
全局通過cacheInstanceMap來維護Caffeine實例對象的名稱和實例的映射關系。
通過維護映射關系能夠通過實例的名稱查詢到緩存實例對象并對緩存實例對象進行各類的操作。
Caffeine實例的命名功能是其他功能整合的基石。

業務層-內存占用的預估

import jdk.nashorn.internal.ir.debug.ObjectSizeCalculator;
 
public abstract class BoundedLocalCache<K, V> extends BLCHeader.DrainStatusRef<K, V>
    implements LocalCache<K, V> {
 
  final ConcurrentHashMap<Object, Node<K, V>> data;
 
  @Override
  public long getMemoryUsed() {
    // 預估內存占用
    return ObjectSizeCalculator.getObjectSize(data);
  }
}

說明：

通過ObjectSizeCalculator.getObjectSize預估內存的緩存值。
data值是Caffeine實例用來保存真實數據的對象。

業務層-數據上報機制

public static StatsData getCacheStats(String instanceName) {
 
    Cache cache = Caffeine.getCacheByInstanceName(instanceName);
 
    CacheStats cacheStats = cache.stats();
    StatsData statsData = new StatsData();
 
    statsData.setInstanceName(instanceName);
    statsData.setTimeStamp(System.currentTimeMillis()/1000);
    statsData.setMemoryUsed(String.valueOf(cache.getMemoryUsed()));
    statsData.setEstimatedSize(String.valueOf(cache.estimatedSize()));
    statsData.setRequestCount(String.valueOf(cacheStats.requestCount()));
    statsData.setHitCount(String.valueOf(cacheStats.hitCount()));
    statsData.setHitRate(String.valueOf(cacheStats.hitRate()));
    statsData.setMissCount(String.valueOf(cacheStats.missCount()));
    statsData.setMissRate(String.valueOf(cacheStats.missRate()));
    statsData.setLoadCount(String.valueOf(cacheStats.loadCount()));
    statsData.setLoadSuccessCount(String.valueOf(cacheStats.loadSuccessCount()));
    statsData.setLoadFailureCount(String.valueOf(cacheStats.loadFailureCount()));
    statsData.setLoadFailureRate(String.valueOf(cacheStats.loadFailureRate()));
 
    Optional<Eviction> optionalEviction = cache.policy().eviction();
    optionalEviction.ifPresent(eviction -> statsData.setMaximumSize(String.valueOf(eviction.getMaximum())));
 
    Optional<Expiration> optionalExpiration = cache.policy().expireAfterWrite();
    optionalExpiration.ifPresent(expiration -> statsData.setExpireAfterWrite(String.valueOf(expiration.getExpiresAfter(TimeUnit.SECONDS))));
 
    optionalExpiration = cache.policy().expireAfterAccess();
    optionalExpiration.ifPresent(expiration -> statsData.setExpireAfterAccess(String.valueOf(expiration.getExpiresAfter(TimeUnit.SECONDS))));
 
    optionalExpiration = cache.policy().refreshAfterWrite();
    optionalExpiration.ifPresent(expiration -> statsData.setRefreshAfterWrite(String.valueOf(expiration.getExpiresAfter(TimeUnit.SECONDS))));
 
    return statsData;
}

說明：

通過Caffeine自帶的統計接口來統計相關數值。
統計數據實例維度進行統計。

public static void sendReportData() {
 
    try {
        if (!VivoConfigManager.getBoolean("memory.caffeine.data.report.switch", true)) {
            return;
        }
 
        // 1、獲取所有的cache實例對象
        Method listCacheInstanceMethod = HANDLER_MANAGER_CLASS.getMethod("listCacheInstance", null);
        List<String> instanceNames = (List)listCacheInstanceMethod.invoke(null, null);
        if (CollectionUtils.isEmpty(instanceNames)) {
            return;
        }
 
        String appName = System.getProperty("app.name");
        String localIp = getLocalIp();
        String localPort = String.valueOf(NetPortUtils.getWorkPort());
        ReportData reportData = new ReportData();
        InstanceData instanceData = new InstanceData();
        instanceData.setAppName(appName);
        instanceData.setIp(localIp);
        instanceData.setPort(localPort);
 
        // 2、遍歷cache實例對象獲取緩存監控數據
        Method getCacheStatsMethod = HANDLER_MANAGER_CLASS.getMethod("getCacheStats", String.class);
        Map<String, StatsData> statsDataMap = new HashMap<>();
        instanceNames.stream().forEach(instanceName -> {
 
            try {
                StatsData statsData = (StatsData)getCacheStatsMethod.invoke(null, instanceName);
 
                statsDataMap.put(instanceName, statsData);
            } catch (Exception e) {
 
            }
        });
 
        // 3、構建上報對象
        reportData.setInstanceData(instanceData);
        reportData.setStatsDataMap(statsDataMap);
 
        // 4、發送Http的POST請求
        HttpPost httpPost = new HttpPost(getReportDataUrl());
        httpPost.setConfig(requestConfig);
 
        StringEntity stringEntity = new StringEntity(JSON.toJSONString(reportData));
        stringEntity.setContentType("application/json");
        httpPost.setEntity(stringEntity);
 
        HttpResponse response = httpClient.execute(httpPost);
        String result = EntityUtils.toString(response.getEntity(),"UTF-8");
        EntityUtils.consume(response.getEntity());
 
        logger.info("Caffeine 數據上報成功 URL {} 參數 {} 結果 {}", getReportDataUrl(), JSON.toJSONString(reportData), result);
    } catch (Throwable throwable) {
        logger.error("Caffeine 數據上報失敗 URL {} ", getReportDataUrl(), throwable);
    }
}

說明：

通過獲取項目中運行的所有Caffeine實例并依次遍歷收集統計數據。
通過http協議負責上報對應的統計數據，采用固定間隔周期進行上報。

業務層-配置動態下發

public static ExecutionResponse dispose(ExecutionRequest request) {
    ExecutionResponse executionResponse = new ExecutionResponse();
    executionResponse.setCmdType(CmdTypeEnum.INSTANCE_CONFIGURE.getCmd());
    executionResponse.setInstanceName(request.getInstanceName());
 
    String instanceName = request.getInstanceName();
    Cache cache = Caffeine.getCacheByInstanceName(instanceName);
 
    // 設置緩存的最大條目
    if (null != request.getMaximumSize() && request.getMaximumSize() > 0) {
        Optional<Eviction> optionalEviction = cache.policy().eviction();
        optionalEviction.ifPresent(eviction ->eviction.setMaximum(request.getMaximumSize()));
    }
 
    // 設置寫后過期的過期時間
    if (null != request.getExpireAfterWrite() && request.getExpireAfterWrite() > 0) {
        Optional<Expiration> optionalExpiration = cache.policy().expireAfterWrite();
        optionalExpiration.ifPresent(expiration -> expiration.setExpiresAfter(request.getExpireAfterWrite(), TimeUnit.SECONDS));
    }
 
    // 設置訪問過期的過期時間
    if (null != request.getExpireAfterAccess() && request.getExpireAfterAccess() > 0) {
        Optional<Expiration> optionalExpiration = cache.policy().expireAfterAccess();
        optionalExpiration.ifPresent(expiration -> expiration.setExpiresAfter(request.getExpireAfterAccess(), TimeUnit.SECONDS));
    }
 
    // 設置寫更新的過期時間
    if (null != request.getRefreshAfterWrite() && request.getRefreshAfterWrite() > 0) {
 
        Optional<Expiration> optionalExpiration = cache.policy().refreshAfterWrite();
        optionalExpiration.ifPresent(expiration -> expiration.setExpiresAfter(request.getRefreshAfterWrite(), TimeUnit.SECONDS));
    }
 
    executionResponse.setCode(0);
    executionResponse.setMsg("success");
 
    return executionResponse;
}

說明：

通過Caffeine自帶接口進行緩存配置的相關設置。

業務層-緩存數據清空

/**
     * 失效緩存的值
     * @param request
     * @return
     */
    public static ExecutionResponse invalidate(ExecutionRequest request) {
 
        ExecutionResponse executionResponse = new ExecutionResponse();
        executionResponse.setCmdType(CmdTypeEnum.INSTANCE_INVALIDATE.getCmd());
        executionResponse.setInstanceName(request.getInstanceName());
 
        try {
            // 查找對應的cache實例
            String instanceName = request.getInstanceName();
            Cache cache = Caffeine.getCacheByInstanceName(instanceName);
 
            // 處理清空指定實例的所有緩存 或 指定實例的key對應的緩存
            Object cacheKeyObj = request.getCacheKey();
 
            // 清除所有緩存
            if (Objects.isNull(cacheKeyObj)) {
                cache.invalidateAll();
            } else {
                // 清除指定key對應的緩存
                if (Objects.equals(request.getCacheKeyType(), 2)) {
                    cache.invalidate(Long.valueOf(request.getCacheKey().toString()));
                } else if (Objects.equals(request.getCacheKeyType(), 3)) {
                    cache.invalidate(Integer.valueOf(request.getCacheKey().toString()));
                } else {
                    cache.invalidate(request.getCacheKey().toString());
                }
            }
 
            executionResponse.setCode(0);
            executionResponse.setMsg("success");
        } catch (Exception e) {
            executionResponse.setCode(-1);
            executionResponse.setMsg("fail");
        }
 
        return executionResponse;
    }
}

業務層-緩存數據查詢

public static ExecutionResponse inspect(ExecutionRequest request) {
 
    ExecutionResponse executionResponse = new ExecutionResponse();
    executionResponse.setCmdType(CmdTypeEnum.INSTANCE_INSPECT.getCmd());
    executionResponse.setInstanceName(request.getInstanceName());
 
    String instanceName = request.getInstanceName();
    Cache cache = Caffeine.getCacheByInstanceName(instanceName);
 
    Object cacheValue = cache.getIfPresent(request.getCacheKey());
    if (Objects.equals(request.getCacheKeyType(), 2)) {
        cacheValue = cache.getIfPresent(Long.valueOf(request.getCacheKey().toString()));
    } else if (Objects.equals(request.getCacheKeyType(), 3)) {
        cacheValue = cache.getIfPresent(Integer.valueOf(request.getCacheKey().toString()));
    } else {
        cacheValue = cache.getIfPresent(request.getCacheKey().toString());
    }
 
    if (Objects.isNull(cacheValue)) {
        executionResponse.setData("");
    } else {
        executionResponse.setData(JSON.toJSONString(cacheValue));
    }
 
    return executionResponse;
}

說明：

通過Caffeine自帶接口進行緩存信息查詢。

通信層-監聽服務

public class ServerManager {
 
    private Server jetty;
 
    /**
     * 創建jetty對象
     * @throws Exception
     */
    public ServerManager() throws Exception {
 
        int port = NetPortUtils.getAvailablePort();
 
        jetty = new Server(port);
 
        ServletContextHandler context = new ServletContextHandler(ServletContextHandler.NO_SESSIONS);
        context.setContextPath("/");
        context.addServlet(ClientServlet.class, "/caffeine");
        jetty.setHandler(context);
    }
 
    /**
     * 啟動jetty對象
     * @throws Exception
     */
    public void start() throws Exception {
        jetty.start();
    }
}
 
 
public class ClientServlet extends HttpServlet {
 
    private static final Logger logger = LoggerFactory.getLogger(ClientServlet.class);
 
    @Override
    protected void doGet(HttpServletRequest req, HttpServletResponse resp) throws ServletException, IOException {
        super.doGet(req, resp);
    }
 
    @Override
    protected void doPost(HttpServletRequest req, HttpServletResponse resp) throws ServletException, IOException {
 
        ExecutionResponse executionResponse = null;
        String requestJson = null;
        try {
            // 獲取請求的相關的參數
            String contextPath = req.getContextPath();
            String servletPath = req.getServletPath();
            String requestUri = req.getRequestURI();
            requestJson = IOUtils.toString(req.getInputStream(), StandardCharsets.UTF_8);
 
            // 處理不同的命令
            ExecutionRequest executionRequest = JSON.parseObject(requestJson, ExecutionRequest.class);
 
            // 通過反射來來處理類依賴問題
            executionResponse = DisposeCenter.dispatch(executionRequest);
 
        } catch (Exception e) {
            logger.error("vivo-memory 處理請求異常 {} ", requestJson, e);
        }
 
        if (null == executionResponse) {
            executionResponse = new ExecutionResponse();
            executionResponse.setCode(-1);
            executionResponse.setMsg("處理異常");
        }
 
        // 組裝相應報文
        resp.setContentType("application/json; charset=utf-8");
        PrintWriter out = resp.getWriter();
        out.println(JSON.toJSONString(executionResponse));
        out.flush();
    }
}

說明：

通信層通過jetty啟動http服務進行監聽，安全考慮端口不對外開放。
通過定義ClientServlet來處理相關的請求包括配置下發和緩存查詢等功能。

通信層-心跳設計

/**
 * 發送心跳數據
 */
public static void sendHeartBeatData() {
 
    try {
 
        if (!VivoConfigManager.getBoolean("memory.caffeine.heart.report.switch", true)) {
            return;
        }
 
        // 1、構建心跳數據
        String appName = System.getProperty("app.name");
        String localIp = getLocalIp();
        String localPort = String.valueOf(NetPortUtils.getWorkPort());
 
        HeartBeatData heartBeatData = new HeartBeatData();
        heartBeatData.setAppName(appName);
        heartBeatData.setIp(localIp);
        heartBeatData.setPort(localPort);
        heartBeatData.setTimeStamp(System.currentTimeMillis()/1000);
 
        // 2、發送Http的POST請求
        HttpPost httpPost = new HttpPost(getHeartBeatUrl());
        httpPost.setConfig(requestConfig);
 
        StringEntity stringEntity = new StringEntity(JSON.toJSONString(heartBeatData));
        stringEntity.setContentType("application/json");
        httpPost.setEntity(stringEntity);
 
        HttpResponse response = httpClient.execute(httpPost);
        String result = EntityUtils.toString(response.getEntity(),"UTF-8");
        EntityUtils.consume(response.getEntity());
 
        logger.info("Caffeine 心跳上報成功 URL {} 參數 {} 結果 {}", getHeartBeatUrl(), JSON.toJSONString(heartBeatData), result);
    } catch (Throwable throwable) {
        logger.error("Caffeine 心跳上報失敗 URL {} ", getHeartBeatUrl(), throwable);
    }
}

說明：

心跳功能上報項目實例的ip和端口用來通信，攜帶時間戳用來記錄上報時間戳。
實際項目中因為機器的回收等場景需要通過上報時間戳定時清理下線的服務。

五、總結

vivo技術團隊在Caffeine的使用經驗上曾有過多次分享，可參考公眾號文章《如何把 Caffeine Cache 用得如絲般順滑》，此篇文章在使用的基礎上基于使用痛點進行進一步的定制。

目前Caffeine可視化的項目已經在相關核心業務場景中落地并發揮作用，整體運行平穩。使用較多的功能包括項目維度的caffeine實例的全局管控，單實例維度的內存占用評估和緩存命中趨勢評估。

如通過單實例的內存占用評估功能能夠合理評估緩存條目設置和內存占用之間的關系；通過分析緩存命中率的整體趨勢評估緩存的參數設置合理性。

期待此篇文章能夠給業界緩存使用和監控帶來一些新思路。

責任編輯：龐桂玉來源： vivo互聯網技術

緩存框架

成人免费xxxxx在线视频软件_久久精品久久久_亚洲国产精品久久久_天天色天天色_亚洲人成一区_欧美一级欧美三级在线观看

緩存框架 Caffeine 的可視化探索與實踐

一、背景

二、業務的技術視角

三、可視化能力

3.1 緩存實例的全局管控

3.2 內存占用趨勢

3.3 命中率趨勢

3.4 配置變更

3.5 緩存查詢

四、原理實現

4.1 整體設計框架

4.2 源碼實現

五、總結