背景：

最近系统内缓存cpu使用率一直报警，超过设置的70%报警阀值，针对此场景，需要对应解决缓存是否有大key使用问题，扫描缓存集群的大key，针对每个key做优化处理。

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



 

 

图1

大key介绍:

想要解决大key，首先我们得知道什么定义为大key。

什么是大key：

大key 并不是指 key 的值很大，而是 key 对应的 value 很大（非常占内存）。此处为中间件给出的定义：

大key带来的影响：

知道了大key的定义，那么我们也得知道大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，采用平铺方式缓存对应数据即可。那么针对这种数据的替换，我这里也总结了下具体要实现的步骤：

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

历史数据刷新到新缓存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;
}

缓存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;
    }

4、压缩存储数据

压缩方法结果：

单个元素时：



 

 

四百个元素集合：



 

 

四万个元素集合时：



 

 

压缩代码样例

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，日常使用中都可以注意以下几点：

以上是我根据现有实际场景总结出的一些解决手段，记录了这些大key的优化经验，希望可以在日常场景中帮助到大家。大家有其他的好的经验，也可以分享出来。