Spring AOP + Redis缓存数据库查询

Bertha 。 2022-06-03 05:17 183阅读 0赞

## 应用场景 ##

我们希望能够将数据库查询结果缓存到Redis中，这样在第二次做同样的查询时便可以直接从redis取结果，从而减少数据库读写次数。

## 需要解决的问题 ##

1.  操作缓存的代码写在哪？必须要做到与业务逻辑代码完全分离。
2.  如何避免脏读？ 从缓存中读出的数据必须与数据库中的数据一致。
3.  如何为一个数据库查询结果生成一个唯一的标识？即通过该标识(Redis中为`Key`)，能唯一确定一个查询结果，同一个查询结果，一定能映射到同一个`key`。只有这样才能保证缓存内容的正确性
4.  如何序列化查询结果？查询结果可能是单个实体对象，也可能是一个`List`。

## 解决方案 ##

### 避免脏读 ###

我们缓存了查询结果，那么一旦数据库中的数据发生变化，缓存的结果就不可用了。为了实现这一保证，可以在执行相关表的更新查询(`update`, `delete`, `insert`)查询前，让相关的缓存过期。这样下一次查询时程序就会重新从数据库中读取新数据缓存到redis中。那么问题来了，在执行一条`insert`前我怎么知道应该让哪些缓存过期呢？对于Redis，我们可以使用`Hash Set`数据结构，让一张表对应一个`Hash Set`，所有在这张表上的查询都保存到该Set下。这样当表数据发生变动时，直接让Set过期即可。我们可以自定义一个注解，在数据库查询方法上通过注解的属性注明这个操作与哪些表相关，这样在执行过期操作时，就能直接从注解中得知应该让哪些Set过期了。

### 为查询生成唯一标识 ###

对于`MyBatis`，我们可以直接使用`SQL`字符串做为`key`。但是这样就必须编写基于`MyBatis`的拦截器，从而使你的缓存代码与`MyBatis`紧紧耦合在一起。如果哪天更换了持久层的框架，你的缓存代码就白写了，所以这个方案并不完美。   
仔细想一想，其实如果两次查询调用的类名、方法名和参数值相同，我们就可以确定这两次查询结果一定是相同的（在数据没有变动的前提下）。因此，我们可以将这三个元素组合成一个字符串做为key, 就解决了标识问题。

### 序列化查询结果 ###

最方便的序列化方式就是使用JDK自带的`ObjectOutputStream`和`ObjectInputStream`。优点是几乎任何一个对象，只要实现了`Serializable`接口，都用同一套代码能被序列化和反序列化。但缺点也很致命，那就是序列化的结果容量偏大，在redis中会消耗大量内存（是对应JSON格式的3倍左右）。那么我们只剩下JSON这一个选择了。   
JSON的优点是结构紧凑，可读性强，但美中不足的是，反序列化对象时必须提供具体的类型参数(`Class`对象)，如果是`List`对象，还必须提供`List`和List中的元素类型两种信息，才能被正确反序列化。这样就增加了代码的复杂度。不过这些困难都是可以克服的，所以我们还是选择JSON作为序列化存储方式。

### 代码写在哪 ###

毫无疑问，该`AOP`上场了。在我们的例子中，持久化框架使用的是`MyBatis`，因此我们的任务就是拦截`Mapper`接口方法的调用，通过`Around(环绕通知)`编写以下逻辑：

1.  方法被调用之前，根据类名、方法名和参数值生成`Key`
2.  通过`Key`向`Redis`发起查询
3.  如果缓存命中，则将缓存结果反序列化作为方法调用的返回值 ，并阻止被代理方法的调用。
4.  如果缓存未命中，则执行代理方法，得到查询结果，序列化，用当前的`Key`将序列化结果放入redis中。

## 代码实现 ##

因为我们要拦截的是`Mapper`接口方法，因此必须命令spring使用JDK的动态代理而不是`cglib`的代理。为此，我们需要做以下配置：

<aop:aspectj-autoproxy proxy-target-class="false" />
     
      
       1
       2
       3
       4

然后定义两个标注在接口方法上的注解，用于传递类型参数：

