IT俱乐部 Java redisson分布式限流RRateLimiter源码解析

redisson分布式限流RRateLimiter源码解析

发布: 2023年 2月 12日 530阅读

分布式限流-单位时间多实例多线程访问次数限制

接前面聊一聊redisson及优雅实现 和 说一说spring boot优雅集成redisson，简单以源码的方式给大家介绍了redisson的：可重入性、阻塞、续约、红锁、联锁、加锁解锁流程和集成spring boot注意点和优雅实现方式。

接下来在讲一讲平时用的比较多的限流模块–RRateLimiter

1.简单使用

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
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. 源码浅析

1
2
3
4
RRateLimiter rateLimiter = redisson.getRateLimiter("myRateLimiter3");
// 初始化
// 最大流速 = 每10秒钟产生3个令牌
rateLimiter.trySetRate(RateType.PER_CLIENT, 3, 10, RateIntervalUnit.SECONDS);

初始化定义没有什么好讲的，就是创建HASH结构

主要还是讲讲： rateLimiter.tryAcquire()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
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，请大家用最新版本，以下为修复后解析。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
-- 获取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代码也并不复杂，令牌桶的数量主要是通过时间窗口来控制，判断上一个请求是否超过了令牌生产周期。

留下一个疑问？

1
2
-- 移除zset中所有元素【要求是同一个限流器permitsName，不然就移除不了，尴尬】
redis.call("zrem", permitsName, unpack(expiredValues))

我自己在本地测试，只要超过10s，permitsName就不一样，这就导致了这部分数据是不能移除的，就产生了冗余数据，从前面的截图也可以看出，是新生成了一个zset数据结构。

相当于直接走到了这一步：

1
2
------------------------- 更新zset结构
redis.call("zadd", permitsName, ARGV[2], struct.pack("fI", ARGV[3], ARGV[1]))

至于为什么会产生这样的结果，会的小伙伴可以留言，或者过段时间我提个issue。

以上就是redisson分布式限流RRateLimiter源码解析的详细内容，更多关于redisson分布式限流RRateLimiter的资料请关注IT俱乐部其它相关文章！

本文收集自网络，不代表IT俱乐部立场，转载请注明出处。https://www.2it.club/code/java/5314.html
50赞
标签:
上一篇
下一篇

相关推荐

联系我们

联系我们

在线咨询: QQ交谈

邮箱: 1120393934@qq.com

工作时间：周一至周五，9:00-17:30，节假日休息

关注微信
微信扫一扫关注我们

微信扫一扫关注我们

返回顶部