前言
在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
# 端口，默认为6379
port: 6379
# 密码，没有不填
password: ''
# 几号库
database: 1
sentinel:
master: master
nodes: 30.46.34.190
cluster:
nodes: 30.46.34.190
lettuce:
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 {
@Resource
protected 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
// dosoming
del distribution-lock
但这个命令存在一个问题，如果执行逻辑中出现问题，可能导致del指令无法执行，那么该锁就会成为死锁了。
可能有小伙伴贴心的想到了，我们可以给这个key再设置一个过期时间呀。比如：
setnx distribution-lock locked
expire distribution-lock 10
// dosoming
del 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
// doSomthing
del 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();
}
}
2.5.2 限流
我们是有面临高并发下需要对接口或者业务逻辑限流的问题，我们可以采用Guaua依赖下的RateLimiter 实现，实际上，Redisssion也有类似的限流功能。
RateLimiter 被称为令牌桶限流，此类限流是首先定义好一个令牌桶，指明在一定时间内生成多少个令牌，每次访问时从令牌桶获取指定数量令牌，如果获取成功，则设为有效访问。
