SpringBoot整合Spring-data-redis实现集中式缓存

Spring Data redis的几种序列化

从框架的角度来看，存储在Redis中的数据只是字节。虽然说Redis支持多种数据类型，但是那只是意味着存储数据的方式，而不是它所代表的内容。由我们将这些数据转化成字符串或者是其他对象。我们通过org.springframework.data.redis.serializer. RedisSerializer将自定义的对象数据和存储在Redis上的原始数据之间相互转换，顾名思义，它处理的就是序列化的过程。

先看一下RedisSerializer接口

public interface RedisSerializer<T> {

/**
* 把一个对象序列化二进制数据
*/
byte[] serialize(T t) throws SerializationException;

/**
* 通过给定的二进制数据反序列化成对象
*/
T deserialize(byte[] bytes) throws SerializationException;
}

注意这里作者提示我们：Redis does not accept null keys or values but can return null replies (fornon existing keys). 大致意思Redis不接受key为null，但是对于那些不存在的key,会返回null。但是这里可以采用官方提供的org.springframework.cache.support.NullValue作为null的占位符.

NullValue源码如下：

public final class NullValue implements Serializable {

static final Object INSTANCE = new NullValue();

private static final long serialVersionUID = 1L;

private NullValue() {
}

private Object readResolve() {
return INSTANCE;
}

}

下面是RedisSerializer接口的几种实现方式：



首先在RedisTemplate中，我们可以看到afterPropertiesSet()方法

public void afterPropertiesSet() {
super.afterPropertiesSet();
boolean defaultUsed = false;
if (defaultSerializer == null) {
defaultSerializer = new JdkSerializationRedisSerializer(
classLoader != null ? classLoader : this.getClass().getClassLoader());
}
if (enableDefaultSerializer) {
if (keySerializer == null) {
keySerializer = defaultSerializer;
defaultUsed = true;
}
if (valueSerializer == null) {
valueSerializer = defaultSerializer;
defaultUsed = true;
}
if (hashKeySerializer == null) {
hashKeySerializer = defaultSerializer;
defaultUsed = true;
}
if (hashValueSerializer == null) {
hashValueSerializer = defaultSerializer;
defaultUsed = true;
}
}
if (enableDefaultSerializer && defaultUsed) {
Assert.notNull(defaultSerializer, "default serializer null and not all serializers initialized");
}
if (scriptExecutor == null) {
this.scriptExecutor = new DefaultScriptExecutor<K>(this);
}
initialized = true;
}

这个方法时org.springframework.beans.factory .InitializingBean接口声明的一个方法，这个接口主要就是做一些初始化动作，或者检查已经设置好bean的属性。或者在XML里面加入一个init-method，这里我们可以知道默认的RedisSerializer就是JdkSerializationRedisSerializer,而StringRedisTemplate的默认序列化全是public
class StringRedisTemplate extendsRedisTemplate<String, String>

public StringRedisTemplate() {
RedisSerializer<String> stringSerializer = new StringRedisSerializer();
setKeySerializer(stringSerializer);
setValueSerializer(stringSerializer);
setHashKeySerializer(stringSerializer);
setHashValueSerializer(stringSerializer);
}

public class StringRedisSerializer implements RedisSerializer<String> {

private final Charset charset;

public StringRedisSerializer() {
this(Charset.forName("UTF8"));
}

public StringRedisSerializer(Charset charset) {
Assert.notNull(charset);
this.charset = charset;
}

public String deserialize(byte[] bytes) {
return (bytes == null ? null : new String(bytes, charset));
}

public byte[] serialize(String string) {
return (string == null ? null : string.getBytes(charset));
}
}

RedisCacheManager

RedisCacheManager的父类是AbstractTransactionSupportingCacheManager，有名字可以知道是对事务支持的一个CacheManager，默认是不会感知事务

privateboolean transactionAware =
false;

对此官网的解释是：

Set whether this CacheManager should exposetransaction-aware Cache objects.

Default is "false". Set this to"true" to synchronize cache put/evict

operations with ongoing Spring-managedtransactions, performing the actual cache

put/evict operation only in theafter-commit phase of a successful transaction.

大致意思是可感知事务的意思，put,evict,意味着会改变cache，所以put,evict操作必须一个事务（同步操作），其他线程必须等正在进行put,evict操作的线程执行完，才能紧接着操作。

再往上抽取的类就是AbstractCacheManager

取几个比较经典的方法：

