SpringBoot使用Redisson操作Redis及使用场景实战

前言

在SpringBoot使用RedisTemplate、StringRedisTemplate操作Redis中，我们介绍了RedisTemplate以及如何SpringBoot如何通过RedisTemplate、StringRedisTemplate操作Redis。
RedisTemplate的好处就是基于SpringBoot自动装配的原理，使得整合redis时比较简单。

那既然SrpingBoot可以通过RedisTemplate操作Redis，为何又出现了Redisson呢？Rddisson 中文文档
Reddissin也是一个redis客户端，其在提供了redis基本操作的同时，还具备其他客户端一些不具备的高精功能，例如：分布式锁+看门狗、分布式限流、远程调用等等。Reddissin的缺点是api抽象，学习成本高。

一、概述

从 spring-boot 2.x 版本开始，spring-boot-data-redis 默认使用 Lettuce 客户端操作数据。

1.1 Lettuce

SpringBoot2之后，默认就采用了lettuce。
是高级Redis客户端，基于Netty框架的事件驱动的通信层，用于线程安全同步，异步和响应使用，支持集群，Sentinel，管道和编码器。
Lettuce的API是线程安全的，可以操作单个Lettuce连接来完成各种操作，连接实例（StatefulRedisConnection）可在多个线程间并发访问。

1.2 Reddisson

基于Netty框架的事件驱动的通信层，方法是异步的，API线程安全，可操作单个Redisson连接来完成各种操作。
实现了分布式和可扩展的Java数据结构，不支持字符串操作，不支持排序、事务、管道、分区等Redis特性。
提供很多分布式相关操作服务，如，分布式锁，分布式集合，可通过 Redis支持延迟队列。

总结：优先使用Lettuce，需要分布式锁，分布式集合等分布式的高级特性，添加Redisson结合使用。

二、Spring-Boot整合Redisson

2.1 引入依赖

<dependency><groupId>org.redisson</groupId><artifactId>redisson-spring-boot-starter</artifactId><version>3.13.6</version>
</dependency>

注意：引入此依赖后，无需再引入spring-boot-starter-data-redis，其redisson-spring-boot-starter内部已经进行了引入，且排除了 Redis 的 Luttuce 以及 Jedis 客户端。因此，在 application.yaml 中 Luttuce 和 Jedis 的配置是不会生效的。

在项目使用 Redisson 时，我们一般会使用 RedissonClient 进行数据操作，但有朋友或许觉得 RedissonClient 操作不方便，或者更喜欢使用 RedisTemplate 进行操作，其实这两者是可以共存的，我们只需要再定义RedisTemplate的配置类即可。参考SpringBoot使用RedisTemplate、StringRedisTemplate操作Redis。

发现项目引入 Redisson 后，RedisTemplate底层所用的连接工厂也是 Redisson。
在这里插入图片描述

2.2 配置文件

在application.yaml中添加redis的配置信息。

spring:data:redis:mode: master# 地址host: 30.46.34.190# 端口，默认为6379port: 6379# 密码，没有不填password: ''# 几号库database: 1sentinel:master: masternodes: 30.46.34.190cluster:nodes: 30.46.34.190lettuce:pool:# 连接池的最大数据库连接数max-active: 200# 连接池最大阻塞等待时间（使用负值表示没有限制）max-wait: -1ms# 连接池中的最大空闲连接max-idle: 50# 连接池中的最小空闲连接min-idle: 8

2.3 配置类

@Configuration
@EnableConfigurationProperties({RedisProperties.class})
public class RedissonConfig {private static final String REDIS_PROTOCOL_PREFIX = "redis://";@Value("${spring.data.redis.mode}")private String redisMode;private final RedisProperties redisProperties;public RedissonConfig(RedisProperties redisProperties) {this.redisProperties = redisProperties;}/*** 逻辑参考 RedissonAutoConfiguration#redisson()*/@Bean(destroyMethod = "shutdown")public RedissonClient redisson(@Autowired(required = false) List<RedissonAutoConfigurationCustomizer> redissonAutoConfigurationCustomizers) throws IOException {Config config = new Config();config.setCheckLockSyncedSlaves(false);int max = redisProperties.getLettuce().getPool().getMaxActive();int min = redisProperties.getLettuce().getPool().getMinIdle();switch (redisMode) {case "master": {SingleServerConfig singleConfig = config.useSingleServer().setAddress(REDIS_PROTOCOL_PREFIX + redisProperties.getHost() + ":" + redisProperties.getPort()).setDatabase(redisProperties.getDatabase()).setPassword(redisProperties.getPassword());if (redisProperties.getConnectTimeout() != null) {singleConfig.setConnectTimeout((int) redisProperties.getConnectTimeout().toMillis());}singleConfig.setConnectionPoolSize(max);singleConfig.setConnectionMinimumIdleSize(min);}break;case "sentinel": {String[] nodes = convert(redisProperties.getSentinel().getNodes());SentinelServersConfig sentinelConfig = config.useSentinelServers().setMasterName(redisProperties.getSentinel().getMaster()).addSentinelAddress(nodes).setDatabase(redisProperties.getDatabase()).setPassword(redisProperties.getPassword());if (redisProperties.getConnectTimeout() != null) {sentinelConfig.setConnectTimeout((int) redisProperties.getConnectTimeout().toMillis());}sentinelConfig.setMasterConnectionPoolSize(max);sentinelConfig.setMasterConnectionMinimumIdleSize(min);sentinelConfig.setSlaveConnectionPoolSize(max);sentinelConfig.setSlaveConnectionMinimumIdleSize(min);}break;case "cluster": {String[] nodes = convert(redisProperties.getCluster().getNodes());ClusterServersConfig clusterConfig = config.useClusterServers().addNodeAddress(nodes).setPassword(redisProperties.getPassword());if (redisProperties.getConnectTimeout() != null) {clusterConfig.setConnectTimeout((int) redisProperties.getConnectTimeout().toMillis());}clusterConfig.setMasterConnectionMinimumIdleSize(min);clusterConfig.setMasterConnectionPoolSize(max);clusterConfig.setSlaveConnectionMinimumIdleSize(min);clusterConfig.setSlaveConnectionPoolSize(max);}break;default:throw new IllegalArgumentException("无效的redis mode配置");}if (redissonAutoConfigurationCustomizers != null) {for (RedissonAutoConfigurationCustomizer customizer : redissonAutoConfigurationCustomizers) {customizer.customize(config);}}return Redisson.create(config);}private String[] convert(List<String> nodesObject) {List<String> nodes = new ArrayList<String>(nodesObject.size());for (String node : nodesObject) {if (!node.startsWith(REDIS_PROTOCOL_PREFIX)) {nodes.add(REDIS_PROTOCOL_PREFIX + node);} else {nodes.add(node);}}return nodes.toArray(new String[0]);}
}

