SpringBoot 集成 RedisTemplate Cluster集群
RedisTemplate
前言
Redis就不多做介绍了，直接进入正题，通过本篇将学习到（代码地址：https://gitee.com/chaitou/leilema.git）：
-
Redis常见功能
-
Redis高可用分布式
-
Springboot集成RedisTemplate的正确姿势
-
集成Spring Cache
常见误区(瑞士军刀开瓶盖)
初学者往往认为Redis就是缓存，这其实是个误区，仅仅拿Redis当缓存好比拿瑞士军刀开瓶盖，但是Redis能做的远不止如此，以下列举几种Redis的常见应用
-
缓存(也是最常见的)
-
分布式锁、数据结构(常见于分布式架构的系统，对分布式有较高要求的小伙伴可以考虑集成Redission)
-
统计(通过RedisBitmap 位图或者hyperLogLog可以实现在极小空间消耗的情况下进行用户统计等功能)
-
消息队列(对于只有简单消息队列需求的系统来说，通过Redis发布订阅 + 队列就足够了，不一定非要集成Rabbitmq之类的中间件)
-
GEO地理位置计算可以用于实现像微信摇一摇、附近商家等功能
Redis高可用分布式
单机版
Redis只有一个实例，没有任何高可用分布式可言，只适合于初学者学习时使用，生产环境是绝对不允许这种情况出现的。一旦这个Redis实例崩溃了，小则缓存失效，全部数据查询走数据库，数据库访问需求暴增。大则影响分布式锁的等功能造成业务异常
高可用Sentinel
如上图，Sentinel模式也称之为哨兵模式，该模式下拥有多个节点，当其中的master节点出现故障时，其他节点会自动顶替master节点，继续提供服务，实现高可用。由于篇幅有限，这里做个简单的原理介绍：
首先可以看到图上只有一个master节点（主节点），多个slave节点（从节点）。slave从节点根据一定的机制去复制主节点的数据，起到备份作用，也就是备胎，随时等待上位的那种。(当然，这里还有一个功能，可以根据系统情况做读写分离，只在master写，只在slave读)
每个Sentinel每隔一段时间就会向所有的Redis节点发送心跳检测，来监控Redis节点是否正常。如果Sentinel1发现其中一个Redis1节点死掉了，为了公平起见，那么他就会表态：“Redis1节点死掉了，谁赞成谁反对？”。
此时的所有Sentinel都会表态，当大多数Sentinel觉得这个redis节点死掉时，那就说明他死掉了。如果这个节点是master节点，那么Sentinel就会挑选一个新的slave节点作为master节点，同时告诉所有slave节点要求成为该新master的slave节点。如果死掉的是slave节点，那就只需要通知以下slave节点死掉了，毕竟他不是master
而对于客户端来说，也就是我们的Java程序来说，我们不再直连Redis节点了，我们需要连接的是Sentinel节点，让Sentinel节点告诉我们真实的Redis节点信息。当然了，这些工作Jedis或者其他客户端都帮我们做好了，只需要做个配置就行
高可用集群Cluster
Sentinel模式做到了高可用，但是实质还是只有一个master在提供服务(读写分离的情况本质也是master在提供服务)，当master节点所在的机器内存不足以支撑系统的数据时，就需要考虑集群了。
如上图所示，Cluster集群有多个Redis节点，每个节点负责一部分槽。也就是说Redis总共拥有16384个哈西槽，我们指定节点各自负责的槽。假设有3个节点，那么1节点可以负责1-5461，2节点负责5462-10922，3节点负责10923-16384。当我们要存储一个key时，key通过一致性hash算法寻找应该落到的槽，然后找到其对应Redis节点进行存储。这样就实现了Redis集群。
当然，考虑到稳定性，我们一般会给没每个节点设置slave从节点，确保该集群的高可用。因此Cluster经常听到的三主三从指的就是3个master集群，同时拥有3个slave从节点。
对比
单机版就不对比了，没什么意义。关键是Cluster集群与Sentinel的对比
-
Cluster集群可扩展性强，当一台机器不够用时，加机器重新分配槽就可以解决性能瓶颈。同时Cluster也是高可用的，一旦出现某个节点宕机，从节点会自动替补上去。同时当数据量大时，Cluster每个节点只负责一小部分槽，在确保命中率的情况下，性能更好 -
说了这么多是不是意味着
Sentinel对比起Cluster就一无是处了呢？当然不是，Cluster虽然好，但是几乎只要涉及多key操作的命令，Cluster都是不支持的。比如mget、mset、pipeline等。原因也很好理解，mget key1 key2 key3 ...，这上面的key都分布在不同的cluster节点上，一条命令怎么可能解决这个问题呢？我们能做的只有将所有key取出来，再进行分类，然后去不同的Redis实例上取（当然还有可能取错实例），其他的命令读者自行分析
因此，其实Cluster并非想象中的那么好，架构师还是得根据系统情况进行分析。虽然大部分情况下我们都会选择Cluster集群，但是当系统缓存的数据量小，但是频繁需要使用sort、mget这类多key指令时，则Sentinel会更合适。还是那句话，没有最完美的架构，只有最适合的架构。
springboot集成RedisTemplate
说了这么多，正餐终于来了，本篇我们还是主要以讲解Redis Cluster为主，在集成之前，我们得先理清楚几个概念
-
Jedis、Lettuce： Jedis想必都有所耳闻，这2个都是Redis客户端，都偏向于底层，个人理解更像是JDBC
-
RedisTemplate： Spring对Redis操作的一层封装，他的底层是通过Jedis、Lettuce实现的。如果我们使用
spring-boot-starter-data-redis则默认时Lettuce
之前我们提到过Springboot使用了约定大于配置的思想，这使得我们集成Redis Cluster的RedisTemplate变得容易许多。只要我们按Springboot的约定来，就可以省去很多Bean的配置。简化归简化，原理我们还是要懂的，如果我们使用Spring集成，我们需要配置以下几个Bean
-
JedisPoolConfig： 也就是连接池配置信息，记载着最大连接数等信息。类似于数据库连接池Druid，当程序需要连接
Redis Server时，程序需要创建连接，使用完后关闭。但是频繁的打开和关闭连接不仅有损性能，同时连接数也不方便管理。连接池解决了以上问题，需要的直接到连接池取，使用完归还 -
RedisClusterConfiguration： 记载
Redis Cluster各个节点信息，如IP端口等 -
JedisConnectionFactory：
JedisPoolConfig + RedisClusterConfiguration记载着Redis连接的所有必要信息 -
RedisTemplate： 这个是我们的终极目标，通过
JedisConnectionFactory的完整信息创建出RedisTemplate Bean。需要注意的是，由于默认的序列化使用的是jdkSerializeable，关于序列话可以参考：阿里Java手册: 序列化。这种序列化存储二进制字节码，不易读也容易出现乱码，因此需要替换另外一种序列化方式，一般是采用Jackson的序列化方式，当然现在国内有很多项目都采用了阿里的Fastjson方式
引入依赖
spring-boot-starter-data-redis引入相关依赖，如果是老版本的Springboot，引入的则是spring-boot-starter-redis。同时由于默认引入的是Lettuce，而本文使用的是Jedis，因此我们需要排除Lettuce的依赖，引入Jedis依赖
<properties>
<jedis-version>3.1.0</jedis-version>
</properties>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-data-redis</artifactId>
<exclusions>
<exclusion>
<groupId>io.lettuce</groupId>
<artifactId>lettuce-core</artifactId>
</exclusion>
</exclusions>
</dependency>
<dependency>
<groupId>redis.clients</groupId>
<artifactId>jedis</artifactId>
<version>${jedis-version}</version>
</dependency>
配置RedisTemplate
既然使用了Springboot，约定大于配置。如果我们遵循了这一法则，JedisPoolConfig、RedisClusterConfiguration、JedisConnectionFactory这3个Bean是可以不需要手动配置的，而Springboot会帮我们做好，我们只需要专注于配置RedisTemplate就行
yml配置：
spring:
cache:
redis:
time-to-live: 10000
redis:
timeout: 5000
database: 0
cluster:
nodes: 148.70.139.121:7000,148.70.139.121:7001,148.70.139.121:7002,148.70.139.121:7003,148.70.139.121:7004,148.70.139.121:7005
max-redirects: 3
jedis:
pool:
max-active: 8
max-wait: -1
max-idle: 8
min-idle: 0
RedisTemplate Bean：这里需要注意一下序列化的操作
package com.bugpool.leilema.freamwork.configuration;
import com.fasterxml.jackson.annotation.JsonAutoDetect;
import com.fasterxml.jackson.annotation.PropertyAccessor;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.springframework.cache.annotation.EnableCaching;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.serializer.Jackson2JsonRedisSerializer;
import org.springframework.data.redis.serializer.StringRedisSerializer;
@Configuration
@EnableCaching
public class RedisConfiguration {
@Bean
public RedisTemplate<String, Object> redisTemplate(RedisConnectionFactory factory) {
RedisTemplate<String, Object> template = new RedisTemplate<>();
template.setConnectionFactory(factory);
// 使用Jackson2JsonRedisSerialize 替换默认的jdkSerializeable序列化
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 stringRedisSerializer = new StringRedisSerializer();
// key采用String的序列化方式
template.setKeySerializer(stringRedisSerializer);
// hash的key也采用String的序列化方式
template.setHashKeySerializer(stringRedisSerializer);
// value序列化方式采用jackson
template.setValueSerializer(jackson2JsonRedisSerializer);
// hash的value序列化方式采用jackson
template.setHashValueSerializer(jackson2JsonRedisSerializer);
template.afterPropertiesSet();
return template;
}
}
由于原生的RedisTemplate也不是非常好用，一般我们会再自己封装一层。有些人习惯把这一层称之为RedisDao，当然也有人习惯把他当RedisUtils工具类来使用，这里笔者并不纠结那种方式跟好，笔者就将他作为Service，需要是注入使用就好
RedisService：
package com.bugpool.leilema.freamwork.utils;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.TimeUnit;
@Component
@Slf4j
public class RedisService {
@Autowired
private RedisTemplate<String, Object> redisTemplate;
/**
* 指定缓存失效时间
*
* @param key 键
* @param time 时间(秒)
* @return
*/
public boolean expire(String key, long time) {
try {
if (time > 0) {
redisTemplate.expire(key, time, TimeUnit.SECONDS);
}
return true;
} catch (Exception e) {
log.error("exception when expire key {}. ", key, e);
return false;
}
}
/**
* 根据key获取过期时间
*
* @param key 键 不能为null
* @return 时间(秒) 返回0代表为永久有效
*/
public long getExpire(String key) {
return redisTemplate.getExpire(key, TimeUnit.SECONDS);
}
/**
* 判断key是否存在
*
* @param key 键
* @return true 存在 false不存在
*/
public boolean hasKey(String key) {
try {
return redisTemplate.hasKey(key);
} catch (Exception e) {
log.error("exception when check key {}. ", key, e);
return false;
}
}
/**
* 删除缓存
*
* @param key 可以传一个值 或多个
*/
@SuppressWarnings("unchecked")
public void del(String... key) {
if (key != null && key.length > 0) {
if (key.length == 1) {
redisTemplate.delete(key[0]);
} else {
redisTemplate.delete(CollectionUtils.arrayToList(key));
}
}
}
/**
* 普通缓存获取
*
* @param key 键
* @return 值
*/
public Object get(String key) {
return key == null ? null : redisTemplate.opsForValue().get(key);
}
/**
* 普通缓存放入
*
* @param key 键
* @param value 值
* @return true成功 false失败
*/
public boolean set(String key, Object value) {
try {
redisTemplate.opsForValue().set(key, value);
return true;
} catch (Exception e) {
log.error("exception when set key {}. ", key, e);
return false;
}
}
/**
* 普通缓存放入并设置时间
*
* @param key 键
* @param value 值
* @param time 时间(秒) time要大于0 如果time小于等于0 将设置无限期
* @return true成功 false 失败
*/
public boolean set(String key, Object value, long time) {
try {
if (time > 0) {
redisTemplate.opsForValue().set(key, value, time, TimeUnit.SECONDS);
} else {
set(key, value);
}
return true;
} catch (Exception e) {
log.error("exception when set key {}. ", key, e);
return false;
}
}
/**
* 递增
*
* @param key 键
* @param delta 要增加几(大于0)
* @return
*/
public long incr(String key, long delta) {
if (delta <= 0) {
throw new RuntimeException("递增因子必须大于0");
}
return redisTemplate.opsForValue().increment(key, delta);
}
/**
* 递减
*
* @param key 键
* @param delta 要减少几(小于0)
* @return
*/
public long decr(String key, long delta) {
if (delta <= 0) {
throw new RuntimeException("递减因子必须大于0");
}
return redisTemplate.opsForValue().increment(key, -delta);
}
/**
* HashGet
*
* @param key 键 不能为null
* @param item 项 不能为null
* @return 值
*/
public Object hget(String key, String item) {
return redisTemplate.opsForHash().get(key, item);
}
/**
* 获取hashKey对应的所有键值
*
* @param key 键
* @return 对应的多个键值
*/
public Map<Object, Object> hmget(String key) {
return redisTemplate.opsForHash().entries(key);
}
/**
* HashSet
*
* @param key 键
* @param map 对应多个键值
* @return true 成功 false 失败
*/
public boolean hmset(String key, Map<String, Object> map) {
try {
redisTemplate.opsForHash().putAll(key, map);
return true;
} catch (Exception e) {
log.error("exception when hash set key {}. ", key, e);
return false;
}
}
/**
* HashSet 并设置时间
*
* @param key 键
* @param map 对应多个键值
* @param time 时间(秒)
* @return true成功 false失败
*/
public boolean hmset(String key, Map<String, Object> map, long time) {
try {
redisTemplate.opsForHash().putAll(key, map);
if (time > 0) {
expire(key, time);
}
return true;
} catch (Exception e) {
log.error("exception when hash set key {}. ", key, e);
return false;
}
}
/**
* 向一张hash表中放入数据,如果不存在将创建
*
* @param key 键
* @param item 项
* @param value 值
* @return true 成功 false失败
*/
public boolean hset(String key, String item, Object value) {
try {
redisTemplate.opsForHash().put(key, item, value);
return true;
} catch (Exception e) {
log.error("exception when hash set key {}, item {} ", key, item, e);
return false;
}
}
/**
* 向一张hash表中放入数据,如果不存在将创建
*
* @param key 键
* @param item 项
* @param value 值
* @param time 时间(秒) 注意:如果已存在的hash表有时间,这里将会替换原有的时间
* @return true 成功 false失败
*/
public boolean hset(String key, String item, Object value, long time) {
try {
redisTemplate.opsForHash().put(key, item, value);
if (time > 0) {
expire(key, time);
}
return true;
} catch (Exception e) {
log.error("exception when hash set key {}, item {} ", key, item, e);
return false;
}
}
/**
* 删除hash表中的值
*
* @param key 键 不能为null
* @param item 项 可以使多个 不能为null
*/
public void hdel(String key, Object... item) {
redisTemplate.opsForHash().delete(key, item);
}
/**
* 判断hash表中是否有该项的值
*
* @param key 键 不能为null
* @param item 项 不能为null
* @return true 存在 false不存在
*/
public boolean hHasKey(String key, String item) {
return redisTemplate.opsForHash().hasKey(key, item);
}
/**
* hash递增 如果不存在,就会创建一个 并把新增后的值返回
*
* @param key 键
* @param item 项
* @param by 要增加几(大于0)
* @return
*/
public double hincr(String key, String item, double by) {
return redisTemplate.opsForHash().increment(key, item, by);
}
/**
* hash递减
*
* @param key 键
* @param item 项
* @param by 要减少记(小于0)
* @return
*/
public double hdecr(String key, String item, double by) {
return redisTemplate.opsForHash().increment(key, item, -by);
}
/**
* 根据key获取Set中的所有值
*
* @param key 键
* @return
*/
public Set<Object> sGet(String key) {
try {
return redisTemplate.opsForSet().members(key);
} catch (Exception e) {
return null;
}
}
/**
* 根据value从一个set中查询,是否存在
*
* @param key 键
* @param value 值
* @return true 存在 false不存在
*/
public boolean sHasKey(String key, Object value) {
try {
return redisTemplate.opsForSet().isMember(key, value);
} catch (Exception e) {
return false;
}
}
/**
* 将数据放入set缓存
*
* @param key 键
* @param values 值 可以是多个
* @return 成功个数
*/
public long sSet(String key, Object... values) {
try {
return redisTemplate.opsForSet().add(key, values);
} catch (Exception e) {
return 0;
}
}
/**
* 将set数据放入缓存
*
* @param key 键
* @param time 时间(秒)
* @param values 值 可以是多个
* @return 成功个数
*/
public long sSetAndTime(String key, long time, Object... values) {
try {
Long count = redisTemplate.opsForSet().add(key, values);
if (time > 0)
expire(key, time);
return count;
} catch (Exception e) {
return 0;
}
}
/**
* 获取set缓存的长度
*
* @param key 键
* @return
*/
public long sGetSetSize(String key) {
try {
return redisTemplate.opsForSet().size(key);
} catch (Exception e) {
return 0;
}
}
/**
* 移除值为value的
*
* @param key 键
* @param values 值 可以是多个
* @return 移除的个数
*/
public long setRemove(String key, Object... values) {
try {
Long count = redisTemplate.opsForSet().remove(key, values);
return count;
} catch (Exception e) {
return 0;
}
}
/**
* 获取list缓存的内容
*
* @param key 键
* @param start 开始
* @param end 结束 0 到 -1代表所有值
* @return
*/
public List<Object> lGet(String key, long start, long end) {
try {
return redisTemplate.opsForList().range(key, start, end);
} catch (Exception e) {
return null;
}
}
/**
* 获取list缓存的长度
*
* @param key 键
* @return
*/
public long lGetListSize(String key) {
try {
return redisTemplate.opsForList().size(key);
} catch (Exception e) {
return 0;
}
}
/**
* 通过索引 获取list中的值
*
* @param key 键
* @param index 索引 index>=0时， 0 表头，1 第二个元素，依次类推；index<0时，-1，表尾，-2倒数第二个元素，依次类推
* @return
*/
public Object lGetIndex(String key, long index) {
try {
return redisTemplate.opsForList().index(key, index);
} catch (Exception e) {
return null;
}
}
/**
* 将list放入缓存
*
* @param key 键
* @param value 值
* @return
*/
public boolean lSet(String key, Object value) {
try {
redisTemplate.opsForList().rightPush(key, value);
return true;
} catch (Exception e) {
return false;
}
}
/**
* 将list放入缓存
*
* @param key 键
* @param value 值
* @param time 时间(秒)
* @return
*/
public boolean lSet(String key, Object value, long time) {
try {
redisTemplate.opsForList().rightPush(key, value);
if (time > 0)
expire(key, time);
return true;
} catch (Exception e) {
return false;
}
}
/**
* 将list放入缓存
*
* @param key 键
* @param value 值
* @return
*/
public boolean lSet(String key, List<Object> value) {
try {
redisTemplate.opsForList().rightPushAll(key, value);
return true;
} catch (Exception e) {
return false;
}
}
/**
* 将list放入缓存
*
* @param key 键
* @param value 值
* @param time 时间(秒)
* @return
*/
public boolean lSet(String key, List<Object> value, long time) {
try {
redisTemplate.opsForList().rightPushAll(key, value);
if (time > 0)
expire(key, time);
return true;
} catch (Exception e) {
return false;
}
}
/**
* 根据索引修改list中的某条数据
*
* @param key 键
* @param index 索引
* @param value 值
* @return
*/
public boolean lUpdateIndex(String key, long index, Object value) {
try {
redisTemplate.opsForList().set(key, index, value);
return true;
} catch (Exception e) {
e.printStackTrace();
return false;
}
}
/**
* 移除N个值为value
*
* @param key 键
* @param count 移除多少个
* @param value 值
* @return 移除的个数
*/
public long lRemove(String key, long count, Object value) {
try {
Long remove = redisTemplate.opsForList().remove(key, count, value);
return remove;
} catch (Exception e) {
return 0;
}
}
}
使用
当我们需要使用到Redis时，使用@Autowired注入。一篇是不可能讲完所有Redis的操作的，因此举个例子，大家自己摸索。
package com.bugpool.leilema.freamwork.utils;
import com.bugpool.leilema.product.entity.ProductInfo;
import org.junit.Assert;
import org.junit.jupiter.api.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;
import java.math.BigDecimal;
import static org.junit.jupiter.api.Assertions.*;
@RunWith(SpringRunner.class)
@SpringBootTest
class RedisServiceTest {
@Autowired
RedisService redisService;
@Test
void get() {
ProductInfo productInfo = new ProductInfo();
productInfo.setProductName("推拿")
.setProductId(1)
.setProductPrice(new BigDecimal(100));
redisService.set("testRedisGet", productInfo, 100);
ProductInfo productInfo1 = (ProductInfo) redisService.get("testRedisGet");
Assert.assertTrue(productInfo1.getProductName().equals(productInfo.getProductName()));
}
}
集成Spring Cache
如果你只是想要使用Redis作为缓存，而在每个方法中都使用redisService.set("testRedisGet", productInfo, 100);去设置缓存，侵入性还是很高的。因此Spring Cache通过注解的方式，方便缓存的使用。Spring Cache的配置我们上方已经配置过了，这里拿出来再讲一遍
配置
yml： 以下配置指定了Spring Cache使用Redis做缓存，并且缓存失效时间是10s(该有效时间只针对使用@Cacheable这些注解，不影响我们RedisService的使用)
spring:
cache:
redis:
time-to-live: 10000
RedisConfiguration： 我们已经在配置RedisTemplate时加上了@EnableCaching的注解，该注解通知Spring Ioc开启Spring Cache，实质是一个后置处理器，它检查每个Spring bean是否在公共方法上有@Cacheable子类的注释。
如果找到这样的注释，则自动创建代理通过拦截方法调用处理缓存。在Jdk动态代理中我曾写过一个例子，大致原理可以参考
@EnableCaching
public class RedisConfiguration {
}
使用
@Override
@Cacheable(value = "redis", key = "#root.targetClass + '::' + #root.methodName + '::' + #productName")
public List getByLikeName(String productName) {
return productInfoMapper.getByLikeName(productName);
}
注解包括@Cacheable、@CacheEvict、@CachePut
-
@Cacheable： 每次执行方法前，会根据key查找redis是否存在缓存，如果存在则直接返回缓存结果。如果不存在，则执行方法，方法结束后，将结果放入缓存中。一般用在select查询类的方法上
-
@CachePut： 执行方法前，不管缓存是否存在，都执行方法，并且把结果放入缓存中。一般用在Update方法上
-
@CacheEvict： 清除缓存，一般放在delete方法上
-
SpringEl表达式：
key = "#root.targetClass + '::' + #root.methodName + '::' + #productName"这句话使用的就是SpringEL表达式，一般我们设置Key都是需要加上类名做前缀，防止与其他类的缓存混淆
最后要强调的是，Spring Cache的使用在大部分的场景下，提升都非常有限，想要用好Redis，还是认真分析业务场景，手动使用RedisTemplate进行优化吧！
华为开发者空间，是为全球开发者打造的专属开发空间，汇聚了华为优质开发资源及工具，致力于让每一位开发者拥有一台云主机，基于华为根生态开发、创新。
更多推荐
0
0- 0
已为社区贡献12条内容
所有评论(0)