• Redisson限流算法

    引入依赖

    
            org.redisson
            redisson-spring-boot-starter
            3.12.3


    • 1
    • 2
    • 3
    • 4
    • 5

    建议版本使用3.15.5以上

    使用

    这边写了一个demo示例，定义了一个叫 “LIMITER_NAME” 的限流器，设置每1秒钟生成3个令牌。

     public static void main(String[] args) throws UnsupportedEncodingException, InterruptedException {
        Config config = new Config();
        config.useSingleServer().setAddress("redis://127.0.0.1:6379");
        RedissonClient redisson = Redisson.create(config);

        String key = "test:rateLimiter01";
        RRateLimiter rateLimiter = redisson.getRateLimiter(key);
        boolean trySetRate = rateLimiter.trySetRate(RateType.OVERALL,  3, 1, RateIntervalUnit.SECONDS);
        for (int i = 0; i < 30; i++) {
            boolean b = rateLimiter.tryAcquire(1,100, TimeUnit.MILLISECONDS);
            System.out.println(Thread.currentThread().getName() + " testRate第" + (i + 1) + "次：" + b);
        }

//        redisson.shutdown();
    }

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15

    在这里插入图片描述

    redis的数据结构

    PRateLimiter接口的实现类几乎都在RedissonRateLimiter上，我们看看前面调用PRateLimiter方法时，这些方法的对应源码实现。

    接下来下面就着重讲讲限流算法中，一共用到的3个redis key。

    key1 Hash结构

    就是前面trySetRate设置的hash key。按照之前限流器命名“LIMITER_NAME”，这个名字就是LIMITER_NAME。一共有3个值。

    1，rate：代表速率
    2，interval：代表多少时间内产生的令牌
    3，type：代表时单机还是集群。
    在这里插入图片描述

    key 2： Zset结构

    zset记录获取令牌的时间戳，用于时间对比，redis key的名字是{LIMITER_NAME}:permits。下面讲讲zset中每个元素的member和score

    • member：包含两个内容，
      1）一段8位随机字符串，为了唯一标志性当次获取令牌
      2）数字，即当次获取令牌的数量。不过这些都是压缩后存储在redis中的，在工具上看时会发现乱码。
    • score：记录获取令牌的时间戳，eg：1709026371728 转成日期是2024-02-27 17:32:51
      在这里插入图片描述

    key 3 string 结构

    记录当前令牌桶中剩余的令牌数。redis key的名字是{LIMITER_NAME}:value。
    在这里插入图片描述

    算法源码分析

    trySetRate尝试设置

    尝试设置是：当没有对应的key的时候设置，如果已经有值了，就不做任何处理。对应实现类中的源码是：

        /**
     * 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 {@code true} if rate was set and {@code false}
     *         otherwise
     */
    boolean trySetRate(RateType mode, long rate, long rateInterval, RateIntervalUnit rateIntervalUnit);


 @Override
    public boolean trySetRate(RateType type, long rate, long rateInterval, RateIntervalUnit unit) {
        return get(trySetRateAsync(type, rate, rateInterval, unit));
    }

    @Override
    public RFuture trySetRateAsync(RateType type, long rate, long rateInterval, RateIntervalUnit unit) {
        return commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,
                "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]);",
                Collections.singletonList(getName()), rate, unit.toMillis(rateInterval), type.ordinal());
    }

    • 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

    核心是lua脚本，摘出来如下：

    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]);

    • 1
    • 2
    • 3

    发现基于一个hash类型的redis key设置了3个值。
    不过这里的命令是hsetnx，redis hsetnx命令用于哈希表中不存在的字段赋值。
    如果哈希表不存在，一个新的哈希表被创建并进行hset操作。
    如果字段已经存在hash表中，操作无效。
    如果key不存在，一个新的哈希表被创建并执行hsetnx命令。

    这意味着，这个方法只能做配置初始化，如果后期想要修改配置参数，该方法并不会生效。我们来看看另外一个方法。

    setRete重新设置

    重新设置是，不管该key之前有没有用，一切清空回到初始化，重新设置。对应类中实现类的源码是。

        /**
     * Updates 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
     */
    void setRate(RateType mode, long rate, long rateInterval, RateIntervalUnit rateIntervalUnit);


    public RFuture setRateAsync(RateType type, long rate, long rateInterval, RateIntervalUnit unit) {
         return commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,
                "redis.call('hset', KEYS[1], 'rate', ARGV[1]);"
                        + "redis.call('hset', KEYS[1], 'interval', ARGV[2]);"
                        + "redis.call('hset', KEYS[1], 'type', ARGV[3]);"
                        + "redis.call('del', KEYS[2], KEYS[3]);",
                Arrays.asList(getName(), getValueName(), getPermitsName()), rate, unit.toMillis(rateInterval), type.ordinal());
    }

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19

    核心是lua脚本

    redis.call('hset', KEYS[1], 'rate', ARGV[1]);
