緩存有大key?你得知道的一些手段

??

?

背景：

最近系統(tǒng)內(nèi)緩存CPU使用率一直報(bào)警，超過(guò)設(shè)置的70%報(bào)警閥值，針對(duì)此場(chǎng)景，需要對(duì)應(yīng)解決緩存是否有大key使用問(wèn)題，掃描緩存集群的大key，針對(duì)每個(gè)key做優(yōu)化處理。

以下是掃描出來(lái)的大key，此處只放置了有效關(guān)鍵信息。

??

圖1

大key介紹:

想要解決大key，首先我們得知道什么定義為大key。

什么是大KEY：

大key 并不是指 key 的值很大，而是 key 對(duì)應(yīng)的 value 很大（非常占內(nèi)存）。此處為中間件給出的定義：

?單個(gè)String類型的Key大小達(dá)到20KB并且OPS高

?單個(gè)String達(dá)到100KB

?集合類型的Key總大小達(dá)到1MB

?集合類型的Key中元素超過(guò)5000個(gè)

大KEY帶來(lái)的影響：

知道了大key的定義，那么我們也得知道大key的帶來(lái)的影響：

?客戶端超時(shí)阻塞。 Redis 執(zhí)行命令是單線程處理，然后在大 key處理時(shí)會(huì)比較耗時(shí)，那么就會(huì)發(fā)生阻塞 ，期間就會(huì)各種業(yè)務(wù)超時(shí)出現(xiàn)。

?引發(fā)網(wǎng)絡(luò)阻塞。每次獲取大 key 產(chǎn)生的網(wǎng)絡(luò)流量較大，如果一個(gè) key 的大小是 1 MB，每秒訪問(wèn)量為 1000，那么每秒會(huì)產(chǎn)生 1000MB 的流量，這對(duì)于服務(wù)器來(lái)說(shuō)是災(zāi)難性的。

?阻塞工作線程。如果使用 del 刪除大 key 時(shí)，會(huì)阻塞工作線程，無(wú)法處理后續(xù)的命令。

?內(nèi)存分布不均。集群各分片內(nèi)存使用不均。某個(gè)分片占用內(nèi)存較高或OOM，發(fā)送緩存區(qū)增大等，導(dǎo)致該分片其他Key被逐出，同時(shí)也會(huì)造成其他分片的資源浪費(fèi)。

大KEY解決手段：

1、歷史key未使用

場(chǎng)景描述：

針對(duì)這種key場(chǎng)景，其實(shí)存在著歷史原因，可能是伴隨著某個(gè)業(yè)務(wù)下線或者不使用，往往對(duì)應(yīng)實(shí)現(xiàn)的緩存操作代碼會(huì)刪除，但是對(duì)于緩存數(shù)據(jù)往往不會(huì)做任何處理，久而久之，這種臟數(shù)據(jù)會(huì)一直堆積，占用著資源。那么如果確定已經(jīng)無(wú)使用，并且可以確認(rèn)有持久化數(shù)據(jù)（如mysql、es等）備份的話，可以直接將對(duì)應(yīng)key刪除。

實(shí)例經(jīng)驗(yàn)：

如圖1上面的元素個(gè)數(shù)488649，其實(shí)整個(gè)系統(tǒng)查看了下，沒(méi)有使用的地方，最近也沒(méi)有訪問(wèn)，相信也是因?yàn)橐恢睕](méi)有用到， 否則系統(tǒng)內(nèi)一旦用了這個(gè)key來(lái)操作hgetall、smembers等，那么緩存服務(wù)應(yīng)該就會(huì)不可用了。

2、元素?cái)?shù)過(guò)多

場(chǎng)景描述：

針對(duì)于Set、HASH這種場(chǎng)景，如果元素?cái)?shù)量超過(guò)5000就視為大的key,以上面圖1為例，可以看到元素個(gè)數(shù)有的甚至達(dá)到了1萬(wàn)以上。針對(duì)這種的如果對(duì)應(yīng)value值不大，我們可以采取平鋪的形式，

