继续创造，加快成长！这是我参与「日新计划 10 月更文应战」的第22天，点击查看活动概况

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

接前面聊一聊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

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" 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
    -- 第2次获取令牌履行
    -------------------------- 获取zset结构：计算之前的恳求令牌数
    -- 范围是0 ~ (第2次恳求时刻戳 - 令牌生产的时刻)
    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。~~

2022-11-4 更新

运用集群衔接redisson，一个周期完毕后，没有在生成新的限流器。

及时当勉励岁月不待人

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

非常感谢

菁英们能看到这儿，假如这个文章写得还不错，觉得有点东西的话求点赞求关注❤️ 求共享对需要继续更新的我来说真的非常有用！！

假如本篇博客有任何错误，请批评指教，不胜感激！

文末福利，最近整理一份面试资料《Java面试通关手册》，覆盖了Java核心技术、JVM、Java并发、SSM、微服务、数据库、数据结构等等。获取方式：【别卷了大兄弟】

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

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

1.简略运用

2. 完成限流redisson运用了哪些redis数据结构

3. 超越10s,我再次获取一个令牌，数据结构发生的改变

4. 源码浅析

作者信息

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

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

1.简略运用

2. 完成限流redisson运用了哪些redis数据结构

3. 超越10s,我再次获取一个令牌，数据结构发生的改变

4. 源码浅析

相关文章

讲透JAVA Stream的collect用法与原理，远比你想象的更强大

远古的火箭追着我跑，跑过了2022走进了2023

KVC 研究

【每日鲜蘑】Linux 服务器配置调优

作者信息