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

---

一、根本使用

1.1 创建限流器

1. /**
2. * Returns rate limiter instance by name
3. *
4. * @param name of rate limiter
5. * @return RateLimiter object
6. */
7. RRateLimiter getRateLimiter(String name);

复制代码

1. /**
2. * Initializes RateLimiter's state and stores config to Redis server.
3. *
4. * @param mode - rate mode
5. * @param rate - rate
6. * @param rateInterval - rate time interval
7. * @param rateIntervalUnit - rate time interval unit
8. * @return true if rate was set and false otherwise
9. */
10. boolean trySetRate(RateType mode, long rate, long rateInterval, RateIntervalUnit rateIntervalUnit);

复制代码

trySetRate 用于设置限流参数。此中 RateType 包罗 OVERALL 和 PER_CLIENT 两个罗列常量，分别表示全局限流和单机限流。后面三个参数表明白令牌的生成速率，即每 rateInterval 生成 rate 个令牌，rateIntervalUnit 为 rateInterval 的时间单位。
1.2 获取令牌

1. /**
2. * Acquires a specified permits from this RateLimiter,
3. * blocking until one is available.
4. *
5. * Acquires the given number of permits, if they are available
6. * and returns immediately, reducing the number of available permits
7. * by the given amount.
8. *
9. * @param permits the number of permits to acquire
10. */
11. void acquire(long permits);
13. /**
14. * Acquires the given number of permits only if all are available
15. * within the given waiting time.
16. *
17. * Acquires the given number of permits, if all are available and returns immediately,
18. * with the value true, reducing the number of available permits by one.
19. *
20. * If no permit is available then the current thread becomes
21. * disabled for thread scheduling purposes and lies dormant until
22. * the specified waiting time elapses.
23. *
24. * If a permits is acquired then the value true is returned.
25. *
26. * If the specified waiting time elapses then the value false
27. * is returned.  If the time is less than or equal to zero, the method
28. * will not wait at all.
29. *
30. * @param permits amount
31. * @param timeout the maximum time to wait for a permit
32. * @param unit the time unit of the timeout argument
33. * @return true if a permit was acquired and false
34. *         if the waiting time elapsed before a permit was acquired
35. */
36. boolean tryAcquire(long permits, long timeout, TimeUnit unit);

复制代码

acquire 和 tryAcquire 均可用于获取指定数量的令牌，不外 acquire 会阻塞等候，而 tryAcquire 会等候 timeout 时间，如果仍然没有获得指定数量的令牌直接返回 false。
1.3 使用示例

1. @Slf4j
2. @SpringBootTest
3. class RateLimiterTest {
4.    
5.     @Autowired
6.     private RedissonClient redissonClient;
7.    
8.     private static final int threadCount = 10;
10.     @Test
11.     void test() throws InterruptedException {
12.         RRateLimiter rateLimiter = redissonClient.getRateLimiter("my_limiter");
13.         rateLimiter.trySetRate(RateType.OVERALL, 10, 1, RateIntervalUnit.SECONDS);
15.         CountDownLatch latch = new CountDownLatch(threadCount);
17.         for (int i = 0; i < threadCount; i++) {
18.             new Thread(() -> {
19.                 rateLimiter.tryAcquire(5, 3, TimeUnit.SECONDS);
20.                 latch.countDown();
21.                 log.info("latch count {}", latch.getCount());
22.             }).start();
23.         }
24.         
25.         latch.await();
26.     }
27. }

复制代码

1. 2024-01-16 20:14:27 INFO  [Thread-2] atreus.ink.rate.RateLimiterTest : latch count 9
2. 2024-01-16 20:14:27 INFO  [Thread-3] atreus.ink.rate.RateLimiterTest : latch count 8
3. 2024-01-16 20:14:28 INFO  [Thread-1] atreus.ink.rate.RateLimiterTest : latch count 7
4. 2024-01-16 20:14:29 INFO  [Thread-10] atreus.ink.rate.RateLimiterTest : latch count 6
5. 2024-01-16 20:14:29 INFO  [Thread-8] atreus.ink.rate.RateLimiterTest : latch count 5
6. 2024-01-16 20:14:30 INFO  [Thread-5] atreus.ink.rate.RateLimiterTest : latch count 4
7. 2024-01-16 20:14:30 INFO  [Thread-4] atreus.ink.rate.RateLimiterTest : latch count 3
8. 2024-01-16 20:14:30 INFO  [Thread-6] atreus.ink.rate.RateLimiterTest : latch count 2
9. 2024-01-16 20:14:30 INFO  [Thread-7] atreus.ink.rate.RateLimiterTest : latch count 1
10. 2024-01-16 20:14:30 INFO  [Thread-9] atreus.ink.rate.RateLimiterTest : latch count 0

