redisson分布式限流RRateLimiter源码解析
作者：你行你上啊 发布时间：2021-05-29 13:10:15
标签：redisson,分布式,限流,RRateLimiter
分布式限流-单位时间多实例多线程访问次数限制
接前面聊一聊redisson及优雅实现 和 说一说spring boot优雅集成redisson，简单以源码的方式给大家介绍了redisson的：可重入性、阻塞、续约、红锁、联锁、加锁解锁流程和集成spring boot注意点和优雅实现方式。
接下来在讲一讲平时用的比较多的限流模块--RRateLimiter
1.简单使用
public static void main(String[] args) throws InterruptedException {
RRateLimiter rateLimiter = createLimiter();
int allThreadNum = 20;
CountDownLatch latch = new CountDownLatch(allThreadNum);
long startTime = System.currentTimeMillis();
for (int i = 0; i < allThreadNum; i++) {
// new Thread(() -> {
if(i ％ 3 == 0) Thread.sleep(1000);
boolean pass = rateLimiter.tryAcquire();
if(pass) {
log.info("get ");
} else {
log.info("no");
}
// latch.countDown();
// }).start();
}
// latch.await();
System.out.println("Elapsed " + (System.currentTimeMillis() - startTime));
}
public static RRateLimiter createLimiter() {
Config config = new Config();
config.useSingleServer()
.setTimeout(1000000)
.setPassword("123456")
.setAddress("redis://xxxx:6379");
RedissonClient redisson = Redisson.create(config);
RRateLimiter rateLimiter = redisson.getRateLimiter("myRateLimiter3");
// 初始化：PER_CLIENT 单实例执行，OVERALL 全实例执行
// 最大流速 = 每10秒钟产生3个令牌
rateLimiter.trySetRate(RateType.OVERALL, 3, 10, RateIntervalUnit.SECONDS);
return rateLimiter;
}
实际结果：
[2022-10-29 14:32:46.261][INFO ][main][][] RedisTest - get
[2022-10-29 14:32:46.312][INFO ][main][][] RedisTest - get
[2022-10-29 14:32:46.358][INFO ][main][][] RedisTest - get
[2022-10-29 14:32:47.416][INFO ][main][][] RedisTest - no
[2022-10-29 14:32:47.469][INFO ][main][][] RedisTest - no
[2022-10-29 14:32:47.517][INFO ][main][][] RedisTest - no
[2022-10-29 14:32:48.577][INFO ][main][][] RedisTest - no
[2022-10-29 14:32:48.623][INFO ][main][][] RedisTest - no
2. 实现限流redisson使用了哪些redis数据结构
Hash结构 -- 限流器结构：
参数rate代表速率
参数interval代表多少时间内产生的令牌
参数type代表单机还是集群
ZSET结构 -- 记录获取令牌的时间戳，用于时间对比。
1667025166312 --> 2022-10-29 14:32:46
1667025166262 --> 2022-10-29 14:32:46
1667025166215 --> 2022-10-29 14:32:46
String结构 --记录的是当前令牌桶中的令牌数【很明显被我用完了现在是0】
3. 超过10s,我再次获取一个令牌，数据结构发生的变化
ZSET结构。-- 新生成一个ZSET结构，存放获取令牌的时间戳
String 结构 --当前令牌桶还有2个令牌
4. 源码浅析
RRateLimiter rateLimiter = redisson.getRateLimiter("myRateLimiter3");
// 初始化
// 最大流速 = 每10秒钟产生3个令牌
rateLimiter.trySetRate(RateType.PER_CLIENT, 3, 10, RateIntervalUnit.SECONDS);
初始化定义没有什么好讲的，就是创建HASH结构
主要还是讲讲： rateLimiter.tryAcquire()
private <T> RFuture<T> tryAcquireAsync(RedisCommand<T> command, Long value) {
return this.commandExecutor.evalWriteAsync(
this.getName(), LongCodec.INSTANCE, command,
"local rate = redis.call('hget', KEYS[1], 'rate');
local interval = redis.call('hget', KEYS[1], 'interval');
local type = redis.call('hget', KEYS[1], 'type');
assert(
rate ~= false and interval ~= false and type ~= false, 'RateLimiter is not initialized'
)
local valueName = KEYS[2];
local permitsName = KEYS[4];
if type == '1' then valueName = KEYS[3];
permitsName = KEYS[5];
end;
local currentValue = redis.call('get', valueName);
if currentValue ~= false
then
local expiredValues = redis.call(
'zrangebyscore', permitsName, 0, tonumber(ARGV[2]
)
- interval
);
local released = 0;
for i, v in ipairs(expiredValues)
do local random, permits = struct.unpack('fI', v);
released = released + permits;end;
if released > 0
then
redis.call('zrem', permitsName, unpack(expiredValues));
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), tonumber(ARGV[2]), 'withscores', 'limit', 0, 1);
local random, permits = struct.unpack('fI', nearest[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
assert(tonumber(rate) >= tonumber(ARGV[1]), 'Requested permits amount could not exceed defined rate');
redis.call('set', valueName, rate);
redis.call('zadd', permitsName, ARGV[2], struct.pack('fI', ARGV[3], ARGV[1]));
redis.call('decrby', valueName, ARGV[1]);
return nil;
end;", Arrays.asList(this.getName(),
this.getValueName(),
this.getClientValueName(),
this.getPermitsName(),
this.getClientPermitsName()),
new Object[]{value,
System.currentTimeMillis(),
ThreadLocalRandom.current().nextLong()
}
);
}
主要就是这段lua代码，下面我详细过一下
作者目前用的3.16.3版本，刚好遇见redisson的bug，见3197，请大家用最新版本，以下为修复后解析。
-- 获取hash结构的速率
local rate = redis.call("hget", KEYS[1], "rate")
-- 获取hash结构的时间区间(ms)
local interval = redis.call("hget", KEYS[1], "interval")
-- 获取hash结构的时间类型
local type = redis.call("hget", KEYS[1], "type")
-- 判断是否初始化限流结构
assert(rate ~= false and interval ~= false and type ~= false, "RateLimiter is not initialized")
-- {name}:value string结构，这个key记录的是当前令牌桶中的令牌数
local valueName = KEYS[2]
-- {name}:permits zset结构，记录了请求的令牌数，score则为请求的时间戳
local permitsName = KEYS[4]
-- 单机限流才会用到，集群模式不用关注
if type == "1" then
valueName = KEYS[3]
permitsName = KEYS[5]
end
-- 生产速率rate必须比请求的令牌数大
assert(tonumber(rate) >= tonumber(ARGV[1]), "Requested permits amount could not exceed defined rate")
-- 初始化RateLimiter并不会初始化stirng结构，因此第一次获取这里currentValue是null
local currentValue = redis.call("get", valueName)
if currentValue ~= false then
-- 第二次获取令牌执行
-------------------------- 获取zset结构：统计之前的请求令牌数
-- 范围是0 ~ (第二次请求时间戳 - 令牌生产的时间)
local expiredValues = redis.call("zrangebyscore", permitsName, 0, tonumber(ARGV[2]) - interval)
local released = 0
-- lua迭代器，遍历expiredValues，如果有值，那么released等于之前所有请求的令牌数之和，表示应该释放多少令牌
for i, v in ipairs(expiredValues) do
-- 获取请求数permits
local random, permits = struct.unpack("fI", v)
released = released + permits
end
-- 之前的请求令牌数 > 0, 例如10s产生3个令牌，现在超过10s了，重置周期并计算剩余令牌数
if released > 0 then
-- 移除zset中所有元素【要求是同一个限流器permitsName，不然就移除不了，尴尬】
redis.call("zrem", permitsName, unpack(expiredValues))
currentValue = tonumber(currentValue) + released
------------------------- 更新string结构：=剩下令牌数+释放令牌数
redis.call("set", valueName, currentValue)
end
-- 如果当前令牌数 ＜ 请求的令牌数
if tonumber(currentValue) < tonumber(ARGV[1]) then
-- 从zset中找到距离当前时间最近的那个请求，也就是上一次放进去的请求信息
local nearest = redis.call('zrangebyscore', permitsName, '(' .. (tonumber(ARGV[2]) - interval), tonumber(ARGV[2]), 'withscores', 'limit', 0, 1);
local random, permits = struct.unpack("fI", nearest[1])
-- 返回 上一次请求的时间戳 - (当前时间戳 - 令牌生成的时间间隔) 这个值表示还需要多久才能生产出足够的令牌
return tonumber(nearest[2]) - (tonumber(ARGV[2]) - interval)
else
-- 如果当前令牌数 ≥ 请求的令牌数，表示令牌够多，更新zset
------------------------- 更新zset结构
redis.call("zadd", permitsName, ARGV[2], struct.pack("fI", ARGV[3], ARGV[1]))
------------------------- 更新Stringt结构，减少一个剩下的令牌数
redis.call("decrby", valueName, ARGV[1])
return nil
end
else
--------汀雨笔记----------------- 初始化Stringt结构，当前限流器的令牌数
redis.call("set", valueName, rate)
--------汀雨笔记----------------- 初始化zset结构
redis.call("zadd", permitsName, ARGV[2], struct.pack("fI", ARGV[3], ARGV[1]))
-- struct.pack第一个参数表示格式字符串，f是浮点数、I是长整数。所以这个格式字符串表示的是把一个浮点数和长整数拼起来的结构体，
-- ARGV[2]就是请求时间戳，ARGV[1]是请求的令牌数，统计会用到，ARGV[3]是当前时间戳为种子的随机数，具体用处还不知道，知道的网友可以留言
------------------------- 更新Stringt结构，因为这是获取令牌操作，减掉一个令牌
-------------------------【本文作者认为，这里可以直接初始化string结构，值为rate - 1】
redis.call("decrby", valueName, ARGV[1])
return nil
end
这段lua代码也并不复杂，令牌桶的数量主要是通过时间窗口来控制，判断上一个请求是否超过了令牌生产周期。
留下一个疑问？
-- 移除zset中所有元素【要求是同一个限流器permitsName，不然就移除不了，尴尬】
redis.call("zrem", permitsName, unpack(expiredValues))
我自己在本地测试，只要超过10s，permitsName就不一样，这就导致了这部分数据是不能移除的，就产生了冗余数据，从前面的截图也可以看出，是新生成了一个zset数据结构。
相当于直接走到了这一步：
------------------------- 更新zset结构
redis.call("zadd", permitsName, ARGV[2], struct.pack("fI", ARGV[3], ARGV[1]))
至于为什么会产生这样的结果，会的小伙伴可以留言，或者过段时间我提个issue。
来源：https://juejin.cn/post/7159872509478141959
0
投稿猜你喜欢
一、前言(吐槽+煽情+简介) &n- java实现在线预览- -之poi实现word、excel、ppt转html，具体内容如下所示：###简介java实现在线预览功能是一个大家
- 前言SQL注入漏洞作为WEB安全的最常见的漏洞之一，在java中随着预编译与各种ORM框架的使用，注入问题也越来越少。新手代码审计者往往对J
- 很多的Android入门程序猿来说对于Android自定义View，可能都是比较恐惧的，但是这又是高手进阶的必经之路，所有准备在自定义Vie
- 一、延迟加载：LazyLoading使用延迟加载，关联的实体必须标注为virtual。本例是标注Destination类里的Lodgings
- （注意：本文基于JDK1.8）前言增删改查，修改元素，Vector提供了3个方法，包括迭代器中的一个，不过本文只分析Vector自身的两个修
- 1、mapper.xml文件中的sql语句不提示1.1 首先，alt+enter，出现如下窗口随后的窗口选择这样在如下窗口中会增加一个upd
- GET请求不支持对象传参问题@GetMapping("/getByParam")String hello(Student
- 本文实例为大家分享了android桌面悬浮窗，实现录屏时间控制显示效果的具体代码，供大家参考，具体内容如下悬浮窗效果如上图所示：很简单的一个
- 本文实例为大家分享了java实现砸金蛋抽奖的具体代码，供大家参考，具体内容如下代码如下需求：用户每一次砸金蛋，抽中一等奖的概率为2％ 二等奖
- 文件流输出文件名中文不显示response返回文件流 用response.setHeader(“Content-disp
- 一、实现效果本篇文章实现了简单的图片轮播，初始化3张资源图片，初始化3秒更换一次图片背景，轮换播放。二、知识点Thread线程start()
- 本文实例为大家分享了C#用NPOI导出导入Excel帮助类的具体代码，供大家参考，具体内容如下1、准备工作在管理NuGet程序包中搜索 NP
- 开发环境JDK1.8 eclipse struts2-2.3.31 1.创建web项目 2.导入struts2核心jar包 3.更改web.
- 一.相关知识：Java多线程程序设计到的知识：（一）对同一个数量进行操作（二）对同一个对象进行操作（三）回调方法使用（四）线程同步，死锁问题
- 本文介绍了Spring Boot + MyBatis读写分离，有需要了解Spring+MyBatis读写分离的朋友可参考。希望此文章对各位有
- 最近要做一个java web项目，因为页面不是很多，所以就没有前后端分离，前后端写在一起，这时候就用到thymeleaf了，以下是不动脑式的
- 引言平时使用ProtoStuff作为序列化工具，对于一些POJO对象序列化，但是在实际使用中，发现针对BigDecimal对象进行序列化时却
- 因重定向无法正常goBack()解决方案首先说下问题，初始页面为A，点击某个链接跳转到B（http://xxx.com.cn/）,B页面重定
- 背景：Spring Boot是由Pivotal团队提供的全新框架，其设计目的是用来简化新Spring应用的初始搭建以及开发过程。该框架使用了
