投稿
  1. 首页
  2. 随笔

【后端-SpringCache】基于Spring Cache封装一个能够批量 *** 作的Redis缓存记录下踩坑历程（pipeline或mget封装）

bio • 2022-12-17 • 随笔 • 阅读 29

【后端-SpringCache】基于Spring Cache封装一个能够批量 *** 作的Redis缓存记录下踩坑历程（pipeline或mget封装）,第1张

【后端-SpringCache】基于Spring Cache封装一个能够批量 *** 作的Redis缓存记录下踩坑历程（pipeline或mget封装）

Spring Cache是一个非常优秀的缓存组件，我们的应用系统正是使用的Spring Cache。但最近在优化应用系统缓存的过程中意外发现了Spring Cache的很多坑点，特意记录一下。

背景

应用系统中存在部分接口循环调用redis获取缓存的场景（例如：通过多个 userId 来批量获取用户信息），此时我们的代码是类似于这样的（仅示例）：

List users = ids.stream().map(id -> getUserById(id)).collect(Collectors.toList());

@Cacheable(key = "#p0", unless = "#result == null")
public User getUserById(Long id) {
 // ···
}

就像上面说的，这种写法的缺点在于：在循环中 *** 作 Redis 。如果数据都命中缓存还好，但大量频繁的连接也会产生一定的影响降低 qps ，再说一旦缓存没有命中，则会访问数据库，效率也可想而知了。

期望达到的效果

理想的逻辑是优先去 redis 获取缓存，redis 查询结束后筛选出来缓存中不存在的去批量查库，最后再将查库得到的结果存入redis。同时整个过程中应该控制减少请求 redis 的次数。

解决过程 通常做法

面向百度编程后发现有些同学可能会这样做：

@Cacheable(key = "#ids.hash")
public Collection getUsersByIds(Collection ids) {
    // ···
}

这种做法的问题是：

缓存是基于 id 列表的 hashcode ，只有在 id 列表的 hashcode 值相等的情况下，缓存才会命中。而且，一旦列表中的其中一个数据被修改，整个列表缓存都要被清除。

例如：

第一次请求 id 列表是 1,2,3

第二次请求的 id 列表为 1,2,4

在这种情况下，前后两次的缓存不能共享。 如果 id 为 1 的数据发生了改变，那么，这两次请求的缓存都要被清空

看看 Spring 官方是怎么说的

Spring Issue：

  • https://github.com/spring-projects/spring-framework/issues/24139
  • https://github.com/spring-projects/spring-framework/issues/23221

简单翻译一下，具体内容大家可以自行查阅相关 issue

译文：

谢谢你的报告。缓存抽象没有这种状态的概念，如果你返回一个集合，这实际上是您要求存储在缓存中的内容。也没有什么强迫您为给定的缓存保留相同的项类型，所以这种假设并不适合这样的高级抽象。

我的理解是，对于 Spring Cache 这种高级抽象框架来说，Cache 是基于方法的，如果方法返回 Collection，那整个 Collection 就是需要被缓存的内容。

我的解决方案

纠结一段时间后，我决定自己来造个轮子为了达到我们期望的效果，下面先简单介绍一下

设计思路

应用系统中扩展实现自定义的RedisCache后注入到CacheManager中去，自定义的RedisCache在批量获取或删除缓存时采用pipeline方式调用Redis以减少请求次数，需要注意的点是我们的应用系统使用了kryo序列化框架以实现更高性能，所以在执行executePipelined()方法体里序列化字符串时需要先拿到RedisOperations的当前实现类（一般就是RedisTemplate）去获取RedisSerializer对象（其实就是注入RedisTemplate对象时设置的序列化实现）再进行序列化。

核心接口：work.lichong.configuration.cache.CustomizedCache

package work.lichong.configuration.cache;

import java.util.*;
import java.util.function.Function;

public interface CustomizedCache {

    
     List getBatch(List keys, Function, Collection> valueLoader, Function keyMapper, Class vClass, boolean isListValue, String prefix);

    
     void putBatch(Map map, String prefix);

    
    void deleteBatch(Map> keys);

}

具体实现：

package work.lichong.configuration.cache;

import com.google.common.collect.Lists;
import com.google.common.collect.Sets;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.springframework.data.redis.cache.RedisCache;
import org.springframework.data.redis.core.Cursor;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.RedisOperations;
import org.springframework.data.redis.core.ScanOptions;
import org.springframework.data.redis.serializer.RedisSerializer;

import java.util.*;
import java.util.function.Function;
import java.util.regex.Pattern;
import java.util.stream.Collectors;


@Slf4j
public class CustomizedRedisCache extends RedisCache implements CustomizedCache {

    private final RedisOperations redisOperations;
    
    // ...省略RedisCache的实现

