redisson分布式限流RRateLimiter怎么使用

今天小編給大家分享一下redisson分布式限流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]))

以上就是“redisson分布式限流RRateLimiter怎么使用”這篇文章的所有內容，感謝各位的閱讀！相信大家閱讀完這篇文章都有很大的收獲，小編每天都會為大家更新不同的知識，如果還想學習更多的知識，請關注億速云行業資訊頻道。