Redisson使用场景及原理

目录

一、前言

二、安装Redis

1、Windows安装Redis

​2、启动方式

3、设置密码

三、项目集成Redission客户端

1、引入依赖

四、实用场景

1、操作缓存

2、分布式锁

3、限流

3.1 创建限流器

3.2 设置限流参数

3.3 获取令牌

3.4 带超时时间获取令牌

3.5 总结

---

一、前言

Redis是一个开源的高性能键值存储数据库，它提供了多种数据结构来存储数据，如字符串、哈希、列表、集合、有序集合等。Redis将数据存储在内存中，以提供快速的读写访问速度，并且能够通过异步的方式将数据持久化到磁盘上。它支持复制、Lua脚本、事务处理、不同级别的持久化选项以及多种客户端语言的接口。Redis广泛用于缓存、消息队列、短时数据存储和高性能的应用场景中。

通常在SpringBoot项目中集成redis有两种方式：spring-boot-starter-data-redis和redisson-spring-boot-starter。但它们在功能、使用方式、性能以及集成方面存在一些差异。下面是对这两者的详细对比：

1. 集成方式
Spring Data Redis:
是Spring框架的一部分，提供了对Redis的高级抽象，使得Redis操作更加面向对象和易于使用。通常与Spring Boot一起使用，通过spring-boot-starter-data-redis依赖自动配置。使用Jedis或Lettuce作为底层客户端。Redisson:
是一个独立的Redis客户端，提供了比Spring Data Redis更丰富的功能，如分布式数据结构（如RMap, RSet, RQueue等），分布式锁和各种原子操作。需要手动配置和使用，不直接集成到Spring框架中，但可以通过Redisson Spring Boot Starter简化集成。主要使用Netty进行网络通信，支持多种序列化机制。2. 功能特性
Spring Data Redis:
支持基本的Redis操作，如键值对存储、列表、集合、有序集合等。提供模板类（如RedisTemplate），简化Redis操作。支持发布/订阅、地理空间等高级功能。Redisson:
提供了比Spring Data Redis更广泛的分布式数据结构支持。支持分布式锁、信号量、原子长整型等分布式数据结构。内置了多种分布式服务（如分布式锁、原子操作），使得在分布式环境中使用Redis更加方便和高效。3. 性能和易用性
Spring Data Redis:
使用Jedis或Lettuce作为客户端，Jedis是基于阻塞IO的，而Lettuce基于Netty是非阻塞的，因此在某些场景下性能更好。易于集成和使用，特别是在Spring生态系统中。Redisson:
基于Netty，通常在性能上优于Jedis和Lettuce（特别是在高并发场景下）。提供了更丰富的分布式数据结构和工具，但在某些简单的使用场景下可能会显得过于复杂。4. 社区和支持
Spring Data Redis:
作为Spring项目的一部分，拥有庞大的社区支持和良好的文档。持续更新和维护，与Spring Boot紧密集成。Redisson:
也是一个活跃的开源项目，拥有自己的社区和文档。由于其专注于分布式数据结构和工具，因此在这些领域有很好的支持和应用案例。结论
选择spring-data-redis还是Redisson取决于你的具体需求：
如果你的项目已经在使用Spring框架，并且需要简单的Redis操作，那么spring-data-redis可能是更好的选择。如果你的项目需要更复杂的分布式数据结构和工具，特别是在分布式锁和原子操作方面，那么Redisson可能更适合你的需求。在这种情况下，虽然需要更多的手动配置，但它的功能和性能优势可能会让你觉得这是一个值得的投资。

二、安装Redis

1、Windows安装Redis

打开Redis官网，下载压缩包

解压到本地目录后，目录结构如下

2、启动方式

1、双击redis-server.exe启动

2、命令行窗口启动，在当前目录打开cmd窗口，输入：redis-server.exe redis.windows.conf

3、设置密码

Redis服务默认没有密码，如果要设置，编辑redis.windows.conf文件，找到requirepass关键字，后边是密码，修改成自定义密码后，把这行注释打开。并重启Redis服务，需要指定配置文件路径。

三、项目集成Redission客户端

本文主要介绍Redission客户端使用方式及部分高级特性原理。

1、引入依赖

<!-- https://mvnrepository.com/artifact/org.redisson/redisson-spring-boot-starter 最新版本3.45.0-->    
<dependency><groupId>org.redisson</groupId><artifactId>redisson-spring-boot-starter</artifactId><version>3.15.6</version>
</dependency>

注意不用再引入spring-boot-starter-data-redis，在redisson-spring-boot-starter内部已经添加了依赖。