    @Override
    @SuppressWarnings({"unchecked", "rawtypes"})
    public  List getBatch(List keys, Function, Collection> valueLoader, Function keyMapper,
                                   Class vClass, boolean isListValue, String prefix) {
        Objects.requireNonNull(redisOperations, "redisOperations required not null");
        List resultList = Collections.emptyList();
        try {
            resultList = redisOperations.executePipelined((RedisCallback) connection -> {
                RedisSerializer keySerializer = redisOperations.getKeySerializer();
                connection.openPipeline();
                for (K k : keys) {
                    byte[] key = keySerializer.serialize(keyPrefix + prefix + k.toString());
                    if (key !=  null) {
                        connection.get(key);
                    } else {
                        log.warn("CustomizedRedisCache 批量 *** 作序列化失败，key={}", k);
                    }
                }
                return null;
            });
        } catch (Exception e) {
            log.error("CustomizedRedisCache 异常", e);
        }

        int keysSize = keys.size();
        // 筛选出缓存中不存在的key
        List dbKeys = new ArrayList<>(keysSize);
        List values = new ArrayList<>();
        if (CollectionUtils.isEmpty(resultList)) {
            dbKeys.addAll(keys);
        } else {
            for (int i = 0; i < resultList.size(); i++) {
                
                Object o = resultList.get(i);

                if (o == null) {
                    dbKeys.add(keys.get(i));
                    continue;
                }
                if (o instanceof NullObject) {
                    continue;
                }
                if (isListValue) {
                    values.addAll((Collection)o);
                    continue;
                }

                values.add((V) o);
            }
        }
        // 缓存中没有就从持久层中查询（需要注意分批次查询）
        if (!CollectionUtils.isEmpty(dbKeys)) {
            Collection dbValue = valueLoader.apply(dbKeys);
            Map dbMap;
            if (isListValue) {
                dbMap = dbValue.stream().filter(Objects::nonNull).collect(Collectors.groupingBy(keyMapper));
            } else {
                dbMap = dbValue.stream().filter(Objects::nonNull).collect(Collectors.toMap(keyMapper, Function.identity()));
            }
            for(K key : dbKeys){
                if(dbMap.containsKey(key)){
                    continue;
                }
                dbMap.put(key, new NullObject());
            }
            putBatch(dbMap, prefix);
            values.addAll(dbValue);
        }
        return values;
    }

    @Override
    @SuppressWarnings({"unchecked", "rawtypes"})
    public  void putBatch(Map map, String prefix) {
        if (map.isEmpty()) {
            return;
        }
        Objects.requireNonNull(redisOperations, "redisTemplate required not null");
        try {
            redisOperations.executePipelined((RedisCallback) connection -> {
                RedisSerializer keySerializer = redisOperations.getKeySerializer();
                RedisSerializer valueSerializer = redisOperations.getValueSerializer();
                connection.openPipeline();
                for (Map.Entry entry : map.entrySet()) {
                    byte[] key = keySerializer.serialize(keyPrefix + prefix + entry.getKey().toString());
                    byte[] value = valueSerializer.serialize(entry.getValue());
                    if (key != null) {
                        connection.set(key, value);
                    } else {
                        log.warn("CustomizedRedisCache 批量 *** 作序列化失败，entry={}", entry);
                    }
                }
                return null;
            });
        } catch (Exception e) {
            log.error("CustomizedRedisCache 异常", e);
        }
    }

    @Override
    @SuppressWarnings({"unchecked", "rawtypes"})
    public void deleteBatch(Map> keys) {
        Objects.requireNonNull(redisOperations, "redisTemplate required not null");
        try {
            RedisSerializer keySerializer = redisOperations.getKeySerializer();
            // 查出目录下所有的key（redis异步，不支持pipeline）
            redisOperations.execute((RedisCallback) connection -> {
                keys.entrySet().forEach(e -> {

                    if(e.getValue().contains("*")){

                        ScanOptions options = ScanOptions.scanOptions().match(getKeyPrefix(e.getKey()) + "*").count(Integer.MAX_VALUE).build();
                        Cursor cursor = connection.scan(options);
                        while (cursor.hasNext()) {
                            connection.del((byte[]) cursor.next());
                        }
                    } else {
                        e.getValue().forEach( k -> {
                            connection.del(keySerializer.serialize(getKeyPrefix(e.getKey()) + k));
                        });
                    }
                });
                return null;
            });
        } catch (Exception e) {
            log.error("CustomizedRedisCache 异常", e);
        }
    }

    
    public static class NullObject {
    }
}
 

调用示例：
 

 
问题记录 

接下来就是戴上痛苦面具的时间了，让我们直面问题：
 

  • 通过我写的putBatch()方法存入Redis的缓存，在触发@CacheEvict注解的删除缓存事件后，这个缓存并不会被删掉！！！
  • 通过调用getBatch()方法获取缓存时可能会获取到null对象！这时会有缓存穿透问题！
 