// Lazy cache initialization on access
@Override
public Cache getCache(String name) {
Cache cache = this.cacheMap.get(name);
if (cache != null) {
return cache;
}
else {
// Fully synchronize now for missing cache creation...
synchronized (this.cacheMap) {
cache = this.cacheMap.get(name);
if (cache == null) {
cache = getMissingCache(name);
if (cache != null) {
cache = decorateCache(cache);
this.cacheMap.put(name, cache);
updateCacheNames(name);
}
}
return cache;
}
}
}

这个cacheMap就是非常经典的并发容器ConcurrentHashMap，它Spring自带管理cache的工具，每个Java开发人员都应该去读一下它的实现思想。。。

private finalConcurrentMap<String, Cache> cacheMap = newConcurrentHashMap<String, Cache>(16);

不用说，我们直接可以想到肯定set管理cache的name。

private volatileSet<String> cacheNames = Collections.emptySet();

关于volatile，不用多说，原子性，可见性每个Java开发人员都应该理解的。。

这个synchronized，不用多说。。也是必须理解的- -，getMissingCache()这个方法默认返回null,决定权交给其实现者，可以根据name创建，也可以记录日志什么的。

protected Cache getMissingCache(String name) {
return null;
}

decorateCache（）这个方法顾名思义就是装饰这个cache,默认直接返回，就是我们经典的装饰模式，IO类库的设计里面也有这个装饰模式。所以说常用的设计模式也必须要掌握啊。

protected Cache decorateCache(Cache cache) {
return cache;
}

这里就在子类AbstractTransactionSupportingCacheManager，里面去根据isTransactionAware字段去判断是否进行事务可感知来修饰这个cache。

@Override
protected Cache decorateCache(Cache cache) {
return (isTransactionAware() ? new TransactionAwareCacheDecorator(cache) : cache);
}

updateCacheNames（）这个方法

private void updateCacheNames(String name) {
Set<String> cacheNames = new LinkedHashSet<String>(this.cacheNames.size() + 1);
cacheNames.addAll(this.cacheNames);
cacheNames.add(name);
this.cacheNames = Collections.unmodifiableSet(cacheNames);
}

一个有顺序的set集合，最后用Collections包装成一个不能修改的set视图，LinkedHashSet也是非常有必要去了解一下底层原理的。。

配置RedisCacheManager非常简单，首先RedisCacheManager依赖RedisTemplate，RedisTemplate又依赖于连接工厂，这里就是我们的RedisConnectionFactory的实现类

JedisConnectionFactory，关于这个连接工厂：

Note: Though the database index isconfigurable, the JedisConnectionFactory only supports connecting to one Redisdatabase at a time.

Because Redis is single threaded, you areencouraged to set up multiple instances of Redis instead of using multipledatabases within a single process. This allows you to get better CPU/resourceutilization.

大意就是：虽然数据库索引是可配置的，但JedisConnectionFactory只支持一次连接到一个Redis数据库。由于Redis是单线程的，因此建议您设置多个Redis实例，而不是在一个进程中使用多个数据库。这可以让你获得更好的CPU /资源利用率。

默认是下面配置：

· hostName=”localhost”
· port=6379
· timeout=2000 ms
· database=0
· usePool=true
先看下属性

//Redis具体操作的类
@SuppressWarnings("rawtypes")//
private final RedisOperations redisOperations;
//是否使用前缀修饰cache
private boolean usePrefix = false;
// usePrefix = true的时候使用默认的前缀DefaultRedisCachePrefix，是：
private RedisCachePrefix cachePrefix = new DefaultRedisCachePrefix();
//远程加载缓存
private boolean loadRemoteCachesOnStartup = false;
//当super的缓存不存在时，是否创建缓存，false的话就不会去创建缓存
private boolean dynamic = true;

// 0 - never expire  永不过期
private long defaultExpiration = 0;
// 针对专门的key设置缓存过期时间
private Map<String, Long> expires = null;
//缓存的名字集合
private Set<String> configuredCacheNames;

这里官方强烈建议我们开启使用前缀。redisCacheManager.setUsePrefix（true），因为这里默认为false。

/**
* @param cacheName must not be {@literal null} or empty.
* @param keyPrefix can be {@literal null}.
*/
public RedisCacheMetadata(String cacheName, byte[] keyPrefix) {

hasText(cacheName, "CacheName must not be null or empty!");
this.cacheName = cacheName;
this.keyPrefix = keyPrefix;

StringRedisSerializer stringSerializer = new StringRedisSerializer();

// name of the set holding the keys
this.setOfKnownKeys = usesKeyPrefix() ? new byte[] {} : stringSerializer.serialize(cacheName + "~keys");
this.cacheLockName = stringSerializer.serialize(cacheName + "~lock");
}

