自动加载缓存框架

jopen 10年前

自动加载缓存框架

代码
现在使用的缓存技术很多，比如Redis、 Memcache 、 EhCache等，甚至还有使用ConcurrentHashMap 或 HashTable 来实现缓存。但在缓存的使用上，每个人都有自己的实现方式，大部分是直接与业务代码绑定，随着业务的变化，要更换缓存方案时，非常麻烦。接下来我们就使用AOP + Annotation 来解决这个问题，同时使用自动加载机制 来实现数据“常驻内存”。

Spring AOP这几年非常热门，使用也越来越多，但个人建议AOP只用于处理一些辅助的功能（比如：接下来我们要说的缓存），而不能把业务逻辑使用AOP中实现，尤其是在需要“事务”的环境中。

如下图所示：
Alt 缓存框架

AOP拦截到请求后：

根据请求参数生成Key，后面我们会对生成Key的规则，进一步说明；

如果是AutoLoad的，则请求相关参数，封装到AutoLoadTO中，并放到AutoLoadHandler中。

根据Key去缓存服务器中取数据，如果取到数据，则返回数据，如果没有取到数据，则执行DAO中的方法，获取数据，同时将数据放到缓存中。如果是AutoLoad的，则把最后加载时间，更新到AutoLoadTO中，最后返回数据；如是AutoLoad的请求，每次请求时，都会更新 AutoLoadTO中的最后请求时间。

为了减少并发，增加等待机制：如果多个用户同时取一个数据，那么先让第一个用户去DAO取数据，其它用户则等待其返回后，去缓存中获取，尝试一定次数后，如果还没获取到，再去DAO中取数据。

AutoLoadHandler（自动加载处理器）主要做的事情：当缓存即将过期时，去执行DAO的方法，获取数据，并将数据放到缓存中。为了防止自动加载队列过大，设置了容量限制；同时会将超过一定时间没有用户请求的也会从自动加载队列中移除，把服务器资源释放出来，给真正需要的请求。

使用自加载的目的:

避免在请求高峰时，因为缓存失效，而造成数据库压力无法承受;

把一些耗时业务得以实现。

把一些使用非常频繁的数据，使用自动加载，因为这样的数据缓存失效时，最容易造成服务器的压力过大。

分布式自动加载

如果将应用部署在多台服务器上，理论上可以认为自动加载队列是由这几台服务器共同完成自动加载任务。比如应用部署在A,B两台服务器上，A服务器自动加载了数据D，（因为两台服务器的自动加载队列是独立的，所以加载的顺序也是一样的），接着有用户从B服务器请求数据D，这时会把数据D的最后加载时间更新给B服务器，这样B服务器就不会重复加载数据D。

使用方法

1. 实现com.jarvis.cache.CacheGeterSeter

下面举个使用Redis做缓存服务器的例子：

