关于Redis大KEY解决方案

（一）相关界说与配景

1.大key界说：

所谓的大key题目是某个key的value比较大，所以本质上是大value题目
   

• String类型的Key，它的值为5MB（数据过大）
  
• List类型的Key，它的列表数量为20000个（列表数量过多）
  
• ZSet类型的Key，它的成员数量为10000个（成员数量过多）
  
• Hash格式的Key，它的成员数量固然只有1000个但这些成员的value总巨细为100MB（成员体积过大）
  

  在现实业务中，大Key的判断仍旧需要根据Redis的现实使用场景、业务场景来举行综合判断。通常都会以数据巨细与成员数量来判断。
一样寻常认为string类型控制在10KB以内，hash、list、set、zset元素个数不要凌驾10000个。
2.大key产生：

大key的产生每每是业务方计划不公道，没有预见vaule的动态增长题目
   

• redis数据结构使用不公道，易造成Key的value过大，如使用String类型的Key存放大体积二进制文件型数据
  
• 业务上线前规划计划不敷，没有对Key中的成员举行公道的拆分，造成个别Key中的成员数量过多
  
• 没有对无效数据举行定期整理，造成如HASH类型Key中的成员持续不断的增加。即不停往value塞数据，却没有删除机制，value只会越来越大
  
• 使用LIST类型Key的业务消耗侧发生代码故障，造成对应Key的成员只增不减
  

   在现实业务中，大概会发生的大key场景：
社交类：粉丝列表，如果某些明星或者大v不精心计划下，必是bigkey；
统计类：例如按天存储某项功能或者网站的用户聚集，除非没几个人用，否则必是bigkey；
缓存类：将数据从数据库load出来序列化放到Redis里，这个方式非常常用，但有两个地方需要注意，第一，是不是有必要把所有字段都缓存，第二，有没有相关关联的数据。
3.现实业务配景

现有场景后管系统导入商品相关业务：
对导入商品的excel数据通过EasyExcel工具解析，然后对解析出来的商品信息同一组装到一个大对象，而且放入到redis当中，同时未设置缓存失效时间。
导入的excel表格数据直接关系到商品对象的巨细，这就存在excel表格数据非常多的场景下，封装的商品对象就越大，导致每次只要上传大量数据，redis中就会存在大key，对redis并发本领形成埋伏影响，由于redis是单线程运行的，如果一次操作的value很大会对整个redis的相应时间造成负面影响，导致IO网络拥塞。
运维反馈：
XXX_IMPORT_XXX相关key已经凌驾10kb+

List类型的key将近200kb巨细

复现环节：
PURCHASE_IMPORT_IDXXX：
测试情况使用Another Redis Desktop Manager工具复现 发现excel导入1条数据152B，4条数据将近500B（如下图），由此可以推断出excel如果导入100条乐成数据，其value巨细根本约等于10KB，深挖其代码逻辑发现只做了对excel一次性导入不能大于200条，且excel文件巨细不能大于5MB前置校验，故而完全大概导致存在10KB~20KB大key

STOCKCHECK_IMPORT_IDXXX:
redisTemplate.opsForList().rightPush(KEY + UUID, Object);接纳默认redisTemplate模板类把list数组聚集添加进Redis，不断的往后添加list元素且没有失效时间

此中一个list元素就占据将近1.5kb，且如上个大key一样只做了一次性导入不能大于200条，且excel文件巨细不能大于5MB前置校验，所以完全存在200KB的超大key产生，接纳默认的序列化方式，将等于null的属性也序列化出来就是需要改进的地方

4.造成的影响

总结用几个词概况来说：阻塞请求，网络拥塞，内存增大而且分配不均，过期删除，主从同步切换影响，从而进一步终极会导致我们的线程阻塞，并发量下降，客户端超时，服务端业务乐成率下降等。
   