四、实用场景

1、操作缓存

直接看一个demo。

public static void main(String[] args) {Config config = new Config().setTransportMode(TransportMode.NIO).setCodec(new JsonJacksonCodec());config.useSingleServer().setAddress("redis://127.0.0.1:6379").setPassword("123456");RedissonClient redissonClient = Redisson.create(config);RBucket<Object> bucket = redissonClient.getBucket("");// 直接设置valuebucket.set("123");// 设置value并设置过期时间bucket.set("123", 3, TimeUnit.SECONDS);redissonClient.shutdown();
}

显然Redission虽然也能支持Redis常见操作，但是api入门门槛较高，相比于RedisTemplate大量简单且直观的方法确实不易使用。

2、分布式锁

不过Redission也有优势，比如在分布式锁，提供了几个简单方法即可实现。比如：

public static void main(String[] args) {Config config = new Config().setTransportMode(TransportMode.NIO).setCodec(new JsonJacksonCodec());config.useSingleServer().setAddress("redis://127.0.0.1:6379").setPassword("123456");RedissonClient redissonClient = Redisson.create(config);RLock lock = redissonClient.getLock("myLock");// 用法1：直接上锁，需要在最后手动释放锁lock.tryLock();// 用法2：上锁并设置等待时间lock.tryLock(10, TimeUnit.SECONDS);// 用法3：上锁并设置等待时间、自动释放时间lock.tryLock(10, 30, TimeUnit.SECONDS);
}

上述3个tryLock方法最终都会执行

private RFuture<Boolean> tryAcquireOnceAsync(long waitTime, long leaseTime, TimeUnit unit, long threadId)
1. waitTime：等待时间，即在尝试获取锁时最多的等待时间。如果超过这个时间仍未获取到锁，则会放弃获取锁。
2. leaseTime：租约时间，即获取到锁后持有的时间。如果在这段时间内没有手动释放锁，则系统会自动释放锁。默认为-1，即如果不手动释放，则锁永久有效。
3. unit：时间单位，用于指定等待时间和租约时间的单位。
4. threadId：当前线程id
区别是，第一个方法传入的waitTime和leaseTime都是-1，第二个方法传入的leaseTime是-1

需要特别注意：如果调用了第3个方法获取锁，并且leaseTime不是-1，则会在leaseTime过期后，释放锁。

这里就该提到大家都知道的看门狗机制。即获取分布式锁后，执行业务方法，如果在业务方法执行耗时比较久，则后台有个线程会一直给锁续约，前提是leaseTime=-1