package com.jarvis.example.cache;  import ... ...  /**   * 缓存切面，用于拦截数据并调用Redis进行缓存操作   */  @Aspect  public class CachePointCut implements CacheGeterSeter<Serializable> {        private static final Logger logger=Logger.getLogger(CachePointCut.class);        private AutoLoadHandler<Serializable> autoLoadHandler;        private static List<RedisTemplate<String, Serializable>> redisTemplateList;        public CachePointCut() {          autoLoadHandler=new AutoLoadHandler<Serializable>(10, this, 20000);      }        @Pointcut(value="execution(public !void com.jarvis.example.dao..*.*(..)) && @annotation(cahce)", argNames="cahce")      public void daoCachePointcut(Cache cahce) {          logger.info("----------------------init daoCachePointcut()--------------------");      }        @Around(value="daoCachePointcut(cahce)", argNames="pjp, cahce")      public Object controllerPointCut(ProceedingJoinPoint pjp, Cache cahce) throws Exception {          return CacheUtil.proceed(pjp, cahce, autoLoadHandler, this);      }        public static RedisTemplate<String, Serializable> getRedisTemplate(String key) {          if(null == redisTemplateList || redisTemplateList.isEmpty()) {              return null;          }          int hash=Math.abs(key.hashCode());          Integer clientKey=hash % redisTemplateList.size();          RedisTemplate<String, Serializable> redisTemplate=redisTemplateList.get(clientKey);          return redisTemplate;      }        @Override      public void setCache(final String cacheKey, final CacheWrapper<Serializable> result, final int expire) {          try {              final RedisTemplate<String, Serializable> redisTemplate=getRedisTemplate(cacheKey);              redisTemplate.execute(new RedisCallback<Object>() {                    @Override                  public Object doInRedis(RedisConnection connection) throws DataAccessException {                      byte[] key=redisTemplate.getStringSerializer().serialize(cacheKey);                      JdkSerializationRedisSerializer serializer=(JdkSerializationRedisSerializer)redisTemplate.getValueSerializer();                      byte[] val=serializer.serialize(result);                      connection.set(key, val);                      connection.expire(key, expire);                      return null;                  }              });          } catch(Exception ex) {              logger.error(ex.getMessage(), ex);          }      }        @Override      public CacheWrapper<Serializable> get(final String cacheKey) {          CacheWrapper<Serializable> res=null;          try {              final RedisTemplate<String, Serializable> redisTemplate=getRedisTemplate(cacheKey);              res=redisTemplate.execute(new RedisCallback<CacheWrapper<Serializable>>() {                    @Override                  public CacheWrapper<Serializable> doInRedis(RedisConnection connection) throws DataAccessException {                      byte[] key=redisTemplate.getStringSerializer().serialize(cacheKey);                      byte[] value=connection.get(key);                      if(null != value && value.length > 0) {                          JdkSerializationRedisSerializer serializer=                              (JdkSerializationRedisSerializer)redisTemplate.getValueSerializer();                          @SuppressWarnings("unchecked")                          CacheWrapper<Serializable> res=(CacheWrapper<Serializable>)serializer.deserialize(value);                          return res;                      }                      return null;                  }              });          } catch(Exception ex) {              logger.error(ex.getMessage(), ex);          }          return res;      }        /**       * 删除缓存       * @param cs Class       * @param method       * @param arguments       * @param subKeySpEL       * @param deleteByPrefixKey 是否批量删除       */      public static void delete(@SuppressWarnings("rawtypes") Class cs, String method, Object[] arguments, String subKeySpEL,          boolean deleteByPrefixKey) {          try {              if(deleteByPrefixKey) {                  final String cacheKey=CacheUtil.getCacheKeyPrefix(cs.getName(), method, arguments, subKeySpEL) + "*";                  for(final RedisTemplate<String, Serializable> redisTemplate : redisTemplateList){                      redisTemplate.execute(new RedisCallback<Object>() {                          @Override                          public Object doInRedis(RedisConnection connection) throws DataAccessException {                              byte[] key=redisTemplate.getStringSerializer().serialize(cacheKey);                              Set<byte[]> keys=connection.keys(key);                              if(null != keys && keys.size() > 0) {                                  byte[][] keys2=new byte[keys.size()][];                                  keys.toArray(keys2);                                  connection.del(keys2);                              }                              return null;                          }                      });                  }                } else {                  final String cacheKey=CacheUtil.getCahcaheKey(cs.getName(), method, arguments, subKeySpEL);                  final RedisTemplate<String, Serializable> redisTemplate=getRedisTemplate(cacheKey);                  redisTemplate.execute(new RedisCallback<Object>() {                        @Override                      public Object doInRedis(RedisConnection connection) throws DataAccessException {                          byte[] key=redisTemplate.getStringSerializer().serialize(cacheKey);                            connection.del(key);                            return null;                      }                  });              }          } catch(Exception ex) {              logger.error(ex.getMessage(), ex);          }      }        public AutoLoadHandler<Serializable> getAutoLoadHandler() {          return autoLoadHandler;      }        public void destroy() {          autoLoadHandler.shutdown();          autoLoadHandler=null;      }        public List<RedisTemplate<String, Serializable>> getRedisTemplateList() {          return redisTemplateList;      }        public void setRedisTemplateList(List<RedisTemplate<String, Serializable>> redisTemplateList) {          CachePointCut.redisTemplateList=redisTemplateList;      }    }

从上面的代码可以看出，对缓存的操作，还是由业务系统自己来实现的，我们只是对AOP拦截到的ProceedingJoinPoint，进行做一些处理。

java代码实现后，接下来要在spring中进行相关的配置：

<aop:aspectj-autoproxy proxy-target-class="true"/>  <bean id="cachePointCut" class="com.jarvis.example.cache.CachePointCut" destroy-method="destroy">      <property name="redisTemplateList">          <list>              <ref bean="redisTemplate1"/>              <ref bean="redisTemplate2"/>          </list>      </property>  </bean>

