引言

在微服務(wù)架構(gòu)中，對(duì) API 網(wǎng)關(guān)或單個(gè)服務(wù)的請(qǐng)求進(jìn)行速率限制至關(guān)重要，以防止惡意攻擊、資源濫用并確保系統(tǒng)的穩(wěn)定性和可用性。 Redis 憑借其高性能、原子操作和豐富的數(shù)據(jù)結(jié)構(gòu)，成為實(shí)現(xiàn)請(qǐng)求限流的理想選擇。

本文將詳細(xì)探討如何利用 Redis 實(shí)現(xiàn) API 網(wǎng)關(guān)或單個(gè)服務(wù)的請(qǐng)求限流，深入分析各種主流算法，并提供在 Python 和 Java 環(huán)境下的具體代碼示例，包括使用 Lua 腳本確保操作的原子性。

為什么選擇 Redis 進(jìn)行限流？

Redis 之所以成為實(shí)現(xiàn)限流的熱門選擇，主要得益于其以下幾個(gè)關(guān)鍵特性：

• 卓越的性能： Redis 是一款內(nèi)存數(shù)據(jù)庫，能夠提供極快的讀寫速度，這對(duì)于需要實(shí)時(shí)處理大量請(qǐng)求的限流場(chǎng)景至關(guān)重要。
• 原子操作： Redis 的 INCR、EXPIRE 等命令是原子性的，可以避免在并發(fā)請(qǐng)求下出現(xiàn)競(jìng)態(tài)條件，確保計(jì)數(shù)和過期的準(zhǔn)確性。
• 豐富的數(shù)據(jù)結(jié)構(gòu)： Redis 提供了字符串、哈希、列表、有序集合等多種數(shù)據(jù)結(jié)構(gòu)，可以靈活地實(shí)現(xiàn)各種復(fù)雜的限流算法。
• 可擴(kuò)展性： Redis 支持集群和橫向擴(kuò)展，能夠應(yīng)對(duì)高并發(fā)的請(qǐng)求環(huán)境。
• 自帶過期機(jī)制： 通過 EXPIRE 命令可以為鍵設(shè)置生存時(shí)間（TTL），輕松實(shí)現(xiàn)時(shí)間窗口的自動(dòng)重置。

核心概念：識(shí)別請(qǐng)求來源

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

• IP 地址： 基于客戶端的 IP 地址進(jìn)行限制，是簡(jiǎn)單有效的常用方法。
• 用戶 ID 或 API 密鑰： 針對(duì)已認(rèn)證的用戶或第三方應(yīng)用進(jìn)行精細(xì)化限流。
• 設(shè)備 ID： 對(duì)移動(dòng)端等特定設(shè)備進(jìn)行限制。

主流限流算法及其 Redis 實(shí)現(xiàn)

以下是幾種主流的限流算法及其使用 Redis 的實(shí)現(xiàn)方式：

1. 固定窗口計(jì)數(shù)器 (Fixed Window Counter)

這是最簡(jiǎn)單的限流算法。它在固定的時(shí)間窗口內(nèi)（例如，每分鐘）統(tǒng)計(jì)請(qǐng)求次數(shù)，如果超過預(yù)設(shè)的閾值，則拒絕后續(xù)的請(qǐng)求，直到下一個(gè)時(shí)間窗口開始。

優(yōu)點(diǎn)： 實(shí)現(xiàn)簡(jiǎn)單，容易理解。

缺點(diǎn)： 在時(shí)間窗口的邊界處可能會(huì)出現(xiàn)“突刺”流量問題。例如，在窗口結(jié)束前的瞬間和新窗口開始的瞬間，可能會(huì)有兩倍于限制的請(qǐng)求通過。

Redis 實(shí)現(xiàn):

主要利用 INCR 和 EXPIRE 命令。

Python 示例:

import redis
import time

r = 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 False

    if int(current_requests) >= limit:
        return True

    r.incr(key)
    return False

# 示例: 每位用戶每60秒最多10個(gè)請(qǐng)求
for i in range(15):
    if is_rate_limited_fixed_window("user123", 10, 60):
        print("請(qǐng)求被限制")
    else:
        print("請(qǐng)求成功")
    time.sleep(1)

使用 Lua 腳本保證原子性:

為了避免 GET 和 INCR 之間的競(jìng)態(tài)條件，強(qiáng)烈建議使用 Lua 腳本將多個(gè)命令作為一個(gè)原子操作執(zhí)行。

-- 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 then
    return 1 -- 1 表示被限制
end

if current == 0 then
    redis.call("INCR", key)
    redis.call("EXPIRE", key, window)
else
    redis.call("INCR", key)
end

return 0 -- 0 表示未被限制

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

import redis.clients.jedis.Jedis;

