本文作者以实践项目遇到的大key问题为线索，场景化地叙述对应的处理计划。经过本文，您能够了解关于大key根底概念、影响以及遇到大key的详细处理手法，帮助您更好把控缓存的运用场景，从而提升软件系统的稳定性。

缓存有大key?你得知道的一些手法
​

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

背景：

最近系统内缓存CPU运用率一向报警，超越设置的70%报警阀值，针对此场景，需求对应处理缓存是否有大key运用问题，扫描缓存集群的大key,针对每个key做优化处理。

以下是扫描出来的大key，此处只放置了有用关键信息。

缓存有大key?你得知道的一些手法
​

大key介绍:

想要处理大key，首要我们得知道什么界说为大key。

什么是大KEY：

大key 并不是指 key 的值很大，而是 key 对应的 value 很大（十分占内存）。此处为中间件给出的界说：

•单个String类型的Key巨细到达20KB而且OPS高

•单个String到达100KB

•调集类型的Key总巨细到达1MB

•调集类型的Key中元素超越5000个

大KEY带来的影响：

知道了大key的界说，那么我们也得知道大key的带来的影响：

•客户端超时堵塞。 Redis 执行命令是单线程处理，然后在大 key处理时会比较耗时，那么就会发生堵塞 ，期间就会各种事务超时呈现。

•引发网络堵塞。每次获取大 key 发生的网络流量较大，假如一个 key 的巨细是 1 MB，每秒拜访量为 1000，那么每秒会发生 1000MB 的流量，这关于服务器来说是灾难性的。

•堵塞工作线程。假如运用 del 删去大 key 时，会堵塞工作线程，无法处理后续的命令。

•内存分布不均。集群各分片内存运用不均。某个分片占用内存较高或OOM，发送缓存区增大等，导致该分片其他Key被逐出，同时也会形成其他分片的资源浪费。

大KEY处理手法：

1、前史key未运用

场景描绘：

针对这种key场景，其实存在着前史原因，或许是伴跟着某个事务下线或许不运用，往往对应完成的缓存操作代码会删去，可是关于缓存数据往往不会做任何处理，一朝一夕，这种脏数据会一向堆积，占用着资源。那么假如确定已经无运用，而且能够确认有耐久化数据（如mysql、es等）备份的话，能够直接将对应key删去。

实例经历：

如图1上面的元素个数488649，其实整个系统查看了下，没有运用的当地，最近也没有拜访，相信也是由于一向没有用到， 否则系统内一旦用了这个key来操作hgetall、smembers等，那么缓存服务应该就会不可用了。

2、元素数过多

场景描绘：

针关于Set、HASH这种场景，假如元素数量超越5000就视为大的key，以上面图1为例，能够看到元素个数有的甚至到达了1万以上。针对这种的假如对应value值不大，我们能够采取平铺的形式，

实例经历：

比方系统内前史的规划是存储下每个品牌对应的名称，那么就设置了统一的key，然后不同的品牌id作为fild，操作了hSet和hGet来存储获取数据，下降查询外围服务的频率。可是跟着品牌数量的增加，导致元素逐渐增多，元素个数就超越了大key的预设值了。这种依据场景，我们其实存储自身只有一个品牌名称，那么我们就针关于品牌id对应加上一个统一前缀作为仅有key，选用平铺办法缓存对应数据即可。那么针对这种数据的替换，我这儿也总结了下详细要完成的步骤：

修正代码查询和赋值逻辑：

•把原始的hGet的逻辑修正为get获取；

•把原始hSet的逻辑修正为set赋值。

前史数据改写到新缓存key：

为了防止上线之后呈现缓存雪崩，由于替换了新的key，我们需求经过现有的HASH的数据改写到新的缓存中，所以需求前史数据处理。

经过hGetAll获取所以元素数据

循环缓存元素数据操作存储新的缓存key和value。

