使用Redis Cluster作为MyBatis的二级缓存并使用protostuff序列化数据

港控/mmm° 2022-06-13 02:45 162阅读 0赞

正如大多数持久层框架一样，MyBatis 同样提供了一级缓存和二级缓存的支持；  
一级缓存基于 PerpetualCache 的 HashMap 本地缓存，其存储作用域为 Session，当 Session flush 或 close 之后，该Session中的所有 Cache 就将清空。  
二级缓存与一级缓存其机制相同，默认也是采用 PerpetualCache，HashMap存储，不同在于其存储作用域为 Mapper(Namespace)，并且可自定义存储源，如 Ehcache、Redis等。

本文将使用Redis作为Mybatis的二级存储源并将缓存数据使用protostuff序列化缓存数据已节约存储空间。并为了缓存的健壮性，使用了Redis集群。

**添加依赖**

<dependency>
                <groupId>redis.clients</groupId>
                <artifactId>jedis</artifactId>
                <version>2.9.0</version>
                <scope>compile</scope>
             </dependency>
               <dependency>
                <groupId>io.protostuff</groupId>
                <artifactId>protostuff-core</artifactId>
                <version>1.4.4</version>
            </dependency>
            <dependency>
                <groupId>io.protostuff</groupId>
                <artifactId>protostuff-runtime</artifactId>
                <version>1.4.4</version>
            </dependency>

开启二级缓存，对于mybatis-config.xml修改cacheEnabled:

由于二级缓存是定义在NameSpace上的，因此需要在Mapping文件中开启二级缓存，只需要添加`<cache/>`标签即可，当然在这个标签中还是其他属性，如缓存收回策略eviction，刷新时间flushInterval，缓存数量size，是否为只读缓存readOnly。如果你是使用的注解来配置Mybatis(与Spring整合)，对于你添加@Mapper的类，此时便需要使用@CacheNamespace注解来开启二级缓存，同样，该注解也支持配置缓存的一些元数据。  
**如前所属，Mybatis的默认二级缓存是PerpetualCache，我们需要自定义二级缓存，并将缓存载体设置为Redis。**

**自定义二级缓存**  
MyBatis提供了一个接口：import org.apache.ibatis.cache.Cache。自定义二级缓存只需要实现该接口。

public interface Cache {
        
    String getId();
    int get Size();
    void putObject(Object key, Object value);
    Object getObject(Object key);
    Object removeObject(Object key);
    void clear();
    ReadWriteLock getReadWriteLock();
    }