这里我们通过追踪源码可以看见构造RedisCaheMetdata的setOfKnownKeys时候会生成一个后缀为~keys的key，而这个key的在Redis中类型是zset，它是维护已知key的一个有序set，底层是LinkedHashSet。同时我们也会在官网中看到：

By default RedisCacheManager does notprefix keys for cache regions, which can lead to an unexpected growth of a ZSETused to maintain known keys. It’s highly recommended to enable the usage ofprefixes in order to avoid this unexpected growth and potential
key clashesusing more than one cache region.

大致意思就是默认情况下，RedisCacheManager不会为缓存区域创建前缀，这样会导致维护管理已知的那些key的那个zset会急剧增长（ps:这个zset的name就是上面说的setOfKnownKeys）。因此强烈建议开启默认前缀，以免这个zset意外增长以及使用多个缓冲区域带来的潜在冲突。

关于这个cacheLockName，是cache名称后缀为~lock的key，作为一把锁存放在Redis服务器上。而RedisCache其中clear方法用于清除当前cache块中所有的元素，这里会加锁，而锁的实现就是在服务器上面放刚才key是cacheLockName的元素，最后清除锁则是在clear方法执行完成后在finally中清除。 put与get方法运行时会查看是否存在lock锁，存在则会sleep
300毫秒。这个过程会一直继续，直到redis服务器上不存在锁时才会进行相应的get与put操作，这里存在一个问题，如果clear方法运行时间很长，这时当前运行clear操作的机子挂了，就导致lock元素一直存在于redis服务器上。

之后就算这个机子重新启动后，也无法正常使用cache。原因是：get与put方法在运行时，锁lock始终存在于redis服务器上，所以在使用时应当小心避免这种问题。下面可以追踪下源码看下：

在RedisCache类中的静态抽象内部类LockingRedisCacheCallback<T>中，我们可以看见在Cache的元数据中设置锁。

@Override
public T doInRedis(RedisConnection connection) throws DataAccessException {

if (connection.exists(metadata.getCacheLockKey())) {
return null;
}
try {
connection.set(metadata.getCacheLockKey(), metadata.getCacheLockKey());
return doInLock(connection);
} finally {
connection.del(metadata.getCacheLockKey());
}
}

在RedisCache类中的静态抽象内部类中，static abstract classAbstractRedisCacheCallback<T>

private long WAIT_FOR_LOCK_TIMEOUT = 300;
protected boolean waitForLock(RedisConnection connection) {

boolean retry;
boolean foundLock = false;
do {
retry = false;
if(connection.exists(cacheMetadata.getCacheLockKey())) {
foundLock = true;
try {
Thread.sleep(WAIT_FOR_LOCK_TIMEOUT);
} catch (InterruptedException ex) {
Thread.currentThread().interrupt();
}
retry = true;
}
} while (retry);

return foundLock;
}

