redisson分布式限流RRateLimiter源码解析

 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;
    }

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

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()
                    }
                );
}

-- 获取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

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

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