public class RateLimiter {
    private final Jedis jedis;
    private final String script;

    public RateLimiter(Jedis jedis) {
        this.jedis = jedis;
        // 實(shí)際項(xiàng)目中應(yīng)從文件加載
        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. 滑動(dòng)窗口日志 (Sliding Window Log)

該算法記錄每個(gè)請(qǐng)求的時(shí)間戳。當(dāng)一個(gè)新請(qǐng)求到達(dá)時(shí)，會(huì)移除時(shí)間窗口之外的舊時(shí)間戳，然后統(tǒng)計(jì)窗口內(nèi)的時(shí)間戳數(shù)量。如果數(shù)量超過限制，則拒絕請(qǐng)求。

優(yōu)點(diǎn)： 限流精度高，有效解決了固定窗口的邊界問題。

缺點(diǎn)： 需要存儲(chǔ)所有請(qǐng)求的時(shí)間戳，當(dāng)請(qǐng)求量很大時(shí)會(huì)占用較多內(nèi)存。

Redis 實(shí)現(xiàn):

使用有序集合 (Sorted Set)，將成員 (member) 設(shè)置為唯一值（如請(qǐng)求 ID 或時(shí)間戳），將分?jǐn)?shù) (score) 設(shè)置為請(qǐng)求的時(shí)間戳。

Python 示例:

import redis
import time

r = 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()
    # 移除窗口外的數(shù)據(jù)
    pipe.zremrangebyscore(key, 0, window_start)
    # 添加當(dāng)前請(qǐng)求
    pipe.zadd(key, {f"{now}:{int(time.time()*1000000)}": now}) # 保證 member 唯一
    # 獲取窗口內(nèi)的請(qǐng)求數(shù)
    pipe.zcard(key)
    # 設(shè)置過期時(shí)間，防止冷數(shù)據(jù)占用內(nèi)存
    pipe.expire(key, window)
    results = pipe.execute()

    current_requests = results[2]
    return current_requests > limit

使用 Lua 腳本實(shí)現(xiàn)滑動(dòng)窗口:

-- 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

-- 移除過期的請(qǐng)求記錄
redis.call("ZREMRANGEBYSCORE", key, 0, window_start)

-- 獲取當(dāng)前窗口內(nèi)的請(qǐng)求數(shù)
local count = redis.call("ZCARD", key)

if count >= limit then
    return 1 -- 被限制
end

-- 添加當(dāng)前請(qǐng)求
redis.call("ZADD", key, now_ms, now_ms .. "-" .. count) -- member 保證唯一性
redis.call("EXPIRE", key, window)

return 0 -- 未被限制

3. 令牌桶算法 (Token Bucket)

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

優(yōu)點(diǎn)： 能夠應(yīng)對(duì)突發(fā)流量，只要桶內(nèi)有足夠的令牌，就可以一次性處理多個(gè)請(qǐng)求。

缺點(diǎn)： 實(shí)現(xiàn)相對(duì)復(fù)雜，需要一個(gè)獨(dú)立的進(jìn)程或定時(shí)任務(wù)來持續(xù)生成令牌。

Redis 實(shí)現(xiàn):

可以使用 Redis 的列表 (List) 作為令牌桶。一個(gè)后臺(tái)任務(wù)（或在每次請(qǐng)求時(shí)計(jì)算）負(fù)責(zé)向列表中添加令牌。

Java (Spring Boot with Bucket4j) 示例:

Bucket4j 是一個(gè)流行的 Java 限流庫，可以與 Redis 結(jié)合使用實(shí)現(xiàn)分布式令牌桶。

// 依賴: bucket4j-core, jcache-api, redisson
// 配置 RedissonClient...

// 創(chuàng)建一個(gè)基于 Redis 的代理管理器
ProxyManager<String> proxyManager = new JCacheProxyManager<>(jcache);

// 定義令牌桶配置：每分鐘補(bǔ)充10個(gè)令牌，桶容量為10
BucketConfiguration configuration = BucketConfiguration.builder()
        .addLimit(Bandwidth.simple(10, Duration.ofMinutes(1)))
        .build();

// 獲取或創(chuàng)建令牌桶
Bucket bucket = proxyManager.getProxy("rate-limit-bucket:user123", configuration);

// 嘗試消費(fèi)一個(gè)令牌
if (bucket.tryConsume(1)) {
    // 請(qǐng)求成功
} else {
    // 請(qǐng)求被限制
}

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

4. 漏桶算法 (Leaky Bucket)

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

優(yōu)點(diǎn)： 能夠平滑請(qǐng)求流量，保證服務(wù)以恒定的速率處理請(qǐng)求。

缺點(diǎn)： 無法有效利用系統(tǒng)空閑資源來處理突發(fā)流量。