public String refreshHistoryData(){
    try {
        String key = "historyKey";
        Map<String, String> redisInfoMap= redisUtils.hGetAll(key);
        if (redisInfoMap.isEmpty()){
            return "查询缓存无数据";
        }
        for (Map.Entry<String, String> 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";
}

留意:这儿一定要先刷前史数据，再上线代码事务逻辑的修正。防止引发 缓存雪崩

3、大目标转化存储形式

场景描绘：

杂乱的大目标能够尝试将目标分拆成几个key-value， 运用mGet和mSet操作对应值或许pipeline的形式，最终拼装成需求回来的大目标。这样含义在于能够分散单次操作的压力，将操作压力平摊到多个redis实例中，下降对单个redis的IO影响；

实例经历：

这儿以系统内订单目标为例：订单目标Order根底属性有几十个，如订单号、金额、时刻、类型等，除此之外还要包含订单下的产品OrderSub、预售信息PresaleOrder、发票信息OrderInvoice、订单时效OrderPremiseInfo、订单轨道OrderTrackInfo、订单详细费用OrderFee等信息。

那么关于每个订单相关信息，我们能够设置为独自的key，把订单信息和几个相关的相关数据每个依照独自key存储，接着经过mGet办法获取每个信息之后，最终封装成整体Order目标。下面仅展现关键伪代码以mSet和mGet完成：

缓存界说：

public enum CacheKeyConstant {
    /**
     * 订单根底缓存key
     */
    REDIS_ORDER_BASE_INFO("ORDER_BASE_INFO"),
    /**
     * 订单产品缓存key
     */
    ORDER_SUB_INFO("ORDER_SUB_INFO"),
    /**
     * 订单预售信息缓存key
     */
    ORDER_PRESALE_INFO("ORDER_PRESALE_INFO"),
    /**
     * 订单履约信息缓存key
     */
    ORDER_PREMISE_INFO("ORDER_PREMISE_INFO"),
    /**
     * 订单发票信息缓存key
     */
    ORDER_INVOICE_INFO("ORDER_INVOICE_INFO"),
    /**
     * 订单轨道信息缓存key
     */
    ORDER_TRACK_INFO("ORDER_TRACK_INFO"),
    /**
     * 订单详细费用信息缓存key
     */
    ORDER_FEE_INFO("ORDER_FEE_INFO"),
    ;
    /**
     * 前缀
     */
    private String prefix;
    /**
     * 项目统一前缀
     */
    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;
    }
}

缓存存储：

/**
 * @description 改写订单到缓存
 * @param order 订单信息
 */
public boolean refreshOrderToCache(Order order){
     if(order == null || order.getOrderId() == null){
        return ;
    }
    String orderId = order.getOrderId().toString();
    //设置存储缓存数据
    Map<String,String> 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);
}

缓存获取：

/**
 * @description 经过订单号获取缓存数据
 * @param orderId 订单号
 * @return Order 订单实体信息
 */
public Order getOrderFromCache(String orderId){
    if(StringUtils.isBlank(orderId)){
            return null;
    }
    //界说查询缓存调集key
    List<String> 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));
    //查询成果
    List<String> result = redisUtils.mGet(queryOrderKey);
    //根底信息
    if(CollectionUtils.isEmpty(result)){
        return null;
    }
    String[] resultInfo = result.toArray(new String[0]);
    //根底信息
    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<OrderSub> orderSubs =JSON.parseObject(result.get(1), new TypeReference<List<OrderSub>>(){});
        order.setCustomerOrderSubs(orderSubs);
    }
    //订单预售
    if(StringUtils.isNotBlank(resultInfo[2])){
        PresaleOrderData presaleOrderData = JSON.parseObject(resultInfo[2],PresaleOrderData.class);
        order.setPresaleOrderData(presaleOrderData);
    }
    //订单发票
    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<OrderPremiseInfo> orderPremiseInfos =JSON.parseObject(result.get(6), new TypeReference<List<OrderPremiseInfo>>(){});
        order.setPremiseInfos(orderPremiseInfos);
    }
    //订单费用明细信息
    if(StringUtils.isNotBlank(resultInfo[7])){
        OrderFeeVo orderFeeVo = JSON.parseObject(resultInfo[7],OrderFeeVo.class);
        order.setOrderFeeVo(orderFeeVo);
    }
    return order;
}

留意：获取缓存的成果跟传入的key的顺序保持对应即可。

缓存util办法封装：

/**
 *
 * @description 同时将多个 key-value (域-值)对设置到缓存中。
 * @param mappings 需求插入的数据信息
 */