• 客户端执行命令的时长变慢，Big Key对应的value较大，我们对其举行读写的时间，需要耗费较长的时间，如许就大概阻塞后续的请求处理
  
• Redis内存到达maxmemory参数界说的上限引发操作阻塞或紧张的Key被逐出，甚至引发内存溢出（Out Of Memory）
  
• 集群架构下，某个数据分片的内存使用率远超其他数据分片，无法使数据分片的内存资源到达平衡
  
• 读取单value较大时会占用服务器网卡较多带宽，自身变慢的同时大概会影响该服务器上的其他Redis实例或者应用
  
• 对大Key执行删除操作，易造成主库较长时间的阻塞，进而大概引发同步中断或主从切换
  

  （二）优化方案

根据大key的现实用途可以分为两种情况：可删除和不可删除。

1.删除大key

起首思量删除大key，如果发现某些大key并非热key就可以在DB中查询使用，则可以在Redis中删掉：
   

• Redis 4.0及之后版本：您可以通过UNLINK命令安全地删除大Key甚至特大Key，该命令能够以非阻塞的方式，徐徐地整理传入的Key。 Redis UNLINK 命令雷同与 DEL 命令，表示删除指定的 key，如果指定 key 不存在，命令则忽略。 UNLINK 命令不同与 DEL
  命令在于它是异步执行的，因此它不会阻塞。 UNLINK 命令是非阻塞删除，非阻塞删除简言之，就是将删除操作放到另外一个线程行止理。
  
• Redis 4.0之前的版本：建议先通过SCAN命令读取部分数据，然后举行删除，避免一次性删除大量key导致Redis阻塞。 Redis Scan 命令用于迭代数据库中的数据库键。 SCAN 命令是一个基于游标的迭代器，每次被调用之后， 都会向用户返回一个新的游标，
  用户在下次迭代时需要使用这个新游标作为 SCAN 命令的游标参数， 以此来延续之前的迭代过程。
  

  

根据运维反馈及现实业务场景，重要xxxIMPORT大key分为两种场景：
（1）字符串类型：PURCHASE_IMPORT_IDXXX
（2）List类型：STOCKCHECK_IMPORT_IDXXX 、STOCKADJUST_IMPORT_IDXXX等等
序号接口名称所在业务场景缓存key缓存时间缓存内容获取缓存后处理1…新建采购单-导入商品PURCHASE_IMPORT_XXX永久见效字符串类型：导入excel商品数据（包括本身自带excel商品，动态数据：查询数据库商品信息，库存、金额等）导入后接口返回展示2…新建入库单-导入商品STOCKCHECK_IMPORT_XXX永久见效list类型：导入excel商品数据（包括本身自带excel商品，动态数据：查询数据库商品信息，库存、金额、部分、堆栈等）导入后接口返回展示3……………… 优化方案：
根据调研的现实场景，之前的大key业务用途范围比较窄，只有导入商品的时间页面回显商品信息，可以思量上线的时间，晚点直接删除原来所有XXX_IMPORT_IDXXX 的大key，后续继续产生的大key代码，采取下面两种方案优化（压缩和拆分大key）
2.压缩大key

现实业务List类型：STOCKCHECK_IMPORT_XXX等大key接纳默认的序列化方式，可以接纳压缩法
思量到使用符合的序列化框架、压缩算法：
   

• 当vaule是string时，比较难拆分，则使用序列化、压缩算法将key的巨细控制在公道范围内，但是序列化和反序列化都会带来更多时间上的消耗
  
• 当value是string，压缩之后仍旧是大key，则需要思量举行拆分
  

  3.拆分大key

现实业务字符串类型：PURCHASE_IMPORT_XXX大key内部实现逻辑使用ProtoStuff序列化框架实现，思量到此框架序列化服从非常高，压缩后的格式是二进制文件，内存巨细已经相当于最优选择，所以思量采取拆分本领
   