redis.call('hset', KEYS[1], 'interval', ARGV[2]);
redis.call('hset', KEYS[1], 'type', ARGV[3]);
redis.call('del', KEYS[2], KEYS[3]);

    • 1
    • 2
    • 3
    • 4

    上述参数如下

    • KEYS[1]：hash key name
    • KEYS[2]：string(value) key name
    • KEYS[3]：zset(permits) key name
    • ARGV[1]：rate
    • ARGV[2]：interval
    • ARGV[3]：type
      通过这个lua的逻辑，就能看出直接用的是hset，会直接重置配置参数，并且同时会将已产生数据的string(value)、zset(ppermis)两个key删掉。是一个彻底的重置方法。

    这里回顾一下trySetRate和setRate（注意setRate在3.12.3这个版本是没有这个方法的），在限流器不变的场景下，我们可以多次调用trySetRate，但是不能调用setRate。因为每调用一次，redis.call(‘del’,keys[2],keys[3])就会将限流器中数据清空，也就达不到限流功能。

    设置过期时间

    有没有发现前面针对限流器设置的3个key，都没有设置过期时间。PRateLimiter接口设计上，将设置过期时间单独拎出来了。

     // 设置过期
    boolean expire(long var1, TimeUnit var3);
    // 清除过期（永不过期）
    boolean clearExpire();

    • 1
    • 2
    • 3
    • 4

    这个方法是针对3个key一起设置过期时间。

    获取令牌（核心）tryAcquire

    获取令牌

     private  RFuture tryAcquireAsync(RedisCommand command, Long value) {
        byte[] random = getServiceManager().generateIdArray();

        return commandExecutor.evalWriteAsync(getRawName(), 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;"

              + "assert(tonumber(rate) >= tonumber(ARGV[1]), 'Requested permits amount could not exceed defined rate'); "

              + "local currentValue = redis.call('get', valueName); "
              + "local res;"
              + "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('Bc0I', v);"
                          + "released = released + permits;"
                     + "end; "

                     + "if released > 0 then "
                          + "redis.call('zremrangebyscore', permitsName, 0, tonumber(ARGV[2]) - interval); "
                          + "if tonumber(currentValue) + released > tonumber(rate) then "
                               + "currentValue = tonumber(rate) - redis.call('zcard', permitsName); "
                          + "else "
                               + "currentValue = tonumber(currentValue) + released; "
                          + "end; "
                          + "redis.call('set', valueName, currentValue);"
                     + "end;"

                     + "if tonumber(currentValue) < tonumber(ARGV[1]) then "
                         + "local firstValue = redis.call('zrange', permitsName, 0, 0, 'withscores'); "
                         + "res = 3 + interval - (tonumber(ARGV[2]) - tonumber(firstValue[2]));"
                     + "else "
                         + "redis.call('zadd', permitsName, ARGV[2], struct.pack('Bc0I', string.len(ARGV[3]), ARGV[3], ARGV[1])); "
                         + "redis.call('decrby', valueName, ARGV[1]); "
                         + "res = nil; "
                     + "end; "
              + "else "
                     + "redis.call('set', valueName, rate); "
                     + "redis.call('zadd', permitsName, ARGV[2], struct.pack('Bc0I', string.len(ARGV[3]), ARGV[3], ARGV[1])); "
                     + "redis.call('decrby', valueName, ARGV[1]); "
                     + "res = nil; "
              + "end;"

              + "local ttl = redis.call('pttl', KEYS[1]); "
              + "if ttl > 0 then "
                  + "redis.call('pexpire', valueName, ttl); "
                  + "redis.call('pexpire', permitsName, ttl); "
              + "end; "
              + "return res;",
                Arrays.asList(getRawName(), getValueName(), getClientValueName(), getPermitsName(), getClientPermitsName()),
                value, System.currentTimeMillis(), random);
    }

    • 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
    • 49
    • 50
    • 51
    • 52
    • 53
    • 54
    • 55
    • 56
    • 57
    • 58
    • 59
    • 60
    • 61
    • 62

    我们先看看执行lua脚本时，所有要传入的参数内容：
    KEYS[1]: hash key name
    KEYS[2]:全局string(value) key name
    KEYS[3]:单机string(value) key name
    KEYS[4]:全局zset(permits) key name
    KEYS[5]:单机zset(permits) key name
    ARGV[1]:当前请求令牌数量
    ARGV[2]:当前时间
    ARGV[3]:8位随机字符串
    然后我们再将其中lua部分提取出来，我再根据自己的理解

    -- rate：间隔时间内产生令牌数量