實(shí)例經(jīng)驗(yàn)：

比如系統(tǒng)內(nèi)歷史的設(shè)計(jì)是存儲(chǔ)下每個(gè)品牌對(duì)應(yīng)的名稱，那么就設(shè)置了統(tǒng)一的key，然后不同的品牌id作為fild，操作了hSet和hGet來(lái)存儲(chǔ)獲取數(shù)據(jù)，降低查詢外圍服務(wù)的頻率。但是隨著品牌數(shù)量的增長(zhǎng)，導(dǎo)致元素逐步增多，元素個(gè)數(shù)就超過(guò)了大key的預(yù)設(shè)值了。這種根據(jù)場(chǎng)景，我們其實(shí)存儲(chǔ)本身只有一個(gè)品牌名稱，那么我們就針對(duì)于品牌id對(duì)應(yīng)加上一個(gè)統(tǒng)一前綴作為唯一key，采用平鋪方式緩存對(duì)應(yīng)數(shù)據(jù)即可。那么針對(duì)這種數(shù)據(jù)的替換，我這里也總結(jié)了下具體要實(shí)現(xiàn)的步驟：

修改代碼查詢和賦值邏輯：

?把原始的hGet的邏輯修改為get獲??；

?把原始hSet的邏輯修改為set賦值。

歷史數(shù)據(jù)刷新到新緩存key：

為了避免上線之后出現(xiàn)緩存雪崩，因?yàn)樘鎿Q了新的key，我們需要通過(guò)現(xiàn)有的HASH的數(shù)據(jù)刷新到新的緩存中，所以需要?dú)v史數(shù)據(jù)處理。

通過(guò)hGetAll獲取所以元素?cái)?shù)據(jù)

循環(huán)緩存元素?cái)?shù)據(jù)操作存儲(chǔ)新的緩存key和value。

public String refreshHistoryData(){
    try {
        String key = "historyKey";
        Map redisInfoMap= redisUtils.hGetAll(key);
        if (redisInfoMap.isEmpty()){
            return "查詢緩存無(wú)數(shù)據(jù)";
        }
        for (Map.Entry entry : redisInfoMap.entrySet()) {
            String redisVal = entry.getValue();
            String filedKey = entry.getKey();
            String newDataRedisKey = "newDataKey"+filedKey;
            redisUtils.set(newDataRedisKey,redisVal);
        }
        return "success";
    }catch (Exception e){
        LOG.error("refreshHistoryData 異常:",e);
    }
    return "failed";
}

注意:這里一定要先刷歷史數(shù)據(jù)，再上線代碼業(yè)務(wù)邏輯的修改。防止引發(fā)緩存雪崩

3、大對(duì)象轉(zhuǎn)換存儲(chǔ)形式

場(chǎng)景描述：

復(fù)雜的大對(duì)象可以嘗試將對(duì)象分拆成幾個(gè)key-value， 使用mGet和mSet操作對(duì)應(yīng)值或者pipeline的形式，最后拼裝成需要返回的大對(duì)象。這樣意義在于可以分散單次操作的壓力，將操作壓力平攤到多個(gè)redis實(shí)例中，降低對(duì)單個(gè)redis的IO影響；

實(shí)例經(jīng)驗(yàn)：

這里以系統(tǒng)內(nèi)訂單對(duì)象為例：訂單對(duì)象Order基礎(chǔ)屬性有幾十個(gè)，如訂單號(hào)、金額、時(shí)間、類型等，除此之外還要包含訂單下的商品OrderSub、預(yù)售信息PresaleOrder、發(fā)票信息OrderInvoice、訂單時(shí)效OrderPremiseInfo、訂單軌跡OrderTrackInfo、訂單詳細(xì)費(fèi)用OrderFee等信息。

那么對(duì)于每個(gè)訂單相關(guān)信息，我們可以設(shè)置為單獨(dú)的key，把訂單信息和幾個(gè)相關(guān)的關(guān)聯(lián)數(shù)據(jù)每個(gè)按照單獨(dú)key存儲(chǔ)，接著通過(guò)mGet方式獲取每個(gè)信息之后，最后封裝成整體Order對(duì)象。下面僅展示關(guān)鍵偽代碼以mSet和mGet實(shí)現(xiàn)：

