本文作者以实际项目遇到的大 key 问题为线索，场景化地讲述对应的解决方案。通过本文，您可以了解关于大 key 基础概念、影响以及遇到大 key 的具体解决手段，帮助您更好把控缓存的使用场景，从而提升软件系统的稳定性。
背景：
最近系统内缓存 CPU 使用率一直报警，超过设置的 70%报警阀值，针对此场景，需要对应解决缓存是否有大 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、压缩存储数据
压缩方法结果：
单个元素时：
四百个元素集合：
四万个元素集合时：
压缩代码样例
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 的优化经验，希望可以在日常场景中帮助到大家。大家有其他的好的经验，也可以分享出来。
作者：全渠道生态 范晓
来源：京东零售技术 转载请注明来源
评论