2. 将需要使用缓存的方法前增加@Cache注解

package com.jarvis.example.dao;  import ... ...  public class UserDAO {      @Cache(expire=600, autoload=true, requestTimeout=72000)      public List<UserTO> getUserList(... ...) {          ... ...      }  }

缓存Key的生成

生成缓存Key的生成方法：CacheUtil.getCahcaheKey(String className, String method, Object[] arguments, String subKeySpEL)

className 类名称
method 方法名称
arguments 参数
subKeySpEL SpringEL表达式

生成的Key格式为：{类名称}.{方法名称}{.SpringEL表达式运算结果}:{参数值的Hash字符串}

SpringEL表达式的作用

根据业务的需要，将缓存Key进行分组。举个例子，商品的评论列表：

package com.jarvis.example.dao;  import ... ...  public class GoodsCommentDAO{      @Cache(expire=600, subKeySpEL="#args[0]", autoload=true, requestTimeout=18000)      public List<CommentTO> getCommentListByGoodsId(Long goodsId, int pageNo, int pageSize) {          ... ...      }  }

如果商品Id为：100，那么生成缓存Key格式为:com.jarvis.example.dao.GoodsCommentDAO.getCommentListByGoodsId.100:xxxx
在Redis中，能精确删除商品Id为100的评论列表，执行命令即可：
del com.jarvis.example.dao.GoodsCommentDAO.getCommentListByGoodsId.100:*

SpringEL表达式使用起来确实非常方便，如果需要，@Cache中的expire，requestTimeout以及autoload参数都可以用SpringEL表达式来动态设置。但使用起来就变得复杂，所以我们没有这样做。

数据实时性

上面商品评论的例子中，如果用户发表了评论，要立即显示该如何来处理？

比较简单的方法就是，在发表评论成功后，立即把缓存中的数据也清除，这样就可以了。