緩存定義：

public enum CacheKeyConstant {

    /**
     * 訂單基礎(chǔ)緩存key
     */
    REDIS_ORDER_BASE_INFO("ORDER_BASE_INFO"),

    /**
     * 訂單商品緩存key
     */
    ORDER_SUB_INFO("ORDER_SUB_INFO"),

    /**
     * 訂單預(yù)售信息緩存key
     */
    ORDER_PRESALE_INFO("ORDER_PRESALE_INFO"),

    /**
     * 訂單履約信息緩存key
     */
    ORDER_PREMISE_INFO("ORDER_PREMISE_INFO"),

    /**
     * 訂單發(fā)票信息緩存key
     */
    ORDER_INVOICE_INFO("ORDER_INVOICE_INFO"),

    /**
     * 訂單軌跡信息緩存key
     */
    ORDER_TRACK_INFO("ORDER_TRACK_INFO"),

    /**
     * 訂單詳細(xì)費(fèi)用信息緩存key
     */
    ORDER_FEE_INFO("ORDER_FEE_INFO"),
    ;
    /**
     * 前綴
     */
    private String prefix;

    /**
     * 項(xiàng)目統(tǒng)一前綴
     */
    public static final String COMMON_PREFIX = "XXX";


    CacheKeyConstant(String prefix){
        this.prefix = prefix;
    }

    public String getPrefix(String subKey) {
        if(StringUtil.isNotEmpty(subKey)){
            return COMMON_PREFIX + prefix + "_" + subKey;
        }
        return COMMON_PREFIX + prefix;
    }

    public String getPrefix() {
        return COMMON_PREFIX + prefix;
    }
}

緩存存儲(chǔ)：

/**
 * @description 刷新訂單到緩存
 * @param order 訂單信息
 */
public boolean refreshOrderToCache(Order order){
     if(order == null || order.getOrderId() == null){
        return ;
    }
    String orderId = order.getOrderId().toString();
    //設(shè)置存儲(chǔ)緩存數(shù)據(jù)
    Map cacheOrderMap = new HashMap(16);
    cacheOrderMap.put(CacheKeyConstant.ORDER_BASE_INFO.getPrefix(orderId), JSON.toJSONString(buildBaseOrderVo(order)));
    cacheOrderMap.put(CacheKeyConstant.ORDER_SUB_INFO.getPrefix(orderId), JSON.toJSONString(order.getCustomerOrderSubs()));
    cacheOrderMap.put(CacheKeyConstant.ORDER_PRESALE_INFO.getPrefix(orderId), JSON.toJSONString(order.getPresaleOrderData()));
    cacheOrderMap.put(CacheKeyConstant.ORDER_INVOICE_INFO.getPrefix(orderId), JSON.toJSONString(order.getOrderInvoice()));
    cacheOrderMap.put(CacheKeyConstant.ORDER_TRACK_INFO.getPrefix(orderId), JSON.toJSONString(order.getOrderTrackInfo()));
    cacheOrderMap.put(CacheKeyConstant.ORDER_PREMISE_INFO.getPrefix(orderId), JSON.toJSONString( order.getPresaleOrderData()));
    cacheOrderMap.put(CacheKeyConstant.ORDER_FEE_INFO.getPrefix(orderId), JSON.toJSONString(order.getOrderFeeVo()));
    superRedisUtils.mSetString(cacheOrderMap);
}

緩存獲?。?/p>

/**
 * @description 通過(guò)訂單號(hào)獲取緩存數(shù)據(jù)
 * @param orderId 訂單號(hào)
 * @return Order 訂單實(shí)體信息
 */
