redisson分布式限流[RRateLimiter]源码分析

接下来在讲一讲平时用的比较多的限流模块–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

3. 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" thenvalueName = 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-- 获取请求数permitslocal random, permits = struct.unpack("fI", v)released = released + permitsend-- 之前的请求令牌数 > 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 nilend
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。

及时当勉励岁月不待人

能看到这里的人呀，都是菁英。❤❤️❤️❤️❤

非常感谢

菁英们能看到这里，如果这个文章写得还不错，觉得有点东西的话求点赞

redisson分布式限流[RRateLimiter]源码分析相关推荐

alibaba sentinel限流组件源码分析
如何使用? maven引入: <dependency> <groupId>com.alibaba.csp</groupId> <artifactId>s ...
阿里开源一站式分布式事务框架seata源码分析(AT模式下TM与RM分析)
序言: 对于阿里开源分布式事务框架seata的详细了解可以参考官网,这里不会详细介绍.本章只会介绍seata中AT模式的源码分析(对阿seata有一定了解或者成功完成过demo). seata中一个事 ...
orleans 限流策略源码剖析
orleans简介:Orleans is a cross-platform framework for building robust, scalable distributed applicatio ...
擎创技术流 | Flink源码分析-JobDispatcher
背景介绍最近一直在阅读Flink基于Yarn的资源管理相关的代码,牵扯的流程比较长,主要包含以下几个环节: 客户端环节:命令参数解析,定位到作业入口,生成JobGraph,翻译成启动对应的Yarn集 ...
sentinel限流相关指标统计源码分析
文章目录 *前言* *官方架构图* 滑动窗口模型 *StatisticSlot* *Node* *NodeSelectorSlot* *ClusterBuilderSlot* *FlowSlot如何使 ...
redisson之RRateLimiter分布式限流
目录一.说明二.代码实现一.说明在redis lua+令牌桶算法实现分布式限流这篇文章有介绍,自己通过Lua脚本来实现分布式限流.如果是集成的Jedis等下,是可以采用这篇文章的方法自我实现. ...
JAVA中useDrlimiter方法_今天来讲讲分布式环境下，怎么达到对象共享，以及实现原子性（atomic），以Redis中的Redisson为例（实现分布式锁、分布式限流等）...
相信各位对redis肯定是不陌生的,一个高吞吐量的内存型结构存储数据库.可用用于很多业务场景,能够有效的解决很多复杂的并发问题,分布式问题. 下面粘一下中文官网介绍: 关于解决对象共享问题,很多方式, ...
精尽 Redisson 源码分析 —— 限流器 RateLimiter
1. 概述限流,无论在系统层面,还是在业务层面,使用都非常广泛.例如说: [业务]为了避免恶意的灌水机或者用户,限制每分钟至允许回复 10 个帖子. [系统]为了避免服务系统被大规模调用,超过极限, ...
Redis分布式锁解析源码分析
Redis分布式锁解析&源码分析概述实战简单的分布式锁 Redisson实现分布式锁 Redission源码分析构造方法获取锁lock 解锁锁失效红锁案例分析原始的写法进化 ...