private <T> RFuture<Long> tryAcquireAsync(long waitTime, long leaseTime, TimeUnit unit, long threadId) {RFuture<Long> ttlRemainingFuture;if (leaseTime != -1) {ttlRemainingFuture = tryLockInnerAsync(waitTime, leaseTime, unit, threadId, RedisCommands.EVAL_LONG);} else {ttlRemainingFuture = tryLockInnerAsync(waitTime, internalLockLeaseTime,TimeUnit.MILLISECONDS, threadId, RedisCommands.EVAL_LONG);}ttlRemainingFuture.onComplete((ttlRemaining, e) -> {if (e != null) {return;}// lock acquiredif (ttlRemaining == null) {if (leaseTime != -1) {internalLockLeaseTime = unit.toMillis(leaseTime);} else {// leaseTime=-1，调用scheduleExpirationRenewal方法为当前线程续约scheduleExpirationRenewal(threadId);}}});return ttlRemainingFuture;
}protected void scheduleExpirationRenewal(long threadId) {ExpirationEntry entry = new ExpirationEntry();ExpirationEntry oldEntry = EXPIRATION_RENEWAL_MAP.putIfAbsent(getEntryName(), entry);if (oldEntry != null) {oldEntry.addThreadId(threadId);} else {entry.addThreadId(threadId);renewExpiration();}
}

通常使用分布式锁处理逻辑如下：

RLock lock = redissonClient.getLock(cacheKey);
boolean isLocked = lock.tryLock(waitTime, timeUnit);
if (isLocked) {try {// 执行业务方法} finally {if (lock.isLocked() && lock.isHeldByCurrentThread()) {lock.unlock();}}
} else {throw new RuntimeException("尝试加锁失败");
}

3、限流

当然Redission还有另外一个比较实用的功能，限流。提到限流，大家可能会想到很多实现方式，比如使用Semaphore控制并发限流，或者使用guava框架提供的限流功能。但是这些大多只适用于单机系统，或者只对单机需要限流。如果遇到分布式服务，需要全局限流，虽然也能通过一定方式实现，但是显然没有那么优雅和高效。

接下来介绍下Redission分布式限流方案：

public static void main(String[] args) {Config config = new Config().setTransportMode(TransportMode.NIO).setCodec(new JsonJacksonCodec());config.useSingleServer().setAddress("redis://127.0.0.1:6379").setPassword("123456");RedissonClient redissonClient = Redisson.create(config);RRateLimiter rateLimiter = redissonClient.getRateLimiter("myRateLimiter");boolean setRate = rateLimiter.trySetRate(RateType.OVERALL, 5, 1, RateIntervalUnit.SECONDS);if (!setRate) {System.out.println("分布式限流器创建失败，已经存在。");rateLimiter.delete();setRate = rateLimiter.trySetRate(RateType.OVERALL, 5, 1, RateIntervalUnit.SECONDS);System.out.println("分布式限流器创建" + (setRate ? "成功" : "失败"));}CountDownLatch latch = new CountDownLatch(10);for (int i = 0; i < 10; i++) {int finalI = i;new Thread(() -> {try {Thread.sleep((long) (800 * Math.random()));} catch (Exception e) {e.printStackTrace();}if (rateLimiter.tryAcquire()) {System.out.println("获取令牌成功");} else {System.out.println("Request" + finalI + "获取令牌失败");}latch.countDown();}).start();}try {latch.await();} catch (InterruptedException e) {throw new RuntimeException(e);}// 删除限流器rateLimiter.delete();
}

 注意看核心代码只有三行：

// 创建限流器
RRateLimiter rateLimiter = redissonClient.getRateLimiter("myRateLimiter");
// 设置限流参数
boolean setRate = rateLimiter.trySetRate(RateType.OVERALL, 5, 1, RateIntervalUnit.SECONDS);
// 获取令牌
rateLimiter.tryAcquire()

至于为什么这么简单神奇，接下来分析下源码。看下上边3个方法都干了啥。

3.1 创建限流器

// 方法签名
RRateLimiter getRateLimiter(String name);

@Override
public RRateLimiter getRateLimiter(String name) {// 只是创建了一个RedissonRateLimiter对象并返回，并且设置了name属性为限流器名称return new RedissonRateLimiter(commandExecutor, name);
}

3.2 设置限流参数

// 方法签名
boolean trySetRate(RateType mode, long rate, long rateInterval, RateIntervalUnit rateIntervalUnit);

@Overridepublic boolean trySetRate(RateType type, long rate, long rateInterval, RateIntervalUnit unit) {return get(trySetRateAsync(type, rate, rateInterval, unit));}@Overridepublic RFuture<Boolean> trySetRateAsync(RateType type, long rate, long rateInterval, RateIntervalUnit unit) {return commandExecutor.evalWriteAsync(getRawName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,"redis.call('hsetnx', KEYS[1], 'rate', ARGV[1]);"+ "redis.call('hsetnx', KEYS[1], 'interval', ARGV[2]);"+ "return redis.call('hsetnx', KEYS[1], 'type', ARGV[3]);",Collections.singletonList(getRawName()), rate, unit.toMillis(rateInterval), type.ordinal());}

 其实也就是设置了几个Redis缓存，key分别是rate、interval、type。

3.3 获取令牌

// 常用几个方法签名

// 方法1：获取1个令牌

boolean tryAcquire();

// 方法2：获取多个令牌

boolean tryAcquire(long permits);

// 方法3：是方法1的变体，多了2个获取令牌超时时间参数

boolean tryAcquire(long timeout, TimeUnit unit);

// 方法4：是方法2的变体，多了2个获取令牌超时时间参数

boolean tryAcquire(long permits, long timeout, TimeUnit unit);

我们先看tryAcquire()方法调用栈：

@Override
public boolean tryAcquire() {return tryAcquire(1);
}@Override
public boolean tryAcquire(long permits) {return get(tryAcquireAsync(RedisCommands.EVAL_NULL_BOOLEAN, permits));
}

最终执行tryAcquireAsync方法，内部执行Lua脚本，保证操作的原子性。每一行脚本都加了说明，其中参数KEYS和ARGS值如下：

KEYS[1]=getRawName()，即限流器名称

KEYS[2]=getValueName()，值是{限流器名称}:value，存放的是数字，当前可用许可

KEYS[3]=getClientValueName()

KEYS[4]=getPermitsName()，值是{限流器名称}:permits，数据结构zset，score是当前获取许可的时间戳

KEYS[5]=getClientPermitsName()

ARGV[1]=value，获取的许可数量

ARGV[2]=System.currentTimeMillis()，当前时间戳

ARGV[3]=ThreadLocalRandom.current().nextLong()，一个随机数

private <T> RFuture<T> tryAcquireAsync(RedisCommand<T> command, Long value) {return commandExecutor.evalWriteAsync(getRawName(), LongCodec.INSTANCE, command,# rate 限流速率"local rate = redis.call('hget', KEYS[1], 'rate');"# interval 限流间隔+ "local interval = redis.call('hget', KEYS[1], 'interval');"# type 限流类型，RateType枚举下标，所以OVERALL=0，PER_CLIENT=1+ "local type = redis.call('hget', KEYS[1], 'type');"+ "assert(rate ~= false and interval ~= false and type ~= false, 'RateLimiter is not initialized')"# valueName 值是{name}:value，存放的是数字，当前可用许可+ "local valueName = KEYS[2];"# permitsName 值是{name}:permits，数据结构zset，score是当前获取许可的时间戳+ "local permitsName = KEYS[4];"# type=PER_CLIENT时+ "if type == '1' then "# valueName 值是{name}:value:managerId+ "valueName = KEYS[3];"# permitsName 值是{name}:permits:managerId+ "permitsName = KEYS[5];"+ "end;"# 参数校验：限流速率rate >= 当前请求许可（不传默认是1）+ "assert(tonumber(rate) >= tonumber(ARGV[1]), 'Requested permits amount could not exceed defined rate'); "# currentValue 获取当前还有多少许可+ "local currentValue = redis.call('get', valueName); "+ "if currentValue ~= false then " # 不是第一次获取许可# expiredValues 已过期的许可# zrangebyscore返回有序集合中指定分数区间(0,当前时间戳-限流区间]的成员列表，有序集成员按分数值递增(从小到大)次序排列。+ "local expiredValues = redis.call('zrangebyscore', permitsName, 0, tonumber(ARGV[2]) - interval); "# 获取过期许可总数+ "local released = 0; "+ "for i, v in ipairs(expiredValues) do "# 函数struct.unpack从一个类结构字符串中解包出多个Lua值+ "local random, permits = struct.unpack('fI', v);"+ "released = released + permits;"+ "end; "# 释放过期许可+ "if released > 0 then "# zremrangebyscore移除有序集合中给定的分数区间的所有成员+ "redis.call('zremrangebyscore', permitsName, 0, tonumber(ARGV[2]) - interval); "# 当前可用许可+释放的许可数+ "currentValue = tonumber(currentValue) + released; "# 重新设置当前可用许可+ "redis.call('set', valueName, currentValue);"+ "end;"+ "if tonumber(currentValue) < tonumber(ARGV[1]) then " # 剩余许可不够#+ "local nearest = redis.call('zrangebyscore', permitsName, '(' .. (tonumber(ARGV[2]) - interval), '+inf', 'withscores', 'limit', 0, 1); "# 返回下一个许可需要等待多少时间+ "return tonumber(nearest[2]) - (tonumber(ARGV[2]) - interval);"+ "else " # 剩余许可足够+ "redis.call('zadd', permitsName, ARGV[2], struct.pack('fI', ARGV[3], ARGV[1])); "# 将当前可用许可数-获取许可数+ "redis.call('decrby', valueName, ARGV[1]); "+ "return nil; "+ "end; "+ "else " # 第一次获取许可# 设置可用许可数，首次等于限流速率rate+ "redis.call('set', valueName, rate); "# 函数struct.pack将多个Lua值打包成一个类结构（struct-like）字符串[fI,nextLong(),获取许可数]+ "redis.call('zadd', permitsName, ARGV[2], struct.pack('fI', ARGV[3], ARGV[1])); "# 将当前可用许可数-获取许可数+ "redis.call('decrby', valueName, ARGV[1]); "+ "return nil; "+ "end;",Arrays.asList(getRawName(), getValueName(), getClientValueName(), getPermitsName(), getClientPermitsName()),value, System.currentTimeMillis(), ThreadLocalRandom.current().nextLong());
}

按照上述Lua脚本逻辑，我们来模拟下获取令牌的过程

假设初始列表：860 1200 1300 1800
第一次，当前是1850，则1850-1000=850，回收(0,850]，没有可回收。直接放入1850，此时列表变成：860 1200 1300 1800 1850。剩余许可0第二次，当前是1900，获取可释放许可，1900-1000=900，即(0,900]，有860，回收后列表变成：1200 1300 1800 1850。剩余许可1
如果获取1个，则足够，并放入1900，此时列表变成：1200 1300 1800 1850 1900。剩余许可0
如果获取3个，则不够，此时最近一个(1900-1000,正无穷大]，所以是1200，需要等待1200-(1900-1000)=300ms。获取失败，当前列表：1200 1300 1800 1850。剩余许可1第三次，只等待了100ms，即当前是2000，获取可释放许可，2000-1000=1000，即(0,1000]，没有可回收，当前列表：1200 1300 1800 1850。剩余许可1第四次，又等待了200ms，即当前是2200，获取可释放许可，2200-1000=1200，即(0,1200]，有1200，回收后列表变成：1300 1800 1850。剩余许可2
如果获取3个，则不够，此时最近一个(2200-1000,正无穷大]，所以是1300，需要等待1300-(2200-1000)=100ms。获取失败，当前列表：1300 1800 1850。剩余许可2第五次，又等待了100ms，即当前是2300，获取可释放许可，2300-1000=1300，即(0,1300]，有1300，回收后列表变成：1800 1850。剩余许可3
如果获取3个，则足够。

3.4 带超时时间获取令牌

@Override
public RFuture<Boolean> tryAcquireAsync(long permits, long timeout, TimeUnit unit) {RPromise<Boolean> promise = new RedissonPromise<Boolean>();long timeoutInMillis = -1;if (timeout >= 0) {timeoutInMillis = unit.toMillis(timeout);}tryAcquireAsync(permits, promise, timeoutInMillis);return promise;
}private void tryAcquireAsync(long permits, RPromise<Boolean> promise, long timeoutInMillis) {long s = System.currentTimeMillis();RFuture<Long> future = tryAcquireAsync(RedisCommands.EVAL_LONG, permits);future.onComplete((delay, e) -> {if (e != null) {// 发生异常，获取令牌失败promise.tryFailure(e);return;}// delay是获取下一个令牌需要等待的时间。如果不需要等待，表示获取令牌成功if (delay == null) {promise.trySuccess(true);return;}// 获取令牌超时时间。如果设置为-1，则退化成tryAcquire(long permits)。即一直到获取成功后返回if (timeoutInMillis == -1) {// 延迟delay之后再获取令牌commandExecutor.getConnectionManager().getGroup().schedule(() -> {tryAcquireAsync(permits, promise, timeoutInMillis);}, delay, TimeUnit.MILLISECONDS);return;}// 上一次获取令牌已消耗时间long el = System.currentTimeMillis() - s;// 剩余超时时间，如果<=0，表示超时时间已到，获取令牌失败long remains = timeoutInMillis - el;if (remains <= 0) {promise.trySuccess(false);return;}// 如果剩余超时时间<下一个令牌等待时间，即等不到获取下一个令牌已经超时了，则延迟remains之后，获取令牌失败if (remains < delay) {commandExecutor.getConnectionManager().getGroup().schedule(() -> {promise.trySuccess(false);}, remains, TimeUnit.MILLISECONDS);} else {long start = System.currentTimeMillis();// 延迟delay之后，开始获取令牌commandExecutor.getConnectionManager().getGroup().schedule(() -> {// 从创建线程到开始执行消耗时间long elapsed = System.currentTimeMillis() - start;if (remains <= elapsed) {// 如果剩余超时时间<从创建线程到开始执行消耗时间，即线程开始时已经超时了，获取令牌失败promise.trySuccess(false);return;}// 重新计算剩余超时时间并获取令牌tryAcquireAsync(permits, promise, remains - elapsed);}, delay, TimeUnit.MILLISECONDS);}});
}

清除限流器

根据如上代码分析，限流器一旦创建并设置参数后，会在Redis中长期缓存几个key，分别是rate、interval、type。如果不处理，会一直存在。假如服务异常宕机，重启时，再次创建限流器可能会创建失败。遇到这种情况，可以先手动删除限流器。

// 删除限流器
rateLimiter.delete();

3.5 小结

1. Redission分布式限流使用脚本巧妙的运用了Lua脚本，以及Redis中zset数据结构及操作方法。实现了获取令牌的逻辑。
2. Redission实现的限流器，在当前db上只能创建一个。因为rate、interval、type都是全局的。如果需要，可以指定db。在其他db上创建别的限流器。因此最多可以创建16个分布式限流器。理论上可以把这几个key也添加分组器前缀，不知道后边版本会不会支持。或者也可以自己重写方法实现。
3. 如果剩余令牌数不足，会返回下一个令牌需要等待多久。但是如果要一次获取多个令牌，可能还需要等待N轮才能成功。
4. 如果在所有接口入口都添加获取令牌代码，侵入性太强。可以通过Spring AOP方式对controller接口拦截并限流。