public void mSetString(Map<String, String> 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 同时将多个key的成果回来。
 * @param queryKeys 查询的缓存key调集
 */
public List<String> mGet(List<String> 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<String>(queryKeys.size());
}

这儿附上经过pipeline的util封装，可参阅。

/**
 * @description pipeline放松查询数据
 * @param redisKeyList
 * @return java.util.List<java.lang.String>
 */
public List<String> getValueByPipeline(List<String> redisKeyList) {
        if(CollectionUtils.isEmpty(redisKeyList)){
            return null;
        }
        List<String> resultInfo = new ArrayList<>(redisKeyList);
        CallerInfo callerInfo = Ump.methodReg(UmpKeyConstants.REDIS.REDIS_STATUS_READ_GET);
        try {
            PipelineClient pipelineClient = redisClient.getClientInstance().pipelineClient();
            //增加批量查询使命
            List<JimFuture> futures = new ArrayList<>();
            redisKeyList.forEach(redisKey -> {
                futures.add(pipelineClient.get(redisKey.getBytes()));
            });
            //处理查询成果
            pipelineClient.flush();
            //能够等待future的回来成果，来判别命令是否成功。
            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不主张用来设置缓存值，由于自身不是原子性的操作。

4、紧缩存储数据

紧缩办法成果：

单个元素时：

缓存有大key?你得知道的一些手法
​

紧缩办法 紧缩前巨细Byte 紧缩后巨细Byte 紧缩耗时 解压耗时 紧缩解压后比对成果
DefaultOutputStream 446（0.43kb） 254 （0.25kb） 1ms 0ms 相同
GzipOutputStream 446（0.43kb） 266 (0.25kbM) 1ms 1ms 相同
ZlibCompress 446（0.43kb） 254 （0.25kb） 1ms 0ms 相同
四百个元素调集：

缓存有大key?你得知道的一些手法
​

紧缩办法 紧缩前巨细Byte 紧缩后巨细Byte 紧缩耗时 解压耗时 紧缩解压后比对成果
DefaultOutputStream 6732（6.57kb） 190 （0.18kb） 2ms 0ms 相同
GzipOutputStream 6732（6.57kb） 202 (0.19kb) 1ms 1ms 相同
ZlibCompress 6732（6.57kb） 190 （0.18kb） 1ms 0ms 相同
四万个元素调集时：

缓存有大key?你得知道的一些手法
​

紧缩办法 紧缩前巨细Byte 紧缩后巨细Byte 紧缩耗时 解压耗时 紧缩解压后比对成果
DefaultOutputStream 640340（625kb） 1732 （1.69kb） 37ms 2ms 相同
GzipOutputStream 640340（625kb） 1744 (1.70kb) 11ms 3ms 相同
ZlibCompress 640340（625kb） 1732 （1.69kb） 69ms 2ms 相同

紧缩代码样例

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内；当然也是看详细场景。不过这儿就是最好是防止调用量大的场景运用，究竟解压和紧缩数据量大会比较消耗cpu功用。假如是黄金链路运用，还需求详细配合压测，比照前后接口功用。

5、替换存储计划

假如数据量巨大，那么其实自身是不是就不太合适redis这种缓存存储了。能够考虑es或许mongo这种文档式存储结构，存储大的数据格式。

总结：

redis缓存的运用是一个支持事务和功用高并发的很好的运用计划，可是跟着运用场景的多样性以及数据的增加，或许逐渐的会呈现大key，日常运用中都能够留意以下几点：

1.分而治之：假如需求存储大量的数据，防止直接放到缓存中。能够将其拆分成多个小的value。就像是我们日常吃饭，盛到碗里，一口一口的吃，俗语说的好呀：“细嚼慢咽”。

2.防止运用不必要的数据结构。例如，假如只需求存储一个字符串结构的数据，就不要过度规划，运用Hash或许List等数据结构。

3.定期整理过期的key。假如Redis中存在大量的过期key，就会导致Redis的功用下降，或许场景非必要以缓存来耐久存储的，能够增加过期时刻，定时整理过期的key，就像是家中的日常垃圾类似，定期的清洁和打扫，居住起来我们才会愈加舒服和便利。

4.目标紧缩。将大的数据紧缩成更小的数据，也是一种好的处理计划，不过要留意紧缩和解压的频率，究竟是比较消耗cpu的。

以上是我依据现有实践场景总结出的一些处理手法，记录了这些大key的优化经历，期望能够在日常场景中帮助到我们。我们有其他的好的经历，也能够共享出来。

作者：全途径生态 范晓

来历：京东零售技术 转载请注明来历