Redisson限流算法

引入依赖

 org.redisson redisson-spring-boot-starter 3.12.3

建议版本使用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();
    }

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

    核心是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]);
    

    发现基于一个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());
        }
    

    核心是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]);
    

    上述参数如下

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

      这个方法是针对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);
          }
      

      我们先看看执行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 ;
      

      注意事项

      trySetRate是基于hsetnx,这意味着一旦设置过Hash中的限流参数,就没法修改。那么如何保证可以修改?

      1,当需要修改时,执行setRate,但最好注意执行时间,因为涉及到zset,string两个key,可能会影响当前的限流窗口。

      2,给限流设置过期时间expire,当到达时间后,自行删除。注意的是:

      expire 要在执行tryAcquire之后执行,才能保证3个key都成功设置过期时间。否则可能只有hash的key才有设置过期时间。