关于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+
https://img-blog.csdnimg.cn/direct/15460345734949a982c6b011497587fb.png#pic_center
List类型的key将近200kb巨细https://img-blog.csdnimg.cn/direct/9dd953d85e554fe6b93eea33d4746583.png#pic_center
复现环节：
PURCHASE_IMPORT_IDXXX：
测试情况使用Another Redis Desktop Manager工具复现 发现excel导入1条数据152B，4条数据将近500B（如下图），由此可以推断出excel如果导入100条乐成数据，其value巨细根本约等于10KB，深挖其代码逻辑发现只做了对excel一次性导入不能大于200条，且excel文件巨细不能大于5MB前置校验，故而完全大概导致存在10KB~20KB大keyhttps://img-blog.csdnimg.cn/direct/1354b7cabc5c4a52b2688e72a966e317.png#pic_center
STOCKCHECK_IMPORT_IDXXX:
redisTemplate.opsForList().rightPush(KEY + UUID, Object);接纳默认redisTemplate模板类把list数组聚集添加进Redis，不断的往后添加list元素且没有失效时间
https://img-blog.csdnimg.cn/direct/c636d1c4e57142d991823b94302aa68f.png#pic_center
此中一个list元素就占据将近1.5kb，且如上个大key一样只做了一次性导入不能大于200条，且excel文件巨细不能大于5MB前置校验，所以完全存在200KB的超大key产生，接纳默认的序列化方式，将等于null的属性也序列化出来就是需要改进的地方
https://img-blog.csdnimg.cn/direct/f21843d3cbdb40cc8e417346d51ae5f5.png#pic_center
4.造成的影响
总结用几个词概况来说：阻塞请求，网络拥塞，内存增大而且分配不均，过期删除，主从同步切换影响，从而进一步终极会导致我们的线程阻塞，并发量下降，客户端超时，服务端业务乐成率下降等。
[*]客户端执行命令的时长变慢，Big Key对应的value较大，我们对其举行读写的时间，需要耗费较长的时间，如许就大概阻塞后续的请求处理
[*]Redis内存到达maxmemory参数界说的上限引发操作阻塞或紧张的Key被逐出，甚至引发内存溢出（Out Of Memory）
[*]集群架构下，某个数据分片的内存使用率远超其他数据分片，无法使数据分片的内存资源到达平衡
[*]读取单value较大时会占用服务器网卡较多带宽，自身变慢的同时大概会影响该服务器上的其他Redis实例或者应用
[*]对大Key执行删除操作，易造成主库较长时间的阻塞，进而大概引发同步中断或主从切换
（二）优化方案
根据大key的现实用途可以分为两种情况：可删除和不可删除。
https://img-blog.csdnimg.cn/direct/64d6a3fcf2c543b7a2c1ac6af6751924.png#pic_center
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 命令的游标参数， 以此来延续之前的迭代过程。
https://img-blog.csdnimg.cn/direct/a8d6a7798dfe439d88baebf9c8c9987d.png#pic_center
根据运维反馈及现实业务场景，重要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处理伪代码（拆分）
监听类：
public class ProductListener extends AnalysisEventListener<ExportExcelVo> {
//省略上面重复代码
//拆分key大小
@Value("${product.splitSize}")
private String splitSize;
//导入商品信息
private List<List<Product>> productList = new ArrayList<>();
/**读取整个Excel完毕后调用*/
@Override
public void doAfterAllAnalysed(AnalysisContext analysisContext) {
//必要校验并组装商品信息数据ProductVO
//重点！
handleMultipleValue(ProductVO);
}
/**拆分多k-v*/
public void handleMultipleValue(ProductVO vo){
Map<String, ProductRO> map= new HashMap<>();
//将必要返回的商品总数量、金额等信息统一封装成keys0
map.put("keys0" + KEY + UUID,ProductRO.builder().adjustNum(po.getAdjustNum())
.priceSum(po.getPriceSum()).build());
//将商品对象内部的list根据实际业务拆分每个key最多包含splitSize条数据
productList = ListUtils.partition(po.getProductList(), splitSize);
for(int i = 1; i <= productList.size(); i++){
ProductRO stock = ProductRO.builder().productList(productList.get(i-1)).build();
map.put("keys" + i + KEY + UUID, stock);
}
//pipeline管道批量添加到redis中并设置10分钟失效
redisHelper.putMultiCacheWithExpireTime(valueMap, 60 * 10L);
//注意此处使用RedisTemplate.opsForValue().multiSet(map)不支持失效时间的设置...又是不断增加key
//虽然可以通过redisTemplate.expire(key, timeout, TimeUnit.SECONDS);循环的去设置map中key失效时间，但不是原子性操作没有意义
}
}
Controller实现：
@RestController
public class ProductFileExcelController {
@PostMapping(value = "api/product/uploadFile")
public Result uploadFile(@RequestParam("file") MultipartFile file) {
ProductListener listener = new ProductListener ();
EasyExcel.read(file.getInputStream(), ExportExcelVo.class, listener).sheet().doRead();
//之前逻辑这里直接取缓存返回
//return Result.success(redisHelper.getCache(key));
//重点！修改后：
List<Product> productList = new ArrayList<>();
ProductRO stock = redisHelper.getCache("keys0" + KEY + UUID, ProductRO.class);
for(int i = 1; i <= productPurchaseDTOListener.getProductList().size(); i++){
ProductRO stockList = redisHelper.getCache("keys" + i + KEY + UUID, ProductRO.class);
productList.addAll(stockList.getProductList());
}
stock.setProductList(productList);
return Result.success(stock);
}
}
Redis工具类实现：
@Component
public class RedisHelper {
@Autowired
private RedisTemplate<String, String> redisTemplate;
//获取key
public <T> T getCache(final String key, Class<T> targetClass) {
byte[] result = (byte[])this.redisTemplate.execute(new RedisCallback<byte[]>() {
@Override
public byte[] doInRedis(RedisConnection connection) throws DataAccessException {
return connection.get(key.getBytes());
}
});
return result != null && result.length != 0 ? ProtoStuffSerializerUtil.deserialize(result, targetClass) : null;
}
//使用管道批量添加
public <T> void putMultiCacheWithExpireTime(Map<String, T> map, final long expireTime) {
this.redisTemplate.executePipelined(new RedisCallback<String>() {
@Override
public String doInRedis(RedisConnection connection) throws DataAccessException {
map.forEach((key, value) -> {
connection.setEx(key.getBytes(), expireTime,
ProtoStuffSerializerUtil.serialize(value));
});
return null;
}
});
}
}
//拆分效果如下图，key0XXX是商品对象公共信息，总金额，数量等
//key1-key5是将10条excel数据拆分成5个key-value 每个key包含2条数据 平均200-300b
https://img-blog.csdnimg.cn/direct/3a33242064654a8dba44ba90ab524464.png#pic_center
上述方法是将大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同样的效果）
@Configuration
public class RedisConfig {
/**
* 修改redisTemplate的序列化方式
* @param factory
* @return
*/
@Bean(name = "redisTemplate")
public RedisTemplate<K, V> redisTemplate(LettuceConnectionFactory factory) {
//创建RedisTemplate对象
RedisTemplate<K, V> template = new RedisTemplate<K, V>();
template.setConnectionFactory(factory);
//设置key的序列化方式
template.setKeySerializer(keySerializer());
template.setHashKeySerializer(keySerializer());
//设置RedisTemplate的Value序列化方式Jackson2JsonRedisSerializer；默认是JdkSerializationRedisSerializer
template.setValueSerializer(valueSerializer());
template.setHashValueSerializer(valueSerializer());
template.afterPropertiesSet();
return template;
}
private RedisSerializer<String> keySerializer() {
return new StringRedisSerializer();
}
private RedisSerializer<Object> valueSerializer() {
Jackson2JsonRedisSerializer<Object> jackson2JsonRedisSerializer = new Jackson2JsonRedisSerializer<Object>(Object.class);
ObjectMapper om = new ObjectMapper();
// 指定要序列化的域，field,get和set,以及修饰符范围，ANY是都有包括private和public
om.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY);
// 指定序列化输入的类型，类必须是非final修饰的，final修饰的类，比如String,Integer等会抛出异常
om.activateDefaultTyping(LaissezFaireSubTypeValidator.instance, ObjectMapper.DefaultTyping.NON_FINAL);
// 解决时间序列化问题
om.disable(SerializationFeature.WRITE_DATES_AS_TIMESTAMPS);
// 对象属性值为null时,不进行序列化存储
om.setSerializationInclusion(JsonInclude.Include.NON_NULL);
om.registerModule(new JavaTimeModule());
jackson2JsonRedisSerializer.setObjectMapper(om);
return jackson2JsonRedisSerializer;
}
}
//实现后可以看到采用自定义的序列化方式，不仅可以只序列化不为null的属性，且压缩后一条数据大小由原来1.5KB->500B,内存大小接近原来的1/3
//但是由于excel最大可以导入200条数据，按照每条500B换算下来也接近100KB的大key，完全不符合我们的预期，所以考虑的压缩后接着拆分多个k-v
//或者采用更高效的序列化方式
https://img-blog.csdnimg.cn/direct/3223e401b763425fac07a8aab581f7cd.png#pic_center
方案二（继续优化）
采取ProtoStuff框架来进一步压缩list类型的key
监听类：
public class ProductListener extends AnalysisEventListener<ExportExcelVo> {
//value库存信息
private List<ProductStockPO> stockPOList = new ArrayList<>();
/**读取整个Excel完毕后调用*/
@Override
public void doAfterAllAnalysed(AnalysisContext analysisContext) {
//必要校验并封装库存信息ProductStockPO
//redisTemplate.opsForList().rightPush(KEY+UUID, ProductStockPO);
//重点！修改后
//添加到redis中并设置10分钟失效
redisHelper.putListCacheWithExpireTime(KEY+UUID, ProductStockPO,60 * 10L);
}
}
Controller实现：
@RestController
public class ProductFileExcelController {
@PostMapping(value = "api/stock/uploadFile")
public Result uploadFile(@RequestParam("file") MultipartFile file) {
ProductListener listener = new ProductListener ();
EasyExcel.read(file.getInputStream(), ExportExcelVo.class, listener).sheet().doRead();
//之前逻辑这里直接取缓存返回
//return Result.success(redisTemplate.opsForList().range(KEY+UUID, 0, -1););
//重点！修改后：
return Result.success(redisHelper.getListCache(KEY+UUID));
}
}
Redis工具类实现：
@Component
public class RedisHelper {
@Autowired
private RedisTemplate<String, String> redisTemplate;
//获取list
public <T> List<T> getListCache(final String key, Class<T> targetClass) {
byte[] result = (byte[])this.redisTemplate.execute(new RedisCallback<byte[]>() {
@Override
public byte[] doInRedis(RedisConnection connection) throws DataAccessException {
return connection.get(key.getBytes());
}
});
return result != null && result.length != 0 ? ProtoStuffSerializerUtil.deserializeList(result, targetClass) : null;
}
//添加list
public <T> boolean putListCacheWithExpireTime(String key, List<T> objList, final long expireTime) {
final byte[] bkey = key.getBytes();
final byte[] bvalue = ProtoStuffSerializerUtil.serializeList(objList);
boolean result = (Boolean)this.redisTemplate.execute(new RedisCallback<Boolean>() {
@Override
public Boolean doInRedis(RedisConnection connection) throws DataAccessException {
connection.setEx(bkey, expireTime, bvalue);
return true;
}
});
return result;
}
}
//发现采用Protostuff序列化框架的确可以更加高效的压缩我们的list数据，压缩1条等比150B，比原始的1.5KB缩小了将近10倍
//但是我们会发现，即使采取了如此高效的压缩手段还是达不到我们的预期，导入200条数据后大小也是相当于20KB，依然是个大key，这不得让我们进一步的考虑拆分...
https://img-blog.csdnimg.cn/direct/13b64ba067bc41f88144c20312a0b13b.png#pic_center
方案三（压缩+拆分）
监听类：
public class ProductListener extends AnalysisEventListener<ExportExcelVo> {
//存放拆分后数据
private List<List<ProductStockPO>> list = new ArrayList<>();
//重复代码逻辑
@Override
public void doAfterAllAnalysed(AnalysisContext analysisContext) {
//必要校验
list = ListUtils.partition(stockPOList, 2);
Map<String, List<ProductStockPO>> valueMap = new HashMap<>();
for(int i = 1; i <= list.size(); i++){
valueMap.put("keys" + i + KEY + UUID, list.get(i-1));
}
//pipeline管道批量添加到redis中并设置10分钟失效
redisHelperConfig.putMultiListCacheWithExpireTime(valueMap,60 * 10L);
}
}
Controller实现：
@RestController
public class ProductFileExcelController {
@PostMapping(value = "api/stock/uploadFile")
public Result uploadFile(@RequestParam("file") MultipartFile file) {
ProductListener listener = new ProductListener ();
EasyExcel.read(file.getInputStream(), ExportExcelVo.class, listener).sheet().doRead();
List<ProductStockPO> stockList = new ArrayList<>();
for(int i = 1; i <= stockCheckDTOListener.getList().size(); i++){
List<ProductStockPO> list = redisHelperConfig.getListCache("keys" + i + KEY + UUID, ProductStockPO.class);
stockList.addAll(list);
}
return Result.success(stockList);
}
}
Redis工具类实现：
@Component
public class RedisHelper {
@Autowired
private RedisTemplate<String, String> redisTemplate;
//使用管道批量添加list
public <T> void putMultiListCacheWithExpireTime(Map<String, List<T>> map, final long expireTime) {
this.redisTemplate.executePipelined(new RedisCallback<String>() {
@Override
public String doInRedis(RedisConnection connection) throws DataAccessException {
map.forEach((key, value) -> {
connection.setEx(key.getBytes(), expireTime,
ProtoStuffSerializerUtil.serializeList(value));
});
return null;
}
});
}
}
//发现采取压缩后进一步拆分5条数据拆分成3个key，其中2条数据的大小为300B左右，按照最大导入200条数据来算，可以拆分成10key，
//每个key包含20条数据将近3KB大小，符合预期,具体可以拆分多少按照自己实际业务场景来估算
https://img-blog.csdnimg.cn/direct/a21551226cb043abb385e8e6d0b48170.png#pic_center
（四）Redis开发规范
1.键值计划
key名计划
(1)【建议】: 可读性和可管理性
以业务名(或数据库名)为前缀(防止key辩论)，用冒号分隔，好比业务名:表名:id
product:stockstock:1
(2)【建议】：简洁性
包管语义的前提下，控制key的长度，当key较多时，内存占用也不容忽视，例如：
user:{uid}:friends:messages:{mid}简化为u:{uid}
页:
[1]