其中，只需要重点关注putObject()与getObject()两个方法。在此之前，先将Redis集群搭建起来。  
**Redis集群的搭建**  
Redis集群有两种，一种是redis sentinel，高可用集群，同时只有一个master，各实例数据保持一致；一种是redis cluster，分布式集群，同时有多个master，数据分片部署在各个master上。  
本文将搭建的是redis cluster模式，redis官方文档介绍的非常详细[http://www.redis.cn/topics/cluster-tutorial.html][http_www.redis.cn_topics_cluster-tutorial.html]。最终将启动6个Redis服务，conf配置为:

port 7000/7001/7002/7003/7004/7006
    cluster-enabled yes
    cluster-config-file nodes.conf
    cluster-node-timeout 5000
    appendonly yes

通过 redis-server xxx.conf。启动成功后则需要使用redis-trib.rb进行哈希槽划分，即这6台集群开始共享1~16384个哈希槽(每个Redis服务拥有1~16384个哈希槽)。

创建集群:

./redis-trib.rb create --replicas 1 127.0.0.1:7000 127.0.0.1:7001 \
    127.0.0.1:7002 127.0.0.1:7003 127.0.0.1:7004 127.0.0.1:7005

**-–replicas 1 表示我们希望为集群中的每个主节点创建一个从节点,则上述执行的结果最终为包含三个主节点并且每个主节点存在一个从节点。**  
可以通过 `redis-cli -h127.0.0.1 -p 7000 cluster nodes`来查看集群状态。对于你想给集群中增加新的Redis机器，则可以调用命令来重新分配哈希槽，具体的这里不在叙述，可以去查询相关文档。

当集群搭建完成后，则在jedis客户端直接通过BinaryJedisCluster调用即可。

public BinaryJedisCluster getJedis(){
            Set<HostAndPort> jedisClusterNodes = new HashSet<HostAndPort>();
            jedisClusterNodes.add(new HostAndPort("127.0.0.1", 7000));
            //xxxx
            BinaryJedisCluster jc = new BinaryJedisCluster(jedisClusterNodes);
            return jc;
        }

**protostuff序列化**  
关于Protostuff的序列化笔者之前写过博客介绍([http://blog.csdn.net/canot/article/details/53750443][http_blog.csdn.net_canot_article_details_53750443]),这里便不在叙述，直接给出实现Cache后的代码。

```@Component
    public class RedisCache implements Cache {
        
        private BinaryJedisCluster redisTemplate = getJedis();
        private final ReadWriteLock readWriteLock = new ReentrantReadWriteLock();
        private  String id; // cache instance id
        private static final int EXPIRE_TIME_IN_MINUTES = 60*60; // redis过期时间
        private static RuntimeSchema<CacheEntry> schema = RuntimeSchema.createFrom(CacheEntry.class);
    
        public RedisCache(){
    
        }
        public RedisCache(String id){
            if (id == null) {
                throw new IllegalArgumentException("Cache instances require an ID");
            }
            this.id = id;
        }
        /*
         *mybatis缓存操作对象的标识符。一个mapper对应一个mybatis的缓存操作对象。
         */
        @Override
        public String getId() {
            return id;
        }
        /*
         *将查询结果塞入缓存
         */
        @Override
        public void putObject(Object key, Object value) {
        //查询结果为ArrayList
            ArrayList<xxx> valueList = (ArrayList<xxx>)value;
            CacheEntry cacheEntry = new CacheEntry(valueList);
            //通过protostuff序列化
            byte[] bytesValue = ProtostuffIOUtil.toByteArray(cacheEntry,schema, LinkedBuffer.allocate(LinkedBuffer.DEFAULT_BUFFER_SIZE));
            byte[] bytesKey = key.toString().getBytes(Charset.forName("UTF-8"));
     //存储与Redis
    redisTemplate.setex(bytesKey,EXPIRE_TIME_IN_MINUTES,bytesValue);
        }
        /*
         * 从缓存中获取被缓存的查询结果。
         */
        @Override
        public Object getObject(Object key) {
            byte[] bytesKey = key.toString().getBytes(Charset.forName("UTF-8"));
            byte[] bytesValue = redisTemplate.get(bytesKey);
            CacheEntry value = schema.newMessage();
            if(bytesValue!=null) {
                ProtostuffIOUtil.mergeFrom(bytesValue, value, schema);
                return value.getSeckills();
            }
            //如果返回null，则会去查询数据库
            return null;
    
        }
        /*
         *从缓存中删除对应的key、value。一般回滚触发
         */
        @Override
        public Object removeObject(Object key) {
            redisTemplate.del(key.toString().getBytes(Charset.forName("UTF-8")));
            return null;
        }
        /*
         *从缓存中删除对应的key、value
         */
        @Override
        public void clear() {
            //redisTemplate.flushDB();
           //清空操作太危险，不建议实现
        }
        /*
         *缓存的数量
         */
        @Override
        public int getSize() {
            return 1024;
        }
        /*
        *实现原子性的缓存操作使用的锁
        */
        @Override
        public ReadWriteLock getReadWriteLock() {
            return readWriteLock;
        }
    
        public BinaryJedisCluster getJedis(){
            Set<HostAndPort> jedisClusterNodes = new HashSet<HostAndPort>();
            jedisClusterNodes.add(new HostAndPort("127.0.0.1", 7000));
            //xxxx
            BinaryJedisCluster jc = new BinaryJedisCluster(jedisClusterNodes);
            return jc;
        }
    
    }
    class CacheEntry{
        private List<xxx> seckills;
        public CacheEntry(List<xxx> seckills){
            this.seckills=seckills;
        }
        public List<xxxx> getSeckills() {
            return seckills;
        }
        public void setSeckills(List<xxx> seckills) {
            this.seckills = seckills;
        }
    }

**将该Cache置为二级缓存**  
`<cache type="package.RedisCache">`  
或者  
`@CacheNamespace(implementation = package.RedisCache.class)`

**使用二级缓存的注意事项**

>  *  只能在【只有单表操作】的表上使用缓存

不只是要保证这个表在整个系统中只有单表操作，而且和该表有关的全部操作必须全部在一个namespace下。

>  *  在可以保证查询远远大于insert,update,delete操作的情况下使用缓存

这一点不需要多说，所有人都应该清楚。记住，这一点需要保证在上述的前提下才可以！

>  *  多表操作一定不能使用缓存

不管多表操作写到那个namespace下，都会存在某个表不在这个namespace下的情况。会出现脏读的情况。

[http_www.redis.cn_topics_cluster-tutorial.html]: http://www.redis.cn/topics/cluster-tutorial.html
[http_blog.csdn.net_canot_article_details_53750443]: http://blog.csdn.net/canot/article/details/53750443