Redis 并發限流控制

作者：拔土豆的程序員 2022-12-12 09:07:06

令牌桶是一種用于控制請求速率的算法。它可以限制在特定時間內可以提交的請求數量，以避免超過系統的處理能力。

令牌桶概念

令牌桶是一種用于控制請求速率的算法。它可以限制在特定時間內可以提交的請求數量，以避免超過系統的處理能力。令牌桶算法基于一個抽象的“令牌桶”，該桶中可以存放一定數量的令牌。在每個時間單位內，新的令牌會按一定的速率添加到桶中。如果一個請求需要處理，就需要從桶中消耗一定數量的令牌。如果桶中沒有足夠的令牌，則請求將被拒絕。令牌桶算法的優點在于它可以根據當前的系統負載動態調整請求的處理速率，并可以控制請求的速率和延遲。

優缺點

可能會導致請求延遲，如果請求速率較高，則桶中的令牌可能會被消耗完，導致新的請求無法被處理。
可以避免系統被大量請求涌入而導致的資源耗盡，并可以根據實際情況動態調整請求處理速率。

分析

核心參數：

桶的容量：它表示桶中最多能存放多少令牌。
令牌添加速率：每個時間單位內（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 }

redis.replicate_commands() 是 Redis 客戶端的一個方法，它用于啟用命令復制（command replication）。命令復制是指，在多個 Redis 實例之間復制命令，以保證數據的一致性。

例如，如果你在一個 Redis 集群中執行了一條寫入命令，那么這條命令就會被復制到集群中的其他實例中。這樣，就可以保證集群中的所有實例都保存了相同的數據，并且可以提供高可用性和數據安全性。

責任編輯：武曉燕來源：今日頭條

Redis 并發限流