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

文章目录

一、基本使用

1.1 创建限流器

/**
 * Returns rate limiter instance by name
 * 
 * @param name of rate limiter
 * @return RateLimiter object
 */RRateLimitergetRateLimiter(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
 */booleantrySetRate(RateType mode,long rate,long rateInterval,RateIntervalUnit rateIntervalUnit);

trySetRate

用于设置限流参数。其中 RateType 包含

OVERALL

和

PER_CLIENT

两个枚举常量，分别表示全局限流和单机限流。后面三个参数表明了令牌的生成速率，即每

rateInterval

生成

rate

个令牌，

rateIntervalUnit

为

rateInterval

的时间单位。

1.2 获取令牌

/**
 * 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
 */voidacquire(longpermits);/**
 * 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
 */booleantryAcquire(longpermits,long timeout,TimeUnit unit);

acquire

和

tryAcquire

均可用于获取指定数量的令牌，不过

acquire

会阻塞等待，而

tryAcquire

会等待

timeout

时间，如果仍然没有获得指定数量的令牌直接返回

false

。

1.3 使用示例

@Slf4j@SpringBootTestclassRateLimiterTest{@AutowiredprivateRedissonClient redissonClient;privatestaticfinalint threadCount =10;@Testvoidtest()throwsInterruptedException{RRateLimiter rateLimiter = redissonClient.getRateLimiter("my_limiter");
        rateLimiter.trySetRate(RateType.OVERALL,10,1,RateIntervalUnit.SECONDS);CountDownLatch latch =newCountDownLatch(threadCount);for(int i =0; i < threadCount; i++){newThread(()->{
                rateLimiter.tryAcquire(5,3,TimeUnit.SECONDS);
                latch.countDown();
                log.info("latch count {}", latch.getCount());}).start();}
        
        latch.await();}}

2024-01-1620:14:27INFO[Thread-2]atreus.ink.rate.RateLimiterTest: latch count 92024-01-1620:14:27INFO[Thread-3]atreus.ink.rate.RateLimiterTest: latch count 82024-01-1620:14:28INFO[Thread-1]atreus.ink.rate.RateLimiterTest: latch count 72024-01-1620:14:29INFO[Thread-10]atreus.ink.rate.RateLimiterTest: latch count 62024-01-1620:14:29INFO[Thread-8]atreus.ink.rate.RateLimiterTest: latch count 52024-01-1620:14:30INFO[Thread-5]atreus.ink.rate.RateLimiterTest: latch count 42024-01-1620:14:30INFO[Thread-4]atreus.ink.rate.RateLimiterTest: latch count 32024-01-1620:14:30INFO[Thread-6]atreus.ink.rate.RateLimiterTest: latch count 22024-01-1620:14:30INFO[Thread-7]atreus.ink.rate.RateLimiterTest: latch count 12024-01-1620:14:30INFO[Thread-9]atreus.ink.rate.RateLimiterTest: latch count 0

二、实现原理

Redisson 的 RRateLimiter 基于令牌桶实现，令牌桶的主要特点如下：

• 令牌以固定速率生成。
• 生成的令牌放入令牌桶中存放，如果令牌桶满了则多余的令牌会直接丢弃，当请求到达时，会尝试从令牌桶中取令牌，取到了令牌的请求可以执行。
• 如果桶空了，那么尝试取令牌的请求会被直接丢弃。

RRateLimiter 在创建限流器时通过下面 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]);

而获取令牌则是通过以下的 Lua 脚本实现：

-- 请求参数示例-- 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 ~=falseand interval ~=falseand type ~=false,'RateLimiter is not initialized')local valueName = KEYS[2];-- {my_limiter}:value 当前可用令牌数字符串的 keylocal 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 ==falsethen-- 首次请求令牌-- 单次请求的令牌数不能超过一个时间窗口内产生的令牌数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]);returnnil;else-- 再次请求令牌-- 查询可以回收的令牌对应的授权记录 即一个时间窗口前的所有授权记录且包括一个时间窗口前这一时刻-- 旧令牌回收的本质是新令牌的加入 如果一个令牌是在一个时间窗口前被分配的 那经过一个时间窗口后这个空出的位置应该已经由新令牌填充-- zrangebyscore {my_limiter}:permits 0 1705396020850local expiredValues = redis.call('zrangebyscore', permitsName,0,tonumber(ARGV[2])- interval);-- [1936135962853113704_2, 536135765023123704_5]-- 统计可以回收的令牌数local released =0;for i, v inipairs(expiredValues)dolocal random, permits = struct.unpack('fI', v);-- released = released + 2-- released = released + 5
        released = released + permits;end;-- 删除授权记录并回收令牌if released >0then-- 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;iftonumber(currentValue)<tonumber(ARGV[1])then-- 如果回收后可用令牌数仍然不足 返回需要等待的时间-- zrangebyscore {my_limiter}:permits (1705396020850 1705396021850 withscores limit 0 1local 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 = 800returntonumber(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]);returnnil;end;end;

参考：

https://github.com/oneone1995/blog/issues/13
https://www.infoq.cn/article/Qg2tX8fyw5Vt-f3HH673