public Order getOrderFromCache(String orderId){
    if(StringUtils.isBlank(orderId)){
            return null;
    }
    //定義查詢緩存集合key
    List queryOrderKey = Arrays.asList(CacheKeyConstant.ORDER_BASE_INFO.getPrefix(orderId),CacheKeyConstant.ORDER_SUB_INFO.getPrefix(orderId),
            CacheKeyConstant.ORDER_PRESALE_INFO.getPrefix(orderId),CacheKeyConstant.ORDER_INVOICE_INFO.getPrefix(orderId),CacheKeyConstant.ORDER_TRACK_INFO.getPrefix(orderId),
            CacheKeyConstant.ORDER_PREMISE_INFO.getPrefix(orderId),CacheKeyConstant.ORDER_FEE_INFO.getPrefix(orderId));

    //查詢結(jié)果
    List result = redisUtils.mGet(queryOrderKey);
    //基礎(chǔ)信息
    if(CollectionUtils.isEmpty(result)){
        return null;
    }
    String[] resultInfo = result.toArray(new String[0]);

    //基礎(chǔ)信息
    if(StringUtils.isBlank(resultInfo[0])){
        return null;
    }
    BaseOrderVo baseOrderVo = JSON.parseObject(resultInfo[0],BaseOrderVo.class);
    Order order = coverBaseOrderVoToOrder(baseOrderVo);

    //訂單商品
    if(StringUtils.isNotBlank(resultInfo[1])){
        List orderSubs =JSON.parseObject(result.get(1), new TypeReference>(){});
        order.setCustomerOrderSubs(orderSubs);
    }
    //訂單預(yù)售
    if(StringUtils.isNotBlank(resultInfo[2])){
        PresaleOrderData presaleOrderData = JSON.parseObject(resultInfo[2],PresaleOrderData.class);
        order.setPresaleOrderData(presaleOrderData);
    }
    //訂單發(fā)票
    if(StringUtils.isNotBlank(resultInfo[3])){
        OrderInvoice orderInvoice = JSON.parseObject(resultInfo[3],OrderInvoice.class);
        order.setOrderInvoice(orderInvoice);
    }
    //訂單軌跡
    if(StringUtils.isNotBlank(resultInfo[5])){
        OrderTrackInfo orderTrackInfo = JSON.parseObject(resultInfo[5],OrderTrackInfo.class);
        order.setOrderTrackInfo(orderTrackInfo);
    }
    //訂單履約信息
    if(StringUtils.isNotBlank(resultInfo[6])){
        List orderPremiseInfos =JSON.parseObject(result.get(6), new TypeReference>(){});
        order.setPremiseInfos(orderPremiseInfos);
    }
    //訂單費(fèi)用明細(xì)信息
    if(StringUtils.isNotBlank(resultInfo[7])){
        OrderFeeVo orderFeeVo = JSON.parseObject(resultInfo[7],OrderFeeVo.class);
        order.setOrderFeeVo(orderFeeVo);
    }
    return order;
}

注意：獲取緩存的結(jié)果跟傳入的key的順序保持對(duì)應(yīng)即可。

緩存util方法封裝：

/**
 *
 * @description 同時(shí)將多個(gè) key-value (域-值)對(duì)設(shè)置到緩存中。
 * @param mappings 需要插入的數(shù)據(jù)信息
 */
public void mSetString(Map mappings) {
    CallerInfo callerInfo = Ump.methodReg(UmpKeyConstants.REDIS.REDIS_STATUS_READ_MSET);
    try {
        redisClient.getClientInstance().mSetString(mappings);
    } catch (Exception e) {
        Ump.funcError(callerInfo);
    }finally {
        Ump.methodRegEnd(callerInfo);
    }
}
/**
 *
 * @description 同時(shí)將多個(gè)key的結(jié)果返回。
 * @param queryKeys 查詢的緩存key集合
 */
public List mGet(List queryKeys) {
    CallerInfo callerInfo = Ump.methodReg(UmpKeyConstants.REDIS.REDIS_STATUS_READ_MGET);
    try {
        return redisClient.getClientInstance().mGet(queryKeys.toArray(new String[0]));
    } catch (Exception e) {
        Ump.funcError(callerInfo);
    }finally {
        Ump.methodRegEnd(callerInfo);
    }
    return new ArrayList(queryKeys.size());
}