复制代码

---

二、实现原理

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

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

RRateLimiter 在创建限流器时通过下面 Lua 脚本设置限流器的相干参数：

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

复制代码

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

1. -- 请求参数示例
2. -- KEYS[1] my_limiter
3. -- KEYS[2] {my_limiter}:value
4. -- KEYS[4] {my_limiter}:permits
5. -- ARGV[1] 3 本次请求的令牌数
6. -- ARGV[2] 1705396021850 System.currentTimeMillis()
7. -- ARGV[3] 6966135962453115904 ThreadLocalRandom.current().nextLong()
9. -- 读取 RRateLimiter.trySetRate 中配置的限流器信息
10. local rate = redis.call('hget', KEYS[1], 'rate');  -- 10 一个时间窗口内产生的令牌数
11. local interval = redis.call('hget', KEYS[1], 'interval');  -- 1000 一个时间窗口对应的毫秒数
12. local type = redis.call('hget', KEYS[1], 'type');  -- 0 全局限流
13. assert(rate ~= false and interval ~= false and type ~= false, 'RateLimiter is not initialized')
15. local valueName = KEYS[2];  -- {my_limiter}:value 当前可用令牌数字符串的 key
16. local permitsName = KEYS[4];  -- {my_limiter}:permits 授权记录有序集合的 key
18. -- 单机限流配置 无需考虑
19. if type == '1' then
20.     valueName = KEYS[3];
21.     permitsName = KEYS[5];
22. end;
24. -- 查询当前可用的令牌数 查询失败表明是首次请求令牌
25. local currentValue = redis.call('get', valueName);
26. if currentValue == false then -- 首次请求令牌
27.     -- 单次请求的令牌数不能超过一个时间窗口内产生的令牌数
28.     assert(tonumber(rate) >= tonumber(ARGV[1]), 'Requested permits amount could not exceed defined rate');
29.    
30.     -- 更新当前可用令牌数以及令牌授权记录 {my_limiter}:permits
31.     -- set {my_limiter}:permits 10
32.     redis.call('set', valueName, rate);
33.     -- zadd {my_limiter}:permits 1705396021850 6966135962453115904_1
34.     redis.call('zadd', permitsName, ARGV[2], struct.pack('fI', ARGV[3], ARGV[1]));
35.     -- decrby {my_limiter}:permits 3
36.     redis.call('decrby', valueName, ARGV[1]);
37.     return nil;
38. else -- 再次请求令牌
39.     -- 查询可以回收的令牌对应的授权记录 即一个时间窗口前的所有授权记录且包括一个时间窗口前这一时刻
40.     -- 旧令牌回收的本质是新令牌的加入 如果一个令牌是在一个时间窗口前被分配的 那经过一个时间窗口后这个空出的位置应该已经由新令牌填充
41.     -- zrangebyscore {my_limiter}:permits 0 1705396020850
42.     local expiredValues = redis.call('zrangebyscore', permitsName, 0, tonumber(ARGV[2]) - interval); -- [1936135962853113704_2, 536135765023123704_5]
43.    
44.     -- 统计可以回收的令牌数
45.     local released = 0;
46.     for i, v in ipairs(expiredValues) do
47.         local random, permits = struct.unpack('fI', v);
48.         -- released = released + 2
49.         -- released = released + 5
50.         released = released + permits;
51.     end;
53.     -- 删除授权记录并回收令牌
54.     if released > 0 then
55.         -- zrem {my_limiter}:permits 1936135962853113704_2 536135765023123704_5
56.         redis.call('zrem', permitsName, unpack(expiredValues));
57.         currentValue = tonumber(currentValue) + released;
58.         -- incrby {my_limiter}:value 7
59.         redis.call('set', valueName, currentValue);
60.     end;
62.     if tonumber(currentValue) < tonumber(ARGV[1]) then
63.         -- 如果回收后可用令牌数仍然不足 返回需要等待的时间
64.         -- zrangebyscore {my_limiter}:permits (1705396020850 1705396021850 withscores limit 0 1
65.         local nearest = redis.call('zrangebyscore', permitsName, '(' .. (tonumber(ARGV[2]) - interval), tonumber(ARGV[2]), 'withscores', 'limit', 0, 1);
66.         local random, permits = struct.unpack('fI', nearest[1]);
67.         -- 1705396021650 - 1705396021850 + 1000 = 800
68.         return tonumber(nearest[2]) - (tonumber(ARGV[2]) - interval);
69.     else
70.         
71.         redis.call('zadd', permitsName, ARGV[2], struct.pack('fI', ARGV[3], ARGV[1]));
72.         redis.call('decrby', valueName, ARGV[1]);
73.         return nil;
74.     end;
75. end;

复制代码

---

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

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。