Redis 并发限流控制

令牌桶概念

令牌桶是一种用于控制请求速率的算法。它可以限制在特定时间内可以提交的请求数量，以避免超过系统的处理能力。令牌桶算法基于一个抽象的“令牌桶”，该桶中可以存放一定数量的令牌。在每个时间单位内，新的令牌会按一定的速率添加到桶中。如果一个请求需要处理，就需要从桶中消耗一定数量的令牌。如果桶中没有足够的令牌，则请求将被拒绝。令牌桶算法的优点在于它可以根据当前的系统负载动态调整请求的处理速率，并可以控制请求的速率和延迟。

优缺点

可能会导致请求延迟，如果请求速率较高，则桶中的令牌可能会被消耗完，导致新的请求无法被处理。
可以避免系统被大量请求涌入而导致的资源耗尽，并可以根据实际情况动态调整请求处理速率。

分析

核心参数：

桶的容量：它表示桶中最多能存放多少令牌。
令牌添加速率：每个时间单位内（1s）令牌桶中能添加的令牌数量。
每个请求需要的令牌数量：表示每个请求需要消耗的令牌数量，一般默认为 1。

其中令牌添加速率的实现方式为：维护一个时间戳，来记录上一次添加令牌的时间，以便在处理请求时计算令牌添加速率。

综上可以进行代码设计：

public class RedisRateLimiterReq {
/**
    * 限流唯一性标识
    */
   @NotBlank
   private String id;
/**
    * 令牌添加速率
    */
   @Min(1)
   private int replenishRate;
/**
    * 桶的容量
    */
   @Min(0)
   private int burstCapacity =1;
/**
    * 每个请求需要的令牌数量
    */
   @Min(1)
   private int requestedTokens =1;
}

基于Redis+lua的分布式令牌桶限流

redis key 设计：

Key[1] ：记录桶的剩余容量
Key[2] ：记录桶上次刷新时间，以此推算当前需要填入的令牌数量

第一次：需要新填入的令牌数量 = (当前时间 – 0) * 速率
其他后：需要新填入的令牌数量 = (当前时间 – Key[2]) * 速率

综上：当前桶内可用令牌数 = 桶的剩余容量 + 需要新填入的令牌数量

参数设计：

capacity：桶的容量：它表示桶中最多能存放多少令牌。
rate：令牌添加速率：每个时间单位内（1s）令牌桶中能添加的令牌数量。
requested：每个请求需要的令牌数量：表示每个请求需要消耗的令牌数量，一般默认为 1。

核心公式：

fill_time：填充时间：capacity / rate，例如 10/2，即每秒填充 5 个令牌。
ttl：redis key[1]、key[2] 的过期时间，填充时间*2；为什么是2倍：这样可以保证令牌桶中的令牌能够被充分利用，并避免过早的过期。例如，如果填充时间的值为 10 秒，那么过期时间的值就应该设置为 20 秒。这样，在令牌桶的生存周期内，用户就有足够的时间来使用令牌桶中的令牌。

LUA 脚本

redis.replicate_commands()
-- 记录桶的剩余容量
local tokens_key = KEYS[1]
-- 记录桶上次刷新时间，以此推算当前需要填入的令牌数量
-- 第一次：需要新填入的令牌数量 = (当前时间 - 0) * 速率
-- 其他后：需要新填入的令牌数量 = (当前时间 - Key[2]) * 速率
local timestamp_key = KEYS[2]
-- 综上：**当前桶内可用令牌数 = 桶的剩余容量 + 需要新填入的令牌数量**
redis.log(redis.LOG_WARNING,"tokens_key " .. tokens_key)

local rate = tonumber(ARGV[1])
local capacity = tonumber(ARGV[2])
local now = redis.call('TIME')[1]
local requested = tonumber(ARGV[4])

local fill_time = capacity/rate
-- redis key[1]、key[2] 的过期时间
-- 令牌过期时间：填充时间*2
-- 返回小于参数x的最大整数
-- 这样可以保证令牌桶中的令牌能够被充分利用，并避免过早的过期。
-- 例如，如果填充时间的值为 10 秒，那么过期时间的值就应该设置为 20 秒。这样，在令牌桶的生存周期内，用户就有足够的时间来使用令牌桶中的令牌。
local ttl = math.floor(fill_time*2)

redis.log(redis.LOG_WARNING,"rate " .. ARGV[1])
redis.log(redis.LOG_WARNING,"capacity " .. ARGV[2])
redis.log(redis.LOG_WARNING,"now " .. now)
redis.log(redis.LOG_WARNING,"requested " .. ARGV[4])
redis.log(redis.LOG_WARNING,"filltime " .. fill_time)
redis.log(redis.LOG_WARNING,"ttl " .. ttl)

local last_tokens = tonumber(redis.call("get", tokens_key))
if last_tokens == nil then
   last_tokens = capacity
end
redis.log(redis.LOG_WARNING,"last_tokens " .. last_tokens)

local last_refreshed = tonumber(redis.call("get", timestamp_key))
if last_refreshed == nil then
   last_refreshed =0
end
redis.log(redis.LOG_WARNING,"last_refreshed " .. last_refreshed)

local delta = math.max(0, now-last_refreshed)
local filled_tokens = math.min(capacity, last_tokens+(delta*rate))
local allowed = filled_tokens >= requested
local new_tokens = filled_tokens
local allowed_num =0
if allowed then
   new_tokens = filled_tokens - requested
   allowed_num =1
end

--redis.log(redis.LOG_WARNING, "delta " .. delta)
--redis.log(redis.LOG_WARNING, "filled_tokens " .. filled_tokens)
--redis.log(redis.LOG_WARNING, "allowed_num " .. allowed_num)
--redis.log(redis.LOG_WARNING, "new_tokens " .. new_tokens)

if ttl >0 then
   redis.call("setex", tokens_key, ttl, new_tokens)
   redis.call("setex", timestamp_key, ttl, now)
end

-- return { allowed_num, new_tokens, capacity, filled_tokens, requested, new_tokens }
return { allowed_num, new_tokens }