• 将一个Big Key拆分为多个key-value如许的小Key，并确保每个key的成员数量或者巨细在公道范围内，然后再举行存储，通过get不同的key或者使用mget批量获取。
  
• 当value是list/set等聚集类型时，根据预估的数据规模来举行分片，不同的元素计算后分到不同的片。
  

  4.当地缓存（Caffeine ）

可以思量当地缓存（Caffeine ）》Redis》数据库 ，本期文档不实现此方案
（三）实现细节

根据现实业务，重要xxxIMPORT大key的两种场景，技能方案选型落地：
   

• 字符串类型：PURCHASE_IMPORT_XXX接纳拆分key方案。
  
• list类型：STOCKCHECK_IMPORT_XXX等接纳压缩key方案，观察其压缩后巨细，如果压缩之后仍旧是大key，思量进一步拆分。
  

  1.PURCHASE_IMPORT_XXX大key处理伪代码（拆分）

1. 监听类：
2. public class ProductListener extends AnalysisEventListener<ExportExcelVo> {
3.     //省略上面重复代码  
4.     //拆分key大小
5.     @Value("${product.splitSize}")
6.     private String splitSize;
8.     //导入商品信息
9.     private List<List<Product>> productList = new ArrayList<>();
11.     /**读取整个Excel完毕后调用*/
12.     @Override
13.     public void doAfterAllAnalysed(AnalysisContext analysisContext) {
14.         //必要校验并组装商品信息数据ProductVO
15.         //重点！
16.         handleMultipleValue(ProductVO);
17.     }
18.      
19.     /**拆分多k-v*/
20.     public void handleMultipleValue(ProductVO vo){
21.         Map<String, ProductRO> map= new HashMap<>();
22.         //将必要返回的商品总数量、金额等信息统一封装成keys0
23.         map.put("keys0" + KEY + UUID,ProductRO.builder().adjustNum(po.getAdjustNum())
24.                  .priceSum(po.getPriceSum()).build());
25.         //将商品对象内部的list根据实际业务拆分每个key最多包含splitSize条数据
26.         productList = ListUtils.partition(po.getProductList(), splitSize);
27.         for(int i = 1; i <= productList.size(); i++){
28.             ProductRO stock = ProductRO.builder().productList(productList.get(i-1)).build();
29.             map.put("keys" + i + KEY + UUID, stock);
30.         }
31.         //pipeline管道批量添加到redis中并设置10分钟失效
32.         redisHelper.putMultiCacheWithExpireTime(valueMap, 60 * 10L);
33.         //注意此处使用RedisTemplate.opsForValue().multiSet(map)不支持失效时间的设置...又是不断增加key
34.         //虽然可以通过redisTemplate.expire(key, timeout, TimeUnit.SECONDS);循环的去设置map中key失效时间，但不是原子性操作没有意义
35.     }
36. }
37.      
38. Controller实现：
39. @RestController
40. public class ProductFileExcelController {
41.     @PostMapping(value = "api/product/uploadFile")
42.     public Result uploadFile(@RequestParam("file") MultipartFile file) {
43.        ProductListener listener = new ProductListener ();
44.        EasyExcel.read(file.getInputStream(), ExportExcelVo.class, listener).sheet().doRead();
45.        //之前逻辑这里直接取缓存返回
46.        //return Result.success(redisHelper.getCache(key));
47.        //重点！修改后：
48.        List<Product> productList = new ArrayList<>();
49.        ProductRO stock = redisHelper.getCache("keys0" + KEY + UUID, ProductRO.class);
50.        for(int i = 1; i <= productPurchaseDTOListener.getProductList().size(); i++){
51.           ProductRO stockList = redisHelper.getCache("keys" + i + KEY + UUID, ProductRO.class);
52.           productList.addAll(stockList.getProductList());
53.        }
54.        stock.setProductList(productList);
55.        return Result.success(stock);
56.     }
57. }
59. Redis工具类实现：
60. @Component
61. public class RedisHelper {
62.     @Autowired
63.     private RedisTemplate<String, String> redisTemplate;
64.    
65.    //获取key
66.    public <T> T getCache(final String key, Class<T> targetClass) {
67.      byte[] result = (byte[])this.redisTemplate.execute(new RedisCallback<byte[]>() {
68.         @Override
69.         public byte[] doInRedis(RedisConnection connection) throws DataAccessException {
70.             return connection.get(key.getBytes());
71.         }
72.     });
73.      return result != null && result.length != 0 ? ProtoStuffSerializerUtil.deserialize(result, targetClass) : null;
74.    }
76.    //使用管道批量添加
77.    public <T> void putMultiCacheWithExpireTime(Map<String, T> map, final long expireTime) {
78.      this.redisTemplate.executePipelined(new RedisCallback<String>() {
79.         @Override
80.         public String doInRedis(RedisConnection connection) throws DataAccessException {
81.             map.forEach((key, value) -> {
82.                 connection.setEx(key.getBytes(), expireTime,
83.                         ProtoStuffSerializerUtil.serialize(value));
84.             });
85.             return null;
86.         }
87.      });
88.    }
89. }
91. //拆分效果如下图，key0XXX是商品对象公共信息，总金额，数量等
92. //key1-key5是将10条excel数据拆分成5个key-value 每个key包含2条数据 平均200-300b