package com.jarvis.example.dao;  import ... ...  public class GoodsCommentDAO{      @Cache(expire=600, subKeySpEL="#args[0]", autoload=true, requestTimeout=18000)      public List<CommentTO> getCommentListByGoodsId(Long goodsId, int pageNo, int pageSize) {          ... ...      }      public void addComment(Long goodsId, String comment) {          ... ...// 省略添加评论代码          deleteCache(goodsId);      }      private void deleteCache(Long goodsId) {          Object arguments[]=new Object[]{goodsId};          CachePointCut.delete(this.getClass(), "getCommentListByGoodsId", arguments, "#args[0]", true);      }  }

@Cache

@Retention(RetentionPolicy.RUNTIME)  @Target(ElementType.METHOD)  public @interface Cache {        /**       * 缓存的过期时间，单位：秒       */      int expire();        /**       * 是否启用自动加载缓存       * @return       */      boolean autoload() default false;        /**       * 当autoload为true时，缓存数据在 requestTimeout 秒之内没有使用了，就不进行自动加载数据,如果requestTimeout为0时，会一直自动加载       * @return       */      long requestTimeout() default 36000L;        /**       * 使用SpEL，将缓存key，根据业务需要进行二次分组       * @return       */      String subKeySpEL() default "";  }

注意事项

1. 当@Cache中 autoload 设置为 ture 时，对应方法的参数必须都是Serializable的。

AutoLoadHandler中需要缓存通过深度复制后的参数。

2. 参数中只设置必要的属性值，在DAO中用不到的属性值尽量不要设置，这样能避免生成不同的缓存Key，降低缓存的使用率。

例如：

public CollectionTO<AccountTO> getAccountByCriteria(AccountCriteriaTO criteria)  {          List<AccountTO> list=null;          PaginationTO paging=criteria.getPaging();          if(null != paging && paging.getPageNo() > 0 && paging.getPageSize() > 0) {// 如果需要分页查询，先查询总数              criteria.setPaging(null);// 减少缓存KEY的变化，在查询记录总数据时，不用设置分页相关的属性值              Integer recordCnt=accountDAO.getAccountCntByCriteria(criteria);              if(recordCnt > 0) {                  criteria.setPaging(paging);                  paging.setRecordCnt(recordCnt);                  list=accountDAO.getAccountByCriteria(criteria);              }              return new CollectionTO<AccountTO>(list, recordCnt, criteria.getPaging().getPageSize());          } else {              list=accountDAO.getAccountByCriteria(criteria);              return new CollectionTO<AccountTO>(list, null != list ? list.size() : 0, 0);          }      }

3. 注意AOP失效的情况;

例如：

TempDAO {            public Object a() {              return b().get(0);          }            @Cache(expire=600)          public List<Object> b(){              return ... ...;          }      }

通过 new TempDAO().a() 调用b方法时，AOP失效，也无法进行缓存相关操作。

4. 自动加载缓存时，不能在缓存方法内叠加查询参数值;

例如：

@Cache(expire=600, autoload=true)      public List<AccountTO> getDistinctAccountByPlayerGet(AccountCriteriaTO criteria) {          List<AccountTO> list;          int count=criteria.getPaging().getThreshold() ;          // 查预设查询数量的10倍          criteria.getPaging().setThreshold(count * 10);          … …      }

因为自动加载时，AutoLoadHandler 缓存了查询参数，执行自动加载时，每次执行时 threshold 都会乘以10，这样threshold的值就会越来越大。

5. 当方法返回值类型改变了怎么办？

在代码重构时，可能会出现改方法返回值类型的情况，而参数不变的情况，那上线部署时，可能会从缓存中取到旧数据类型的数据，可以通过以下方法处理：

上线后，快速清理缓存中的数据；
在CacheGeterSeter的实现类中统一加个version；
在@Cache中加version（未实现）。

6. 对于一些比较耗时的方法尽量使用自动加载。

7. 对于查询条件变化比较剧烈的，不要使用自动加载机制。

比如，根据用户输入的关键字进行搜索数据的方法，不建议使用自动加载。

使用规范

将调接口或数据库中取数据，封装在DAO层，不能什么地方都有调接口的方法。
自动加载缓存时，不能在缓存方法内叠加（或减）查询条件值，但允许设置值。
DAO层内部，没使用@Cache的方法，不能调用加了@Cache的方法，避免AOP失效。
因缓存Key是方法参数转为字符串获得的，为了避免生成的Key不同，尽量只设置必要的参数及属性，也便于反向定位。
对于比较大的系统，要进行模块化设计，这样可以将自动加载，均分到各个模块中。

为什么要使用自动加载机制？

首页我们想一下系统的瓶颈在哪里？

在高并发的情况下数据库性能极差，即使查询语句的性能很高；如果没有自动加载机制的话，在当缓存过期时，访问洪峰到来时，很容易就使数据压力大增。
往缓存写数据与从缓存读数据相比，效率也差很多，因为写缓存时需要分配内存等操作。使用自动加载，可以减少同时往缓存写数据的情况，同时也能提升缓存服务器的吞吐量。

如何减少DAO层并发

使用缓存；
使用自动加载机制；“写”数据往往比读数据性能要差，使用自动加载也能减少写并发。
从DAO层加载数据时，增加等待机制（拿来主义）：如果有多个请求同时请求同一个数据，会先让其中一个请求去取数据，其它的请求则等待它的数据。

可扩展性及维护性

通过AOP实现缓存与业务逻辑的解耦；如果要实时显示数据，还是会有点耦合。
非常方便更换缓存服务器或缓存实现（比如：从Memcache换成Redis）；
非常方便增减缓存服务器（如：增加Redis的节点数）；
非常方便增加或去除缓存，方便测试期间排查问题；

自动加载缓存框架

自动加载缓存框架

使用方法

1. 实现com.jarvis.cache.CacheGeterSeter

2. 将需要使用缓存的方法前增加@Cache注解

缓存Key的生成

SpringEL表达式的作用

数据实时性

@Cache

注意事项

1. 当@Cache中 autoload 设置为 ture 时，对应方法的参数必须都是Serializable的。

2. 参数中只设置必要的属性值，在DAO中用不到的属性值尽量不要设置，这样能避免生成不同的缓存Key，降低缓存的使用率。

3. 注意AOP失效的情况;

4. 自动加载缓存时，不能在缓存方法内叠加查询参数值;

5. 当方法返回值类型改变了怎么办？

6. 对于一些比较耗时的方法尽量使用自动加载。

7. 对于查询条件变化比较剧烈的，不要使用自动加载机制。

使用规范

为什么要使用自动加载机制？

如何减少DAO层并发

可扩展性及维护性

相关经验

目录