這里附上通過(guò)pipeline的util封裝，可參考。

/**
 * @description pipeline放松查詢數(shù)據(jù)
 * @param redisKeyList
 * @return java.util.List
 */
public List getValueByPipeline(List redisKeyList) {
        if(CollectionUtils.isEmpty(redisKeyList)){
            return null;
        }
        List resultInfo = new ArrayList(redisKeyList);
        CallerInfo callerInfo = Ump.methodReg(UmpKeyConstants.REDIS.REDIS_STATUS_READ_GET);
        try {

            PipelineClient pipelineClient = redisClient.getClientInstance().pipelineClient();

            //添加批量查詢?nèi)蝿?wù)
            List futures = new ArrayList();
            redisKeyList.forEach(redisKey -> {
                futures.add(pipelineClient.get(redisKey.getBytes()));
            });
            //處理查詢結(jié)果
            pipelineClient.flush();
            //可以等待future的返回結(jié)果，來(lái)判斷命令是否成功。
            for (JimFuture future : futures) {
                resultInfo.add(new String((byte[])future.get()));
            }

        } catch (Exception e) {
            log.error("getValueByPipeline error:",e);
            Ump.funcError(callerInfo);
            return new ArrayList(redisKeyList.size());
        }finally {
            Ump.methodRegEnd(callerInfo);
        }
        return resultInfo;
    }

注意：Pipeline不建議用來(lái)設(shè)置緩存值，因?yàn)楸旧聿皇窃有缘牟僮鳌?/p>

4、壓縮存儲(chǔ)數(shù)據(jù)

壓縮方法結(jié)果：

單個(gè)元素時(shí)：

??

四百個(gè)元素集合：

??

四萬(wàn)個(gè)元素集合時(shí)：

??

壓縮代碼樣例

DefaultOutputStream

public static byte[] compressToByteArray(String text) throws IOException {
    ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
    Deflater deflater = new Deflater();
    DeflaterOutputStream deflaterOutputStream = new DeflaterOutputStream(outputStream, deflater);

    deflaterOutputStream.write(text.getBytes());
    deflaterOutputStream.close();

    return outputStream.toByteArray();
}

public static String decompressFromByteArray(byte[] bytes) throws IOException {
    ByteArrayInputStream inputStream = new ByteArrayInputStream(bytes);
    Inflater inflater = new Inflater();
    InflaterInputStream inflaterInputStream = new InflaterInputStream(inputStream, inflater);
    ByteArrayOutputStream outputStream = new ByteArrayOutputStream();

    byte[] buffer = new byte[1024];
    int length;
    while ((length = inflaterInputStream.read(buffer)) != -1) {
        outputStream.write(buffer, 0, length);
    }

    inflaterInputStream.close();
    outputStream.close();

    byte[] decompressedData = outputStream.toByteArray();
    return new String(decompressedData);
}

GZIPOutputStream