复制代码

上述方法是将大key拆分成多个key-value，如许分拆的意义在于分拆单次操作的压力，将操作压力平摊到多个redis实例中，降低对单个redis的IO影响；
实用于当前的业务场景（该对象需要每次都整存整取），除此之外另有别的拆分方法，好比该对象每次只需要存取部分数据，也可以拆分多个key-value
方案二：将这个存储在一个hash中，每个field代表一个具体的属性，使用hget,hmget来获取部分的value，使用hset，hmset来更新部分属性，此处不做这方面的伪代码实现，有兴趣的小伙伴可以自行研究~~
2.STOCKCHECK_IMPORT_XXX大key处理伪代码：（压缩）

方案一

RedisSerializer使用JdkSerializationRedisSerializer（默认值），需要被序列化Class实现Serializable接口，且重要是不方便人工排查数据，有许多属性为null的值，这些属性的序列化和反序列化完全没故意义，只需要序列化属性值不为空的对象就行，如许不仅能降低机器序列化与反序列化时产生的CPU，减少时间，还能减少Redis内存使用，减少网络开销。所以需要根据现实大key的业务场景可以自界说RedisTemplate序列化方式。
网上有许多这方面的博客，写的很具体可以参考，这里只贴出轮子部分核心代码和效果（接纳Jackson2JsonRedisSerializer，也可以使用FastJsonRedisSerializer同样的效果）