2.4 使用方式

@Component
public class RedissonService {@Resourceprotected RedissonClient redissonClient;public void redissonExists(String key){RBucket<String> rBucketValue = redissonClient.getBucket(key, StringCodec.INSTANCE);if (rBucketValue.isExists()){String value = rBucketValue.get();// doSomething} else {// doElseSomething}}}

2.5 实用场景

2.5.1 分布式锁

有点经验的同学一提到使用分布式锁便联想到了redis，那redis如何实现分布式锁呢？

分布式锁本质上要实现的目标就是在Redis中占一个坑（简单的说，就是萝卜占坑的道理),当别的进程也要来占坑时，发现那个坑里已经有一个颗大萝卜时，就只好放弃或者稍后重试。

分布式锁常用手段

1.使用setNx命令
这个命令的详细描述是（set if not exists)，如果指定key不存在则设置(成功占坑)，在业务执行完成后，调用del命令删该key（释放坑）。比如：

# set 锁名 值
setnx distribution-lock  locked// dosomingdel  distribution-lock

但这个命令存在一个问题，如果执行逻辑中出现问题，可能导致del指令无法执行，那么该锁就会成为死锁了。
可能有小伙伴贴心的想到了，我们可以给这个key再设置一个过期时间呀。比如：

setnx distribution-lock  lockedexpire distribution-lock  10// dosomingdel  distribution-batch

即使这样操作后，该逻辑仍有问题，由于 setnx 与 expire 是两条命令，如果在 setnx 与 expire 之间，redis 服务器挂了，就会导致 expire 不会执行，从而过期时间设置失败，该锁仍会成为死锁。

根源是 setnx 与 expire 两条命令并不是原子命令

且redis的事物也无法解决 setnx 与 expire 的问题，因为 expire 是依赖于 setnx 的执行结果的，如果 setnx 没有成功，expire则不应该执行。事物又无法进行if else判断，故 setnx+expire 方式实现分布式锁，并不是优解。

2.使用setNx Ex 命令
上方已经说了 setNx+expire 的问题，Redis官方为了解决这个问题，在2.8版本时引入了 set指令的扩展参数，使得 setnx 与 expire命令可以一起执行。比如：

# set 锁名 值 ex 过期时间（单位:秒) nx
set distribution-lock locked ex 5 nx// doSomthingdel distribution-lock

从逻辑上来讲，setNx Ex 已是优解了，不会使该分布式锁成为死锁。

但在我们开发中，或许仍会出现问题，为什么呢？
由于我们一开始为此锁设置了一个过期时间，那假如我们的业务逻辑执行耗时超过了设置的过期时间呢？就会出现一个线程未执行完毕，第二个线程可能持有了这个分布式锁的情况。
所以呢，如果使用 setNx Ex 组合，必须要确保自己的锁的超时时间大于占锁后的业务执行时间

3.使用lua脚本+watch dog自动延期机制
这个方案在网上一找一大堆，在此就不做详细赘述。

Redisson实现分布式锁

上方介绍的 setNx 与 setNx Ex 命令，都是Redis 服务器为我们提供的原生命令，也或多或少的存在着一部分问题，为解决setNx Ex命令存在着业务逻辑大于锁超时时间的问题，Redisson内部提供了一个监控锁的看门狗，它的作用是在Redisson实例被关闭前，不断的延长锁的有效期。默认情况下，看门狗的检查锁的超时时间是30秒钟（就是续期30s)，也可以通过修改Config.lockWatchdogTimeout来另行指定，锁的初始过期时间默认也是30s。

// 加锁以后10秒钟自动解锁
// 无需调用unlock方法手动解锁
lock.lock(10, TimeUnit.SECONDS);// 尝试加锁，最多等待100秒，上锁以后10秒自动解锁
boolean res = lock.tryLock(100, 10, TimeUnit.SECONDS);
if (res) {try {...} finally {lock.unlock();}
}

@Resource
RedissonClient redissonClient;@GetMapping("/testDistributionLock")
public BaseResponse<String> testRedission(){RLock lock = redissonClient.getLock("redis:distributionLock");try {boolean locked = lock.tryLock(10, 3, TimeUnit.SECONDS);if(locked){log.info("获取锁成功");Thread.sleep(100);return ResultUtils.success("ok" );}else{log.error("获取锁失败");return ResultUtils.error(ErrorCode.SYSTEM_ERROR);}} catch (InterruptedException e) {throw new BusinessException(ErrorCode.SYSTEM_ERROR,"出异常了");} finally {lock.unlock();}
}