问题分析 

  1.  
    这个问题是在我意料之外的，一顿排查之后发现Redis下有一个维护key的ZSET集合引起了关注，查阅文档得知注入CacheManager时如果设置了cacheManager.setUsePrefix(false)的话，存入Redis时就会有一个这样的ZSET集合，如下图中userKey~keys：
     
     
    较高版本SpringBoot的CacheManager中usePrefix属性默认为true，而我们的应用系统使用的SpringBoot版本为1.5.4，这个属性默认为false！！！不自觉的就想到了一种常见的绿色植物（吐槽一下这个坑点，Spring在Redis中维护这个keys是很不合理的，如果这个路径下的key特别多那不是每次增删 *** 作时都要修改这个ZSET集合吗？这正是Redis最忌讳的大key阻塞问题啊！所幸后面的版本改掉了），解释一下，其实原因很简单：Spring去批量删除缓存时会先去读取这个集合中的keys再将这些keys删除，而通过我写的putBatch()方法存入Redis的缓存的key并不在这个集合中！
     
    解决方法：
     
  
    • 方案一（当前）：手动设置CacheManager中usePrefix属性默认为true，并修改应用系统中全部缓存的前缀。
    • 方案二：升级SpringBoot版本，已经在日程了，敬请期待！
     
  2.  
    第二个问题其实比较好排查，同时这也是典型的缓存穿透问题：
     
     
    解决方法：
     
  
    • 当前方案：从数据库查出有null对象时，存入一个空对象到Redis暂存，防止后面继续查库。
    • 后续优化：可集成布隆过滤器或布谷鸟过滤器
     
 
反思 

其实在我看来这还不是最完美的解决方案，总结一下：
 
可优化点 

  1.  
    有个明显的问题，要调用这个getBatch()方法还必须传入根据value获取key的映射方法用于查库之后存入redis，但如果要改造之前的代码，有很多业务数据只能根据key查出来value，根据value查key会比较困难。
     
  2.  
    这种写法是编程式的，虽然已经把绝大部分逻辑抽象出来了，但还可以想办法搞成声明式的会更加通用，例如：
     
 

   @CollectionCacheable(cacheNames = "myCache")
   public List findByIds(Collection ids) {
       // ...
   }

 
总结 

我们保证代码健壮的同时也得抗住足够的并发、具备极致的性能，这样的优化点今后可能会越来越多，优化粒度越来越细，hhhhh干就对了！
 

  • 最后：欢迎点赞、关注、收藏！！！
 

 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~往期精选~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 

【前端-开发环境】使用NVM实现不同nodejs版本的自由切换（NVM完整安装使用手册）
 【前端-NPM私服】内网使用verdaccio搭建私有npm服务器
 【前端-IE兼容】Win10和Win11使用Edge调试前端兼容IE6、IE7、IE8、IE9、IE10、IE11问题
 【工具-Shell脚本】java程序产品包模板-linux和windows通用shell启动停止脚本(无需系统安装Java运行环境)
 【工具-Nginx】Nginx高性能通用配置文件-注释版-支持防刷限流、可控高并发、HTTP2、防XSS、Gzip、OCSP Stapling、负载、SSL
 【工具-WireShark】网络HTTP抓包使用教程
 【后端-maven打包】通过profile标签解决同时打jar包 war包需求
 【后端-SkyWalking】SkyWalking前后端开发环境搭建详细教程步骤-6.x/7.x/8.x版本通用-插件二次开发利器（一）
					
										


					

						
欢迎分享，转载请注明来源：内存溢出
原文地址: http://outofmemory.cn/zaji/5696345.html

					

				
								

										
										
										
										
										
									

								

					
						
							
						 赞
						(0)
					
					

						
							 打赏
						
							
								微信扫一扫 微信扫一扫
							
							
								支付宝扫一扫 支付宝扫一扫
							
						
					

				

				

					

						

							
								bio
								bio
								一级用户组
							
						

						

							

								
									
										
									
									0
								
								
									
										
									
									0
								
							

							
							

								
									
										
									
								
							

						

					

				

			
			
            

            	            

            
			

								

					
					

         				
							
								
							 上一篇
						
						2022-12-17
					

				

												

					
					

         				
							下一篇 
           						
							
						
						2022-12-17
					

				

							


			
			

				

					

						发表评论
					

										

						
						

							

								登录后才能评论
							

							
						

					

									

								
 评论列表（0条）

				
    
									

				
											

		

	

	





	


	

		

			

			

				
				

					

						Copyright © 2022 内存溢出 版权所有
						
							湘ICP备2022025235号						
						Powered by
						
							outofmemory.cn
						
					
					

				

			

			

				

					
					
						
						
					
					
						
					
				

			

		

	




























    
    

    

        
            保存