详解Redisson分布式限流的使用及原理

<dependency>
    <groupid>org.redisson</groupid>
    <artifactid>redisson-spring-boot-starter</artifactid>
    <version>3.43.0</version>
</dependency>

import lombok.data;
import org.redisson.api.rateintervalunit;
import org.springframework.boot.context.properties.configurationproperties;
import org.springframework.stereotype.component;

import java.util.list;

@component
@configurationproperties("rate-limit")
@data
public class ratelimitproperties {
    private config defaultconfig;
    private list<config> configs;

    @data
    public static class config {
        private string keyprefix;
        private long timeout;
        private rateintervalunit rateintervalunit;
        private long count;
    }

    /**
     * 根据key获取对应匹配的前缀的限流配置
     * @param key key
     * @return 限流配置
     */
    public config getratelimitconfig(string key) {
        return configs.stream()
                .filter(e -> key.startswith(e.getkeyprefix()))
                .findfirst()
                .orelse(defaultconfig);
    }

}

# 限流配置
rate-limit:
  # 默认限流配置，如果前缀没有匹配上，则使用默认配置
  default-config:
    model-name-prefix: default
    time-out: 30
    rate-interval-unit: seconds
    count: 1
  configs:
    - key-prefix: test
      time-out: 1
      rate-interval-unit: seconds
      count: 2

import com.gzb.app.config.ratelimitproperties;
import com.gzb.app.constant.redisconstant;
import org.redisson.api.*;
import org.springframework.beans.factory.annotation.autowired;
import org.springframework.stereotype.component;

import java.util.concurrent.timeunit;

/**
 * 限流工具类
 */
@component
public class ratelimitutil {
    @autowired
    private ratelimitproperties ratelimitproperties;
    @autowired
    private redissonclient redissonclient;
    
    /**
     * 尝试获取一个限流器许可，如果当前没有足够的令牌，调用线程将被阻塞，直到有足够的令牌
     * @param key 限流器的唯一标识
     */
    public void acquire(string key) {
        getredissonratelimiter(key).acquire();
    }

    /**
     * 尝试立即获取一个限流器许可，如果当前没有足够的令牌，立即返回 false
     * @param key 限流器的唯一标识
     * @return 成功：true 失败：false
     */
    public boolean tryacquire(string key) {
        return getredissonratelimiter(key).tryacquire();
    }

    /**
     * 尝试在指定时间内获取一个限流器许可，如果在指定时间内没有足够的令牌，立即返回 false
     * @param key 限流器的唯一标识
     * @param time 等待的时间长度
     * @param timeunit 时间单位
     * @return 成功：true 失败：false
     */
    public boolean tryacquire(string key, long time, timeunit timeunit) {
        return getredissonratelimiter(key).tryacquire(time, timeunit);
    }

    /**
     * 获取限流器，如果限流器不存在，则根据配置创建一个新的限流器
     * @param key 限流器的唯一标识
     * @return redissonratelimiter 限流器实例
     */
    public rratelimiter getredissonratelimiter(string key) {
        ratelimitproperties.config config = ratelimitproperties.getratelimitconfig(key);
        long count = config.getcount();
        long timeout = config.gettimeout();
        rateintervalunit rateintervalunit = config.getrateintervalunit();
        rratelimiter ratelimiter = redissonclient.getratelimiter(key);
        // 如果限流器不存在，就创建一个限流器
        if (!ratelimiter.isexists()) {
            ratelimiter.trysetrate(ratetype.overall, count, timeout, rateintervalunit);
            return ratelimiter;
        }
        checkifconfigneedupdate(ratelimiter, config);
        return ratelimiter;
    }

    /**
     * 检查限流器配置是否发生修改，如果配置发生变化，则删除旧的限流器并重新创建
     *
     * @param ratelimiter 限流器
     * @param ratelimitconfig 限流配置
     */
    private void checkifconfigneedupdate(rratelimiter ratelimiter, ratelimitproperties.config ratelimitconfig) {
        long count = ratelimitconfig.getcount();
        long timeout = ratelimitconfig.gettimeout();
        rateintervalunit rateintervalunit = ratelimitconfig.getrateintervalunit();
        // 获取之前限流器的配置信息
        ratelimiterconfig config = ratelimiter.getconfig();
        if (rateintervalunit.tomillis(timeout) != config.getrateinterval() || count != config.getrate()) {
            // 如果当前限流器的配置与配置文件不一致，说明服务器重启过
            ratelimiter.delete();
            // 配置文件为准，重新设置
            ratelimiter.trysetrate(ratetype.overall, count, timeout, rateintervalunit);
        }
    }
}

@restcontroller
@slf4j
public class redissoncontroller {
    @autowired
    private ratelimitutil ratelimitutil;

    @getmapping("/ratelimit/{key}")
    public void ratelimit(@pathvariable string key) {
        log.info("{} start get access token, current time = {}", thread.currentthread().getname(), system.currenttimemillis());
        ratelimitutil.acquire(key);
        log.info("{} successfully start get access token, current time = {}", thread.currentthread().getname(), system.currenttimemillis());
    }
}

-- 获取限流器的配置值：rate、interval 和 type
local rate = redis.call('hget', keys[1], 'rate');          -- 获取限制速率（每秒允许的请求次数）
local interval = redis.call('hget', keys[1], 'interval');  -- 获取限流的时间间隔（单位通常为秒）
local type = redis.call('hget', keys[1], 'type');          -- 获取限流类型（如 overall 或 pre_warming）

-- 如果配置没有初始化（即获取的值为空），抛出错误
-- 这里对应着一开始要设置好限流配置参数，否则lua脚本执行过程会抛出异常
assert(rate ~= false and interval ~= false and type ~= false, 'ratelimiter is not initialized')

-- 当前请求令牌数的键
local valuename = keys[2];   
-- 请求令牌记录的键
local permitsname = keys[4]; 

-- 如果限流类型是 pre_warming，调整键值名
if type == '1' then
    valuename = keys[3];   -- 如果是 pre_warming，则使用不同的令牌数键
    permitsname = keys[5];  -- 如果是 pre_warming，则使用不同的令牌记录键
end;

-- 检查请求的令牌数是否超过了令牌桶的大小
-- 例如：rate设置为10，在acqure的时候大于10，lua脚本就会报错
assert(tonumber(rate) >= tonumber(argv[1]), 'requested permits amount cannot exceed defined rate');

-- 获取当前剩余可用的令牌数
local currentvalue = redis.call('get', valuename);

-- 变量 res 用于存储返回值，指示是否可以获取令牌
-- 如果是nil表示令牌充足，获取令牌成功
-- 如果是一个数字，则表示令牌不足，需要要等待的时间
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-else代码用于保证可用的令牌个数和令牌访问记录中使用的令牌个数之和小于桶的容量rate
        if tonumber(currentvalue) + released > tonumber(rate) then
            local values = redis.call('zrange', permitsname, 0, -1);  -- 获取所有令牌记录
            local used = 0;  -- 已使用的令牌数
            -- 统计所有已使用的令牌数
            for i, v in ipairs(values) do
                local random, permits = struct.unpack('bc0i', v);
                used = used + permits;
            end;
            -- 调整当前令牌数，确保不会超过桶的大小
            currentvalue = tonumber(rate) - used;
        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;

-- 返回结果，可能是等待时间或空值
-- 如果是nil表示令牌充足，获取令牌成功
-- 如果是一个数字，则表示令牌不足，需要要等待的时间
return res;