前言

Redis就不多做介绍了，直接进入正题，通过本篇将学习到（代码地址：https://gitee.com/chaitou/leilema.git）：

1. Redis常见功能
2. Redis高可用分布式
3. Springboot集成RedisTemplate的正确姿势
4. 集成Spring Cache
   

常见误区(瑞士军刀开瓶盖)

初学者往往认为Redis就是缓存，这其实是个误区，仅仅拿Redis当缓存好比拿瑞士军刀开瓶盖，但是Redis能做的远不止如此，以下列举几种Redis的常见应用

1. 缓存(也是最常见的)
2. 分布式锁、数据结构(常见于分布式架构的系统，对分布式有较高要求的小伙伴可以考虑集成Redission)
3. 统计(通过RedisBitmap 位图或者hyperLogLog可以实现在极小空间消耗的情况下进行用户统计等功能)
4. 消息队列(对于只有简单消息队列需求的系统来说，通过Redis发布订阅 + 队列就足够了，不一定非要集成Rabbitmq之类的中间件)
5. 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的对比

1. Cluster集群可扩展性强，当一台机器不够用时，加机器重新分配槽就可以解决性能瓶颈。同时Cluster也是高可用的，一旦出现某个节点宕机，从节点会自动替补上去。同时当数据量大时，Cluster每个节点只负责一小部分槽，在确保命中率的情况下，性能更好
2. 说了这么多是不是意味着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为主，在集成之前，我们得先理清楚几个概念

1. Jedis、Lettuce：Jedis想必都有所耳闻，这2个都是Redis客户端，都偏向于底层，个人理解更像是JDBC
2. RedisTemplate：Spring对Redis操作的一层封装，他的底层是通过Jedis、Lettuce实现的。如果我们使用spring-boot-starter-data-redis则默认时Lettuce

之前我们提到过Springboot使用了约定大于配置的思想，这使得我们集成Redis Cluster的RedisTemplate变得容易许多。只要我们按Springboot的约定来，就可以省去很多Bean的配置。简化归简化，原理我们还是要懂的，如果我们使用Spring集成，我们需要配置以下几个Bean

1. JedisPoolConfig：也就是连接池配置信息，记载着最大连接数等信息。类似于数据库连接池Druid，当程序需要连接Redis Server时，程序需要创建连接，使用完后关闭。但是频繁的打开和关闭连接不仅有损性能，同时连接数也不方便管理。连接池解决了以上问题，需要的直接到连接池取，使用完归还
2. RedisClusterConfiguration： 记载Redis Cluster各个节点信息，如IP端口等
3. JedisConnectionFactory：JedisPoolConfig + RedisClusterConfiguration记载着Redis连接的所有必要信息
4. 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的操作的，因此举个例子，大家自己摸索。下一篇专门写一篇：RedisTemplate实现分布式锁

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，实质是一个后置处理器(postProcessor生命周期)，它检查每个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的使用，还是参考：Spring Cache吧，本文篇幅有限就不赘述了。但是还是要强调的是，Spring Cache的使用在大部分的场景下，提升都非常有限，想要用好Redis，还是认真分析业务场景，手动使用RedisTemplate进行优化吧

本专题目录：一步到位springboot目录
gitee代码：https://gitee.com/chaitou/leilema.git