1. @Configuration
2. public class RedisConfig {
3.     /**
4.     * 修改redisTemplate的序列化方式
5.     * @param factory
6.     * @return
7.     */
8.     @Bean(name = "redisTemplate")
9.     public RedisTemplate<K, V> redisTemplate(LettuceConnectionFactory factory) {
10.         //创建RedisTemplate对象
11.         RedisTemplate<K, V> template = new RedisTemplate<K, V>();
12.         template.setConnectionFactory(factory);
13.         //设置key的序列化方式
14.         template.setKeySerializer(keySerializer());
15.         template.setHashKeySerializer(keySerializer());
16.         //设置RedisTemplate的Value序列化方式Jackson2JsonRedisSerializer；默认是JdkSerializationRedisSerializer
17.         template.setValueSerializer(valueSerializer());
18.         template.setHashValueSerializer(valueSerializer());
20.         template.afterPropertiesSet();
21.         return template;
22. }
24.     private RedisSerializer<String> keySerializer() {
25.         return new StringRedisSerializer();
26.     }
28.     private RedisSerializer<Object> valueSerializer() {
29.         Jackson2JsonRedisSerializer<Object> jackson2JsonRedisSerializer = new Jackson2JsonRedisSerializer<Object>(Object.class);
31.         ObjectMapper om = new ObjectMapper();
32.         // 指定要序列化的域，field,get和set,以及修饰符范围，ANY是都有包括private和public
33.         om.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY);
34.         // 指定序列化输入的类型，类必须是非final修饰的，final修饰的类，比如String,Integer等会抛出异常
35.         om.activateDefaultTyping(LaissezFaireSubTypeValidator.instance, ObjectMapper.DefaultTyping.NON_FINAL);
36.         // 解决时间序列化问题
37.         om.disable(SerializationFeature.WRITE_DATES_AS_TIMESTAMPS);
38.         // 对象属性值为null时,不进行序列化存储
39.         om.setSerializationInclusion(JsonInclude.Include.NON_NULL);
40.         om.registerModule(new JavaTimeModule());
42.         jackson2JsonRedisSerializer.setObjectMapper(om);
43.         return jackson2JsonRedisSerializer;
44.      }
45. }
46. //实现后可以看到采用自定义的序列化方式，不仅可以只序列化不为null的属性，且压缩后一条数据大小由原来1.5KB->500B,内存大小接近原来的1/3
47. //但是由于excel最大可以导入200条数据，按照每条500B换算下来也接近100KB的大key，完全不符合我们的预期，所以考虑的压缩后接着拆分多个k-v
48. //或者采用更高效的序列化方式

复制代码

方案二（继续优化）

采取ProtoStuff框架来进一步压缩list类型的key

1. 监听类：
2. public class ProductListener extends AnalysisEventListener<ExportExcelVo> {
4.     //value库存信息
5.     private List<ProductStockPO> stockPOList = new ArrayList<>();
7.     /**读取整个Excel完毕后调用*/
8.     @Override
9.     public void doAfterAllAnalysed(AnalysisContext analysisContext) {
10.         //必要校验并封装库存信息ProductStockPO
11.         //redisTemplate.opsForList().rightPush(KEY+UUID, ProductStockPO);
12.         //重点！修改后
13.         //添加到redis中并设置10分钟失效
14.         redisHelper.putListCacheWithExpireTime(KEY+UUID, ProductStockPO,60 * 10L);
15.     }
16. }
17.      
18. Controller实现：
19. @RestController
20. public class ProductFileExcelController {
21.     @PostMapping(value = "api/stock/uploadFile")
22.     public Result uploadFile(@RequestParam("file") MultipartFile file) {
23.        ProductListener listener = new ProductListener ();
24.        EasyExcel.read(file.getInputStream(), ExportExcelVo.class, listener).sheet().doRead();
25.        //之前逻辑这里直接取缓存返回
26.        //return Result.success(redisTemplate.opsForList().range(KEY+UUID, 0, -1););
27.        //重点！修改后：
28.        return Result.success(redisHelper.getListCache(KEY+UUID));
29.     }
30. }
32. Redis工具类实现：
33. @Component
34. public class RedisHelper {
35.     @Autowired
36.     private RedisTemplate<String, String> redisTemplate;
37.    
38.    //获取list
39.    public <T> List<T> getListCache(final String key, Class<T> targetClass) {
40.     byte[] result = (byte[])this.redisTemplate.execute(new RedisCallback<byte[]>() {
41.         @Override
42.         public byte[] doInRedis(RedisConnection connection) throws DataAccessException {
43.             return connection.get(key.getBytes());
44.         }
45.     });
46.     return result != null && result.length != 0 ? ProtoStuffSerializerUtil.deserializeList(result, targetClass) : null;
47.    }
49.    //添加list
50.    public <T> boolean putListCacheWithExpireTime(String key, List<T> objList, final long expireTime) {
51.     final byte[] bkey = key.getBytes();
52.     final byte[] bvalue = ProtoStuffSerializerUtil.serializeList(objList);
53.     boolean result = (Boolean)this.redisTemplate.execute(new RedisCallback<Boolean>() {
54.         @Override
55.         public Boolean doInRedis(RedisConnection connection) throws DataAccessException {
56.             connection.setEx(bkey, expireTime, bvalue);
57.             return true;
58.         }
59.     });
60.     return result;
61.    }
62. }
64. //发现采用Protostuff序列化框架的确可以更加高效的压缩我们的list数据，压缩1条等比150B，比原始的1.5KB缩小了将近10倍
65. //但是我们会发现，即使采取了如此高效的压缩手段还是达不到我们的预期，导入200条数据后大小也是相当于20KB，依然是个大key，这不得让我们进一步的考虑拆分...

