Redisson分布式限流器RRateLimiter的使用及原理小结

/**
 * returns rate limiter instance by name
 * 
 * @param name of rate limiter
 * @return ratelimiter object
 */
rratelimiter getratelimiter(string name);

/**
 * initializes ratelimiter's state and stores config to redis server.
 * 
 * @param mode - rate mode
 * @param rate - rate
 * @param rateinterval - rate time interval
 * @param rateintervalunit - rate time interval unit
 * @return true if rate was set and false otherwise
 */
boolean trysetrate(ratetype mode, long rate, long rateinterval, rateintervalunit rateintervalunit);

/**
 * acquires a specified permits from this ratelimiter, 
 * blocking until one is available.
 *
 * acquires the given number of permits, if they are available 
 * and returns immediately, reducing the number of available permits 
 * by the given amount.
 * 
 * @param permits the number of permits to acquire
 */
void acquire(long permits);

/**
 * acquires the given number of permits only if all are available
 * within the given waiting time.
 *
 * acquires the given number of permits, if all are available and returns immediately,
 * with the value true, reducing the number of available permits by one.
 *
 * if no permit is available then the current thread becomes
 * disabled for thread scheduling purposes and lies dormant until
 * the specified waiting time elapses.
 *
 * if a permits is acquired then the value true is returned.
 *
 * if the specified waiting time elapses then the value false
 * is returned.  if the time is less than or equal to zero, the method
 * will not wait at all.
 *
 * @param permits amount
 * @param timeout the maximum time to wait for a permit
 * @param unit the time unit of the timeout argument
 * @return true if a permit was acquired and false
 *         if the waiting time elapsed before a permit was acquired
 */
boolean tryacquire(long permits, long timeout, timeunit unit);

@slf4j
@springboottest
class ratelimitertest {
    
    @autowired
    private redissonclient redissonclient;
    
    private static final int threadcount = 10;

    @test
    void test() throws interruptedexception {
        rratelimiter ratelimiter = redissonclient.getratelimiter("my_limiter");
        ratelimiter.trysetrate(ratetype.overall, 10, 1, rateintervalunit.seconds);

        countdownlatch latch = new countdownlatch(threadcount);

        for (int i = 0; i < threadcount; i++) {
            new thread(() -> {
                ratelimiter.tryacquire(5, 3, timeunit.seconds);
                latch.countdown();
                log.info("latch count {}", latch.getcount());
            }).start();
        }
        
        latch.await();
    }
}

redis.call('hsetnx', keys[1], 'rate', argv[1]);
redis.call('hsetnx', keys[1], 'interval', argv[2]);
return redis.call('hsetnx', keys[1], 'type', argv[3]);

-- 请求参数示例
-- keys[1] my_limiter
-- keys[2] {my_limiter}:value
-- keys[4] {my_limiter}:permits
-- argv[1] 3 本次请求的令牌数
-- argv[2] 1705396021850 system.currenttimemillis()
-- argv[3] 6966135962453115904 threadlocalrandom.current().nextlong()

-- 读取 rratelimiter.trysetrate 中配置的限流器信息
local rate = redis.call('hget', keys[1], 'rate');  -- 10 一个时间窗口内产生的令牌数
local interval = redis.call('hget', keys[1], 'interval');  -- 1000 一个时间窗口对应的毫秒数
local type = redis.call('hget', keys[1], 'type');  -- 0 全局限流
assert(rate ~= false and interval ~= false and type ~= false, 'ratelimiter is not initialized')

local valuename = keys[2];  -- {my_limiter}:value 当前可用令牌数字符串的 key
local permitsname = keys[4];  -- {my_limiter}:permits 授权记录有序集合的 key

-- 单机限流配置 无需考虑
if type == '1' then
    valuename = keys[3];
    permitsname = keys[5];
end;

-- 查询当前可用的令牌数 查询失败表明是首次请求令牌
local currentvalue = redis.call('get', valuename);
if currentvalue == false then -- 首次请求令牌
    -- 单次请求的令牌数不能超过一个时间窗口内产生的令牌数
    assert(tonumber(rate) >= tonumber(argv[1]), 'requested permits amount could not exceed defined rate');
    
    -- 更新当前可用令牌数以及令牌授权记录 {my_limiter}:permits
    -- set {my_limiter}:permits 10
    redis.call('set', valuename, rate);
    -- zadd {my_limiter}:permits 1705396021850 6966135962453115904_1
    redis.call('zadd', permitsname, argv[2], struct.pack('fi', argv[3], argv[1]));
    -- decrby {my_limiter}:permits 3
    redis.call('decrby', valuename, argv[1]);
    return nil;
else -- 再次请求令牌
    -- 查询可以回收的令牌对应的授权记录 即一个时间窗口前的所有授权记录且包括一个时间窗口前这一时刻
    -- 旧令牌回收的本质是新令牌的加入 如果一个令牌是在一个时间窗口前被分配的 那经过一个时间窗口后这个空出的位置应该已经由新令牌填充
    -- zrangebyscore {my_limiter}:permits 0 1705396020850
    local expiredvalues = redis.call('zrangebyscore', permitsname, 0, tonumber(argv[2]) - interval); -- [1936135962853113704_2, 536135765023123704_5]
    
    -- 统计可以回收的令牌数
    local released = 0;
    for i, v in ipairs(expiredvalues) do
        local random, permits = struct.unpack('fi', v);
        -- released = released + 2
        -- released = released + 5
        released = released + permits;
    end;

    -- 删除授权记录并回收令牌
    if released > 0 then
        -- zrem {my_limiter}:permits 1936135962853113704_2 536135765023123704_5
        redis.call('zrem', permitsname, unpack(expiredvalues));
        currentvalue = tonumber(currentvalue) + released;
        -- incrby {my_limiter}:value 7
        redis.call('set', valuename, currentvalue);
    end;

    if tonumber(currentvalue) < tonumber(argv[1]) then
        -- 如果回收后可用令牌数仍然不足 返回需要等待的时间
        -- zrangebyscore {my_limiter}:permits (1705396020850 1705396021850 withscores limit 0 1
        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]);
        -- 1705396021650 - 1705396021850 + 1000 = 800
        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;
end;