如何使用 Redis 實現 API 網關或單個服務的請求限流？

使用 Redis 高效實現 API 網關與服務的請求限流

在微服務架構中，對 API 網關或單個服務的請求進行速率限制至關重要，以防止惡意攻擊、資源濫用并確保系統的穩定性和可用性。 Redis 憑借其高性能、原子操作和豐富的數據結構，成為實現請求限流的理想選擇。

本文將詳細探討如何利用 Redis 實現 API 網關或單個服務的請求限流，深入分析各種主流算法，并提供在 Python 和 Java 環境下的具體代碼示例，包括使用 Lua 腳本確保操作的原子性。

為什么選擇 Redis 進行限流？

Redis 之所以成為實現限流的熱門選擇，主要得益于其以下幾個關鍵特性：

卓越的性能： Redis 是一款內存數據庫，能夠提供極快的讀寫速度，這對于需要實時處理大量請求的限流場景至關重要。
原子操作： Redis 的 INCR、EXPIRE 等命令是原子性的，可以避免在并發請求下出現競態條件，確保計數和過期的準確性。
豐富的數據結構： Redis 提供了字符串、哈希、列表、有序集合等多種數據結構，可以靈活地實現各種復雜的限流算法。
可擴展性： Redis 支持集群和橫向擴展，能夠應對高并發的請求環境。
自帶過期機制： 通過 EXPIRE 命令可以為鍵設置生存時間（TTL），輕松實現時間窗口的自動重置。

核心概念：識別請求來源

在實施限流之前，首先需要確定如何識別和區分不同的請求來源。常見的識別方式包括：

IP 地址： 基于客戶端的 IP 地址進行限制，是簡單有效的常用方法。
用戶 ID 或 API 密鑰： 針對已認證的用戶或第三方應用進行精細化限流。
設備 ID： 對移動端等特定設備進行限制。

主流限流算法及其 Redis 實現

以下是幾種主流的限流算法及其使用 Redis 的實現方式：

1. 固定窗口計數器 (Fixed Window Counter)

這是最簡單的限流算法。它在固定的時間窗口內（例如，每分鐘）統計請求次數，如果超過預設的閾值，則拒絕后續的請求，直到下一個時間窗口開始。

優點： 實現簡單，容易理解。

缺點： 在時間窗口的邊界處可能會出現“突刺”流量問題。例如，在窗口結束前的瞬間和新窗口開始的瞬間，可能會有兩倍于限制的請求通過。

Redis 實現:

主要利用 INCR 和 EXPIRE 命令。

Python 示例:

import redis
import timer = redis.Redis()def is_rate_limited_fixed_window(user_id: str, limit: int, window: int) -> bool:key = f"rate_limit:{user_id}"current_requests = r.get(key)if current_requests is None:# 使用 pipeline 保證原子性pipe = r.pipeline()pipe.incr(key)pipe.expire(key, window)pipe.execute()return Falseif int(current_requests) >= limit:return Truer.incr(key)return False# 示例: 每位用戶每60秒最多10個請求
for i in range(15):if is_rate_limited_fixed_window("user123", 10, 60):print("請求被限制")else:print("請求成功")time.sleep(1)

使用 Lua 腳本保證原子性:

為了避免 GET 和 INCR 之間的競態條件，強烈建議使用 Lua 腳本將多個命令作為一個原子操作執行。

-- rate_limiter.lua
local key = KEYS[1]
local limit = tonumber(ARGV[1])
local window = tonumber(ARGV[2])local current = tonumber(redis.call("GET", key) or "0")if current >= limit thenreturn 1 -- 1 表示被限制
endif current == 0 thenredis.call("INCR", key)redis.call("EXPIRE", key, window)
elseredis.call("INCR", key)
endreturn 0 -- 0 表示未被限制

Java (Jedis) 調用 Lua 腳本示例:

import redis.clients.jedis.Jedis;public class RateLimiter {private final Jedis jedis;private final String script;public RateLimiter(Jedis jedis) {this.jedis = jedis;// 實際項目中應從文件加載this.script = "local key = KEYS[1]..."}public boolean isRateLimited(String key, int limit, int window) {Object result = jedis.eval(this.script, 1, key, String.valueOf(limit), String.valueOf(window));return "1".equals(result.toString());}
}

2. 滑動窗口日志 (Sliding Window Log)

該算法記錄每個請求的時間戳。當一個新請求到達時，會移除時間窗口之外的舊時間戳，然后統計窗口內的時間戳數量。如果數量超過限制，則拒絕請求。

優點： 限流精度高，有效解決了固定窗口的邊界問題。

缺點： 需要存儲所有請求的時間戳，當請求量很大時會占用較多內存。

Redis 實現:

使用有序集合 (Sorted Set)，將成員 (member) 設置為唯一值（如請求 ID 或時間戳），將分數 (score) 設置為請求的時間戳。

Python 示例:

import redis
import timer = redis.Redis()def is_rate_limited_sliding_log(user_id: str, limit: int, window: int) -> bool:key = f"rate_limit_log:{user_id}"now = int(time.time() * 1000)window_start = now - window * 1000# 使用 pipeline 保證原子性pipe = r.pipeline()# 移除窗口外的數據pipe.zremrangebyscore(key, 0, window_start)# 添加當前請求pipe.zadd(key, {f"{now}:{int(time.time()*1000000)}": now}) # 保證 member 唯一# 獲取窗口內的請求數pipe.zcard(key)# 設置過期時間，防止冷數據占用內存pipe.expire(key, window)results = pipe.execute()current_requests = results[2]return current_requests > limit

使用 Lua 腳本實現滑動窗口:

-- sliding_window.lua
local key = KEYS[1]
local limit = tonumber(ARGV[1])
local window = tonumber(ARGV[2])
local now = redis.call("TIME")
local now_ms = (now[1] * 1000) + math.floor(now[2] / 1000)
local window_start = now_ms - window * 1000-- 移除過期的請求記錄
redis.call("ZREMRANGEBYSCORE", key, 0, window_start)-- 獲取當前窗口內的請求數
local count = redis.call("ZCARD", key)if count >= limit thenreturn 1 -- 被限制
end-- 添加當前請求
redis.call("ZADD", key, now_ms, now_ms .. "-" .. count) -- member 保證唯一性
redis.call("EXPIRE", key, window)return 0 -- 未被限制

3. 令牌桶算法 (Token Bucket)

該算法的核心是一個固定容量的“令牌桶”，系統會以恒定的速率向桶里放入令牌。每個請求需要從桶里獲取一個令牌才能被處理。如果桶里沒有令牌，請求將被拒絕或排隊等待。

優點： 能夠應對突發流量，只要桶內有足夠的令牌，就可以一次性處理多個請求。

缺點： 實現相對復雜，需要一個獨立的進程或定時任務來持續生成令牌。

Redis 實現:

可以使用 Redis 的列表 (List) 作為令牌桶。一個后臺任務（或在每次請求時計算）負責向列表中添加令牌。

Java (Spring Boot with Bucket4j) 示例:

Bucket4j 是一個流行的 Java 限流庫，可以與 Redis 結合使用實現分布式令牌桶。

// 依賴: bucket4j-core, jcache-api, redisson
// 配置 RedissonClient...// 創建一個基于 Redis 的代理管理器
ProxyManager<String> proxyManager = new JCacheProxyManager<>(jcache);// 定義令牌桶配置：每分鐘補充10個令牌，桶容量為10
BucketConfiguration configuration = BucketConfiguration.builder().addLimit(Bandwidth.simple(10, Duration.ofMinutes(1))).build();// 獲取或創建令牌桶
Bucket bucket = proxyManager.getProxy("rate-limit-bucket:user123", configuration);// 嘗試消費一個令牌
if (bucket.tryConsume(1)) {// 請求成功
} else {// 請求被限制
}

Spring Cloud Gateway 也內置了基于 Redis 的令牌桶算法限流器。

4. 漏桶算法 (Leaky Bucket)

漏桶算法將請求看作是流入“漏桶”的水，而漏桶以固定的速率漏出水（處理請求）。如果流入的速率過快，導致桶溢出，則多余的請求將被丟棄。

優點： 能夠平滑請求流量，保證服務以恒定的速率處理請求。

缺點： 無法有效利用系統空閑資源來處理突發流量。

Redis 實現:

Redis-Cell 模塊提供了基于 GCRA (Generic Cell Rate Algorithm) 的漏桶算法實現。

API 網關限流 vs. 單個服務限流

限流邏輯可以部署在不同的層面，主要分為 API 網關層和單個微服務層。

API 網關限流

在 API 網關層面進行統一限流是一種常見的做法。

優點：
- 集中管理： 可以在一個地方統一配置和管理所有服務的限流規則。
- 保護后端服務： 將惡意或超額流量擋在微服務集群之外，保護整個系統的穩定性。
實現方式：
- 利用網關自帶功能： 像 Spring Cloud Gateway、Kong、KrakenD 等 API 網關都提供了基于 Redis 的限流插件或模塊。
- 自定義中間件： 在網關中編寫自定義的中間件或過濾器，嵌入上述的 Redis 限流邏輯。

Spring Cloud Gateway 示例 (application.yml):

spring:cloud:gateway:routes:- id: my_routeuri: lb://my-servicepredicates:- Path=/my-api/**filters:- name: RequestRateLimiterargs:redis-rate-limiter.replenishRate: 10      # 令牌桶每秒填充速率redis-rate-limiter.burstCapacity: 20     # 令牌桶容量redis-rate-limiter.requestedTokens: 1  # 每次請求消耗的令牌數key-resolver: "#{@ipKeyResolver}"     # 使用 IP 地址作為限流的 key

單個服務限流

將限流邏輯直接實現在單個微服務內部。

優點：
- 靈活性： 每個服務可以根據自身的負載能力和業務需求，定制更精細化的限流策略。
- 獨立性： 不依賴于特定的 API 網關。
實現方式：
- 通過 AOP (面向切面編程) 創建注解，對需要限流的 Controller 方法進行攔截。
- 在服務的入口處，如攔截器或過濾器中，調用 Redis 進行限流判斷。

Python (FastAPI 中間件) 示例:

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
import redisapp = FastAPI()
r = redis.Redis()@app.middleware("http")
async def rate_limit_middleware(request: Request, call_next):# 簡單示例：使用固定窗口計數器ip = request.client.hostkey = f"rate_limit:{ip}"limit = 5window = 60# 此處應使用 Lua 腳本保證原子性current = r.incr(key)if current == 1:r.expire(key, window)if current > limit:return JSONResponse(status_code=429, content={"message": "Too Many Requests"})response = await call_next(request)return response