复制代码

方案三（压缩+拆分）

1. 监听类：
2. public class ProductListener extends AnalysisEventListener<ExportExcelVo> {
3.     //存放拆分后数据
4.     private List<List<ProductStockPO>> list = new ArrayList<>();
5.     //重复代码逻辑
6.     @Override
7.     public void doAfterAllAnalysed(AnalysisContext analysisContext) {
8.         //必要校验
9.         list = ListUtils.partition(stockPOList, 2);
10.         Map<String, List<ProductStockPO>> valueMap = new HashMap<>();
11.         for(int i = 1; i <= list.size(); i++){
12.            valueMap.put("keys" + i + KEY + UUID, list.get(i-1));
13.         }
14.        //pipeline管道批量添加到redis中并设置10分钟失效
15.         redisHelperConfig.putMultiListCacheWithExpireTime(valueMap,60 * 10L);
16.     }
17. }
18.      
19. Controller实现：
20. @RestController
21. public class ProductFileExcelController {
22.     @PostMapping(value = "api/stock/uploadFile")
23.     public Result uploadFile(@RequestParam("file") MultipartFile file) {
24.        ProductListener listener = new ProductListener ();
25.        EasyExcel.read(file.getInputStream(), ExportExcelVo.class, listener).sheet().doRead();
26.        List<ProductStockPO> stockList = new ArrayList<>();
27.        for(int i = 1; i <= stockCheckDTOListener.getList().size(); i++){
28.          List<ProductStockPO> list = redisHelperConfig.getListCache("keys" + i + KEY + UUID, ProductStockPO.class);
29.          stockList.addAll(list);
30.       }
31.        return Result.success(stockList);
32.     }
33. }
35. Redis工具类实现：
36. @Component
37. public class RedisHelper {
38.     @Autowired
39.     private RedisTemplate<String, String> redisTemplate;
40.    
41.    //使用管道批量添加list
42.     public <T> void putMultiListCacheWithExpireTime(Map<String, List<T>> map, final long expireTime) {
43.       this.redisTemplate.executePipelined(new RedisCallback<String>() {
44.         @Override
45.         public String doInRedis(RedisConnection connection) throws DataAccessException {
46.             map.forEach((key, value) -> {
47.                 connection.setEx(key.getBytes(), expireTime,
48.                         ProtoStuffSerializerUtil.serializeList(value));
49.             });
50.             return null;
51.         }
52.       });
53.     }
54. }
56. //发现采取压缩后进一步拆分5条数据拆分成3个key，其中2条数据的大小为300B左右，按照最大导入200条数据来算，可以拆分成10key，
57. //每个key包含20条数据将近3KB大小，符合预期,具体可以拆分多少按照自己实际业务场景来估算

复制代码

（四）Redis开发规范

1.键值计划

key名计划

(1)【建议】: 可读性和可管理性
以业务名(或数据库名)为前缀(防止key辩论)，用冒号分隔，好比业务名:表名:id
product:stockstock:1
(2)【建议】：简洁性
包管语义的前提下，控制key的长度，当key较多时，内存占用也不容忽视，例如：
user:{uid}:friends:messages:{mid}简化为u:{uid}