public static byte[] compressGzip(String str) {
        ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
        GZIPOutputStream gzipOutputStream = null;
        try {
            gzipOutputStream = new GZIPOutputStream(outputStream);
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
        try {
            gzipOutputStream.write(str.getBytes("UTF-8"));
        } catch (IOException e) {
            throw new RuntimeException(e);
        }finally {
            try {
                gzipOutputStream.close();
            } catch (IOException e) {
                throw new RuntimeException(e);
            }
        }
        return outputStream.toByteArray();
    }

 public static String decompressGzip(byte[] compressed) throws IOException {
        ByteArrayInputStream inputStream = new ByteArrayInputStream(compressed);
        GZIPInputStream gzipInputStream = new GZIPInputStream(inputStream);
        ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
        byte[] buffer = new byte[1024];
        int length;
        while ((length = gzipInputStream.read(buffer)) > 0) {
            outputStream.write(buffer, 0, length);
        }
        gzipInputStream.close();
        outputStream.close();
        return outputStream.toString("UTF-8");
    }

ZlibCompress

 public  byte[] zlibCompress(String message) throws Exception {
        String chatacter = "UTF-8";
        byte[] input = message.getBytes(chatacter);
        BigDecimal bigDecimal = BigDecimal.valueOf(0.25f);
        BigDecimal length = BigDecimal.valueOf(input.length);
        byte[] output = new byte[input.length + 10 + new Double(Math.ceil(Double.parseDouble(bigDecimal.multiply(length).toString()))).intValue()];
        Deflater compresser = new Deflater();
        compresser.setInput(input);
        compresser.finish();
        int compressedDataLength = compresser.deflate(output);
        compresser.end();
        return Arrays.copyOf(output, compressedDataLength);
    }

public static String zlibInfCompress(byte[] data) {
        String s = null;

        Inflater decompresser = new Inflater();
        decompresser.reset();
        decompresser.setInput(data);
        ByteArrayOutputStream o = new ByteArrayOutputStream(data.length);
        try {
            byte[] buf = new byte[1024];
            while (!decompresser.finished()) {
                int i = decompresser.inflate(buf);
                o.write(buf, 0, i);
            }
            s = o.toString("UTF-8");
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            try {
                o.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
        decompresser.end();
        return s;
    }

可以看到壓縮效率比較好，壓縮效率可以從幾百kb壓縮到幾kb內(nèi)；當(dāng)然也是看具體場(chǎng)景。不過(guò)這里就是最好是避免調(diào)用量大的場(chǎng)景使用，畢竟解壓和壓縮數(shù)據(jù)量大會(huì)比較耗費(fèi)cpu性能。如果是黃金鏈路使用，還需要具體配合壓測(cè)，對(duì)比前后接口性能。

5、替換存儲(chǔ)方案

如果數(shù)據(jù)量龐大，那么其實(shí)本身是不是就不太適合redis這種緩存存儲(chǔ)了?？梢钥紤]es或者mongo這種文檔式存儲(chǔ)結(jié)構(gòu)，存儲(chǔ)大的數(shù)據(jù)格式。

總結(jié)：

redis緩存的使用是一個(gè)支持業(yè)務(wù)和功能高并發(fā)的很好的使用方案，但是隨著使用場(chǎng)景的多樣性以及數(shù)據(jù)的增加，可能逐漸的會(huì)出現(xiàn)大key，日常使用中都可以注意以下幾點(diǎn)：

1.分而治之：如果需要存儲(chǔ)大量的數(shù)據(jù)，避免直接放到緩存中?？梢詫⑵洳鸱殖啥鄠€(gè)小的value。就像是咱們?nèi)粘３燥?，盛到碗里，一口一口的吃，俗話說(shuō)的好呀：“細(xì)嚼慢咽”。

2.避免使用不必要的數(shù)據(jù)結(jié)構(gòu)。例如，如果只需要存儲(chǔ)一個(gè)字符串結(jié)構(gòu)的數(shù)據(jù)，就不要過(guò)度設(shè)計(jì)，使用Hash或者List等數(shù)據(jù)結(jié)構(gòu)。

3.定期清理過(guò)期的key。如果Redis中存在大量的過(guò)期key，就會(huì)導(dǎo)致Redis的性能下降，或者場(chǎng)景非必要以緩存來(lái)持久存儲(chǔ)的，可以添加過(guò)期時(shí)間，定時(shí)清理過(guò)期的key，就像是家中的日常垃圾類似，定期的清潔和打掃，居住起來(lái)咱們才會(huì)更加舒服和方便。

4.對(duì)象壓縮。將大的數(shù)據(jù)壓縮成更小的數(shù)據(jù)，也是一種好的解決方案，不過(guò)要注意壓縮和解壓的頻率，畢竟是比較耗費(fèi)cpu的。

以上是我根據(jù)現(xiàn)有實(shí)際場(chǎng)景總結(jié)出的一些解決手段，記錄了這些大key的優(yōu)化經(jīng)驗(yàn)，希望可以在日常場(chǎng)景中幫助到大家。大家有其他的好的經(jīng)驗(yàn)，也可以分享出來(lái)。

審核編輯 黃宇