行业知识
Redisson限流算法
2026-11-01 12:34:14
来源：互联网转载
引入依赖
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 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) 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 ;-- 如果：当前请求令牌数 = 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 ; -- 如果 剩余令牌数量 = 当前请求令牌数量，可直接记录签发令牌，并从剩余令牌数量中减去当前签发令牌数量 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才有设置过期时间。
上一篇：自制关卡服务器怎么用的?