-- interval：间隔时间
-- type：类型：0-全局限流；1-单机限
local rate = redis.call('hget', KEYS[1], 'rate');
local interval = redis.call('hget', KEYS[1], 'interval');
local type = redis.call('hget', KEYS[1], 'type');
-- 如果3个参数存在空值，错误提示初始化未完成
assert(rate ~= false and interval ~= false and type ~= false, 'RateLimiter is not initialized')
local valueName = KEYS[2];
local permitsName = KEYS[4];
-- 如果是单机限流，在全局key后拼接上机器唯一标识字符
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');
-- currentValue = 当前剩余令牌数量
local currentValue = redis.call('get', valueName);
-- 非第一次访问，存储剩余令牌数量的 string(value) key 存在，有值（包括 0）
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
        -- lua struct的pack/unpack方法，可以理解为文本压缩/解压缩方法
        local random, permits = struct.unpack('fI', v);
        released = released + permits;
    end ;
    -- 移除 zset(permits) 中过期的令牌签发记录
    -- 将过期令牌回收回来，重新更新剩余令牌数量
    if released > 0 then
        redis.call('zremrangebyscore', permitsName, 0, tonumber(ARGV[2]) - interval);
        currentValue = tonumber(currentValue) + released;
        redis.call('set', valueName, currentValue);
    end ;
    -- 如果 剩余令牌数量 < 当前请求令牌数量，返回推测可以获得所需令牌数量的时间
    -- （1）最近一次签发令牌的释放时间 = 最近一次签发令牌的签发时间戳 + 间隔时间(interval)
    -- （2）推测可获得所需令牌数量的时间 = 最近一次签发令牌的释放时间 - 当前时间戳
    -- （3）"推测"可获得所需令牌数量的时间，"推测"，是因为不确定最近一次签发令牌数量释放后，加上到时候的剩余令牌数量，是否满足所需令牌数量
    if tonumber(currentValue) < tonumber(ARGV[1]) then
        local nearest = redis.call('zrangebyscore', permitsName, '(' .. (tonumber(ARGV[2]) - interval), '+inf', 'withscores', 'limit', 0, 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 ;
    -- 第一次访问，存储剩余令牌数量的 string(value) key 不存在，为 null，走初始化逻辑
else
    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 ;

    • 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
    • 49
    • 50
    • 51
    • 52
    • 53
    • 54
    • 55
    • 56
    • 57

    注意事项

    trySetRate是基于hsetnx，这意味着一旦设置过Hash中的限流参数，就没法修改。那么如何保证可以修改？
    1，当需要修改时，执行setRate，但最好注意执行时间，因为涉及到zset，string两个key，可能会影响当前的限流窗口。
    2，给限流设置过期时间expire，当到达时间后，自行删除。注意的是：
    expire 要在执行tryAcquire之后执行，才能保证3个key都成功设置过期时间。否则可能只有hash的key才有设置过期时间。

  • 相关阅读:
    262-视口，布局视口，视觉视口，移动端适配，less语法，比哪里，DPR,RRI，less的弊端，运算，嵌套，混合，继承，混入，运算，
    python脚本合并多个pdf文件
    小啊呜产品读书笔记001：《邱岳的产品手记-11》第21讲 产品案例分析：Fabulous的精致养成
    Util应用框架基础(三) - 面向切面编程(AspectCore AOP)
    SpringBoot注解
    Docker概念通讲
    由于.git/config导致的Git存储库泄露
    JAVA实现Jfilechooser搜索功能
    静态博客搭建工具汇总
    asp.net特色商品购物网站系统VS开发sqlserver数据库web结构c#编程Microsoft Visual Studio
  • 原文地址：https://blog.csdn.net/weixin_30409927/article/details/136327750