@Retention(RetentionPolicy.RUNTIME)
    @Target(ElementType.METHOD)
    @Documented
    public @interface RedisCache { 
        Class type();
    }
     
      
       1
       2
       3
       4
       5
       6

@Retention(RetentionPolicy.RUNTIME)
    @Target(ElementType.METHOD)
    public @interface RedisEvict { 
        Class type();
    }
     
      
       1
       2
       3
       4
       5

注解的使用方式如下：

// 表示该方法需要执行 (缓存是否命中 ? 返回缓存并阻止方法调用 : 执行方法并缓存结果)的缓存逻辑
    @RedisCache(type = JobPostModel.class)
    JobPostModel selectByPrimaryKey(Integer id);
     
      
       1
       2
       3

// 表示该方法需要执行清除缓存逻辑
    @RedisEvict(type = JobPostModel.class)
    int deleteByPrimaryKey(Integer id);
     
      
       1
       2
       3

AOP的代码如下:

@Aspect
    @Component
    public class RedisCacheAspect { 
        public static final Logger infoLog = LogUtils.getInfoLogger();
    
        @Qualifier("redisTemplateForString")
        @Autowired
        StringRedisTemplate rt;
    
    
        /** * 方法调用前，先查询缓存。如果存在缓存，则返回缓存数据，阻止方法调用; * 如果没有缓存，则调用业务方法，然后将结果放到缓存中 * @param jp * @return * @throws Throwable */
        @Around("execution(* com.fh.taolijie.dao.mapper.JobPostModelMapper.select*(..))" +
                "|| execution(* com.fh.taolijie.dao.mapper.JobPostModelMapper.get*(..))" +
                "|| execution(* com.fh.taolijie.dao.mapper.JobPostModelMapper.find*(..))" +
                "|| execution(* com.fh.taolijie.dao.mapper.JobPostModelMapper.search*(..))")
        public Object cache(ProceedingJoinPoint jp) throws Throwable {
            // 得到类名、方法名和参数
            String clazzName = jp.getTarget().getClass().getName();
            String methodName = jp.getSignature().getName();
            Object[] args = jp.getArgs();
    
            // 根据类名，方法名和参数生成key
            String key = genKey(clazzName, methodName, args);
            if (infoLog.isDebugEnabled()) {
                infoLog.debug("生成key:{}", key);
            }
    
            // 得到被代理的方法
            Method me = ((MethodSignature) jp.getSignature()).getMethod();
            // 得到被代理的方法上的注解
            Class modelType = me.getAnnotation(RedisCache.class).type();
    
            // 检查redis中是否有缓存
            String value = (String)rt.opsForHash().get(modelType.getName(), key);
    
            // result是方法的最终返回结果
            Object result = null;
            if (null == value) {
                // 缓存未命中
                if (infoLog.isDebugEnabled()) {
                    infoLog.debug("缓存未命中");
                }
    
                // 调用数据库查询方法
                result = jp.proceed(args);
    
                // 序列化查询结果
                String json = serialize(result);
    
                // 序列化结果放入缓存
                rt.opsForHash().put(modelType.getName(), key, json);
            } else {
                // 缓存命中
                if (infoLog.isDebugEnabled()) {
                    infoLog.debug("缓存命中, value = {}", value);
                }
    
                // 得到被代理方法的返回值类型
                Class returnType = ((MethodSignature) jp.getSignature()).getReturnType();
    
                // 反序列化从缓存中拿到的json
                result = deserialize(value, returnType, modelType);
    
                if (infoLog.isDebugEnabled()) {
                    infoLog.debug("反序列化结果 = {}", result);
                }
            }
    
            return result;
        }
    
        /** * 在方法调用前清除缓存，然后调用业务方法 * @param jp * @return * @throws Throwable */
        @Around("execution(* com.fh.taolijie.dao.mapper.JobPostModelMapper.insert*(..))" +
                "|| execution(* com.fh.taolijie.dao.mapper.JobPostModelMapper.update*(..))" +
                "|| execution(* com.fh.taolijie.dao.mapper.JobPostModelMapper.delete*(..))" +
                "|| execution(* com.fh.taolijie.dao.mapper.JobPostModelMapper.increase*(..))" +
                "|| execution(* com.fh.taolijie.dao.mapper.JobPostModelMapper.decrease*(..))" +
                "|| execution(* com.fh.taolijie.dao.mapper.JobPostModelMapper.complaint(..))" +
                "|| execution(* com.fh.taolijie.dao.mapper.JobPostModelMapper.set*(..))")
        public Object evictCache(ProceedingJoinPoint jp) throws Throwable {
    
            // 得到被代理的方法
            Method me = ((MethodSignature) jp.getSignature()).getMethod();
            // 得到被代理的方法上的注解
            Class modelType = me.getAnnotation(RedisEvict.class).type();
    
            if (infoLog.isDebugEnabled()) {
                infoLog.debug("清空缓存:{}", modelType.getName());
            }
    
            // 清除对应缓存
            rt.delete(modelType.getName());
    
            return jp.proceed(jp.getArgs());
        }
    
    
    
        /** * 根据类名、方法名和参数生成key * @param clazzName * @param methodName * @param args 方法参数 * @return */
        protected String genKey(String clazzName, String methodName, Object[] args) {
            StringBuilder sb = new StringBuilder(clazzName);
            sb.append(Constants.DELIMITER);
            sb.append(methodName);
            sb.append(Constants.DELIMITER);
    
            for (Object obj : args) {
                sb.append(obj.toString());
                sb.append(Constants.DELIMITER);
            }
    
            return sb.toString();
        }
    
        protected String serialize(Object target) {
            return JSON.toJSONString(target);
        }
    
        protected Object deserialize(String jsonString, Class clazz, Class modelType) {
            // 序列化结果应该是List对象
            if (clazz.isAssignableFrom(List.class)) {
                return JSON.parseArray(jsonString, modelType);
            }
    
            // 序列化结果是普通对象
            return JSON.parseObject(jsonString, clazz);
        }
    }
     
      
       1
       2
       3
       4
       5
       6
       7
       8
       9
       10
       11
       12
       13
       14
       15
       16
       17
       18
       19
       20
       21
       22
       23
       24
       25
       26
       27
       28
       29
       30
       31
       32
       33
       34
       35
       36
       37
       38
       39
       40
       41
       42
       43
       44
       45
       46
       47
       48
       49
       50
       51
       52
       53
       54
       55
       56
       57
       58
       59
       60
       61
       62
       63
       64
       65
       66
       67
       68
       69
       70
       71
       72
       73
       74
       75
       76
       77
       78
       79
       80
       81
       82
       83
       84
       85
       86
       87
       88
       89
       90
       91
       92
       93
       94
       95
       96
       97
       98
       99
       100
       101
       102
       103
       104
       105
       106
       107
       108
       109
       110
       111
       112
       113
       114
       115
       116
       117
       118
       119
       120
       121
       122
       123
       124
       125
       126
       127
       128
       129
       130
       131
       132
       133
       134
       135
       136
       137
       138
       139
       140
       141
       142
       143
       144

这样我们就完成了数据库查询缓存的实现。

--------------------

UPDATE:   
最好为Hash Set设置一个过期时间，这样即使缓存策略有误（导致读出脏数据），过期时间到了以后依然可以与数据库保持同步：

// 序列化结果放入缓存
                rt.execute(new RedisCallback<Object>() {
                    @Override
                    public Object doInRedis(RedisConnection redisConn) throws DataAccessException {
                        // 配置文件中指定了这是一个String类型的连接
                        // 所以这里向下强制转换一定是安全的
                        StringRedisConnection conn = (StringRedisConnection) redisConn;
    
                        // 判断hash名是否存在
                        // 如果不存在，创建该hash并设置过期时间
                        if (false == conn.exists(hashName) ){
                            conn.hSet(hashName, key, json);
                            conn.expire(hashName, Constants.HASH_EXPIRE_TIME);
                        } else {
                            conn.hSet(hashName, key, json);
                        }
    
                        return null;
                    }
                });
    作者：司青，本文链接http://blog.csdn.net/neosmith/article/details/46811089