Redis 實(shí)現(xiàn):

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

API 網(wǎng)關(guān)限流 vs. 單個(gè)服務(wù)限流

限流邏輯可以部署在不同的層面，主要分為 API 網(wǎng)關(guān)層和單個(gè)微服務(wù)層。

API 網(wǎng)關(guān)限流

在 API 網(wǎng)關(guān)層面進(jìn)行統(tǒng)一限流是一種常見的做法。

優(yōu)點(diǎn)：

• 集中管理： 可以在一個(gè)地方統(tǒng)一配置和管理所有服務(wù)的限流規(guī)則。
• 保護(hù)后端服務(wù)： 將惡意或超額流量擋在微服務(wù)集群之外，保護(hù)整個(gè)系統(tǒng)的穩(wěn)定性。

實(shí)現(xiàn)方式：

• 利用網(wǎng)關(guān)自帶功能： 像 Spring Cloud Gateway、Kong、KrakenD 等 API 網(wǎng)關(guān)都提供了基于 Redis 的限流插件或模塊。
• 自定義中間件： 在網(wǎng)關(guān)中編寫自定義的中間件或過濾器，嵌入上述的 Redis 限流邏輯。

Spring Cloud Gateway 示例 (application.yml):

spring:
  cloud:
    gateway:
      routes:
      - id: my_route
        uri: lb://my-service
        predicates:
        - Path=/my-api/**
        filters:
        - name: RequestRateLimiter
          args:
            redis-rate-limiter.replenishRate: 10      # 令牌桶每秒填充速率
            redis-rate-limiter.burstCapacity: 20     # 令牌桶容量
            redis-rate-limiter.requestedTokens: 1  # 每次請(qǐng)求消耗的令牌數(shù)
            key-resolver: "#{@ipKeyResolver}"     # 使用 IP 地址作為限流的 key

單個(gè)服務(wù)限流

將限流邏輯直接實(shí)現(xiàn)在單個(gè)微服務(wù)內(nèi)部。

優(yōu)點(diǎn)：

• 靈活性： 每個(gè)服務(wù)可以根據(jù)自身的負(fù)載能力和業(yè)務(wù)需求，定制更精細(xì)化的限流策略。
• 獨(dú)立性： 不依賴于特定的 API 網(wǎng)關(guān)。

實(shí)現(xiàn)方式：

• 通過 AOP (面向切面編程) 創(chuàng)建注解，對(duì)需要限流的 Controller 方法進(jìn)行攔截。
• 在服務(wù)的入口處，如攔截器或過濾器中，調(diào)用 Redis 進(jìn)行限流判斷。

Python (FastAPI 中間件) 示例:

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
import redis

app = FastAPI()
r = redis.Redis()

@app.middleware("http")
async def rate_limit_middleware(request: Request, call_next):
    # 簡(jiǎn)單示例：使用固定窗口計(jì)數(shù)器
    ip = request.client.host
    key = f"rate_limit:{ip}"
    limit = 5
    window = 60

    # 此處應(yīng)使用 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

總結(jié)

使用 Redis 實(shí)現(xiàn)請(qǐng)求限流是一種高效且可擴(kuò)展的方案。開發(fā)者應(yīng)根據(jù)具體的業(yè)務(wù)場(chǎng)景和需求選擇合適的限流算法。

• 對(duì)于簡(jiǎn)單的限流需求，固定窗口計(jì)數(shù)器是一個(gè)不錯(cuò)的起點(diǎn)。
• 為了更精確地控制流量并避免邊界問題，滑動(dòng)窗口日志或滑動(dòng)窗口計(jì)數(shù)器是更好的選擇。
• 如果需要應(yīng)對(duì)突發(fā)流量，令牌桶算法則非常適用。

在實(shí)施時(shí)，務(wù)必使用 Lua 腳本來保證操作的原子性，避免在并發(fā)環(huán)境下出現(xiàn)數(shù)據(jù)不一致的問題。將限流邏輯部署在 API 網(wǎng)關(guān)層可以對(duì)整個(gè)系統(tǒng)起到保護(hù)作用，而部署在單個(gè)服務(wù)層則提供了更高的靈活性。在實(shí)際應(yīng)用中，兩者也常常結(jié)合使用，構(gòu)建多層級(jí)的防護(hù)體系。

以上就是使用Redis實(shí)現(xiàn)API網(wǎng)關(guān)或單個(gè)服務(wù)的請(qǐng)求限流的具體代碼的詳細(xì)內(nèi)容，更多關(guān)于Redis API網(wǎng)關(guān)或服務(wù)請(qǐng)求限流的資料請(qǐng)關(guān)注腳本之家其它相關(guān)文章！

最新評(píng)論