分布式限流-单位时间多实例多线程访问次数限制
接前面聊一聊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 { 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 RFuture tryAcquireAsync(RedisCommand 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]), '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)
这段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。
以上就是redisson分布式限流RRateLimiter源码解析的详细内容,更多关于redisson分布式限流RRateLimiter的资料请关注IT俱乐部其它相关文章!