connection.exists(cacheMetadata.getCacheLockKey()就是判断哪个锁是否还在Redis中。下面我们可以简单测试下：

@RequestMapping(value = "save/{key}.do", method = RequestMethod.POST)
@Cacheable(value = "cache",key = "#key",condition = "#key !=  ''")
public String save( @PathVariable String key)  {
System.out.println("走数据库");
System.out.println(cacheManager.getCacheNames());
return "succful";
}

当我们这样设置的时候

redisCacheManager.setUsePrefix(false);

redisCacheManager.setDefaultExpiration(60*30);

这时候就会产生一个cache名+~keys的一个zset维护key的名字的一个集合



redisCacheManager.setUsePrefix(true);
redisCacheManager.setCachePrefix(new MyRedisCachePrefix());

这个MyRedisCachePrefix实现了MyRedisCachePrefix接口，默认是DefaultRedisCachePrefix()是：作为分隔符，这里只是换成了#。同时这时候我们的这些缓存都有了命名空间cache加上我们自定义的#分隔符，防止了缓存的冲突。



上面对于值的序列化都统一采用了Jackson2JsonRedisSerializer

template.setValueSerializer(jackson2JsonRedisSerializer);，对于key的序列化采用了

StringRedisSerializer。

对于序列的化采用是跟String交互的多就用StringRedisSerializer，存储POLO类的Json数据时就用Jackson2JsonRedisSerializer。

对于data-redis封装Cache来说，好处是非常明显的，既可以很方便的缓存对象，相比较于SpringCache、Ecache这些进程级别的缓存来说，现在缓存的内存的是使用redis的内存，不会消耗JVM的内存，提升了性能。当然这里Redis不是必须的，换成其他的缓存服务器一样可以，只要实现Spring的Cache类，并配置到XML里面就行。和原生态的jedis相比，只要方法上加上注解，就可以实现缓存，对于应用开发人员来说，使用缓存变的简单。同时也有利于对不同的业务缓存进行分组统计、监控。

附录：SpringBoot整合data-redis

Redis单节点：

@Bean
public RedisConnectionFactory redisConnectionFactory() {
JedisConnectionFactory cf = new JedisConnectionFactory();
cf.setHostName("10.188.182.140");
cf.setPort(6379);
cf.setPassword("root");
cf.afterPropertiesSet();
return cf;
}

或者在application.properites、application.yml文件里配置
Spring.redis.host: 172.26.223.153
Spring.redis.port: 6379
哨兵模式：
类注解：@RedisSentinelConfiguration或者@ PropertySource
配置文件
·

spring.redis.sentinel.master

:mymaster
·

spring.redis.sentinel.nodes

: 127.0.0.1:6379

@Bean
public RedisConnectionFactory jedisConnectionFactory() {
RedisSentinelConfiguration sentinelConfig = new RedisSentinelConfiguration() .master("mymaster")
.sentinel("127.0.0.1", 26379) .sentinel("127.0.0.1", 26380);
return new JedisConnectionFactory(sentinelConfig);
}

集群模式：
·

spring.redis.cluster.nodes

:node1,node2…….
·

spring.redis.cluster.max-redirects

: 集群之间最大重定向次数

@Component
@ConfigurationProperties(prefix = "spring.redis.cluster")
public class ClusterConfigurationProperties {

/*
* spring.redis.cluster.nodes[0] = 127.0.0.1:7379
* spring.redis.cluster.nodes[1] = 127.0.0.1:7380
* ...
*/
List<String> nodes;

/**
* Get initial collection of known cluster nodes in format {@code host:port}.
*
* @return
*/
public List<String> getNodes() {
return nodes;
}

public void setNodes(List<String> nodes) {
this.nodes = nodes;
}
}

@Configuration
public class AppConfig {

/**
* Type safe representation of application.properties
*/
@Autowired ClusterConfigurationProperties clusterProperties;

public @Bean RedisConnectionFactory connectionFactory() {

return new JedisConnectionFactory(
new RedisClusterConfiguration(clusterProperties.getNodes()));
}
}

开启缓存,注意默认@SpringBootApplication会扫描当前同级目录及其子目录的带有@Configuration的类(仅仅支持1.2+版本的springboot，之前是有三个注解@Configuration、@ComponentScan、@EnableAtuoConfiguration)，当启动类上使用@ComponentScan注解的时候就只会扫描你自定义的基础包。当有xml.文件的时候，建议在@Configuration类上面
@ImportResource({"classpath:xxx.xml","classpath:yyy.xml"})导入
又或者当你的你@Configuration配置类没有默认在@SpringBootApplication扫描的路径下，可以使用@Import({xxx.class,yyy.class})

@Configuration
@EnableCaching
public class RedisConfig extends CachingConfigurerSupport {
@Bean
public CacheManager cacheManager(@SuppressWarnings("rawtypes") RedisTemplate redisTemplate) {
RedisCacheManager redisCacheManager = new RedisCacheManager(redisTemplate);
redisCacheManager.setDefaultExpiration(60*30);
redisCacheManager.setTransactionAware(true);
redisCacheManager.setUsePrefix(true);
redisCacheManager.setCachePrefix(new MyRedisCachePrefix());
return redisCacheManager;
}

@Bean
@SuppressWarnings({ "rawtypes", "unchecked" })
public RedisTemplate<String, String> redisTemplate(RedisConnectionFactory factory) {
StringRedisTemplate template = new StringRedisTemplate(factory);
//使用Jackson2JsonRedisSerializer来序列化和反序列化redis的value值
Jackson2JsonRedisSerializer jackson2JsonRedisSerializer = new Jackson2JsonRedisSerializer(Object.class);
ObjectMapper om = new ObjectMapper();
om.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY);
om.enableDefaultTyping(ObjectMapper.DefaultTyping.NON_FINAL);
jackson2JsonRedisSerializer.setObjectMapper(om);
//使用StringRedisSerializer来序列化和反序列化redis的key值
template.setValueSerializer(jackson2JsonRedisSerializer);
template.afterPropertiesSet();
return template;
}

}