1. API速率限制的基本概念

API速率限制（Rate Limiting）是控制用戶訪問API的請求速率的一種機(jī)制，防止系統(tǒng)被過多請求淹沒。通過對用戶的請求進(jìn)行計(jì)數(shù)與限制，API能夠在高并發(fā)情況下維持性能與穩(wěn)定性。

速率限制的主要目的是：

• 防止過多請求導(dǎo)致服務(wù)器負(fù)載過高。
• 限制惡意或非正常行為。
• 提高API的可用性，確保公平分配資源。

常見的速率限制算法

1.漏桶算法（Leaky Bucket Algorithm）

漏桶算法在一定時(shí)間窗口內(nèi)以固定速率處理請求，若請求速率超過預(yù)定限度，則會(huì)被丟棄或延遲。

2.令牌桶算法（Token Bucket Algorithm）

令牌桶算法是一種靈活的速率控制機(jī)制，適合處理突發(fā)流量。每個(gè)請求都需要獲取一個(gè)令牌，如果令牌桶為空，則請求被丟棄。

3.固定窗口計(jì)數(shù)法（Fixed Window Counter）

在固定時(shí)間窗口內(nèi)，計(jì)數(shù)器記錄請求的次數(shù)，一旦請求超出限制，后續(xù)請求將被拒絕。

4.滑動(dòng)窗口計(jì)數(shù)法（Sliding Window Counter）

滑動(dòng)窗口比固定窗口更加精細(xì)，每個(gè)請求都在一個(gè)滑動(dòng)的時(shí)間窗口內(nèi)進(jìn)行計(jì)數(shù)，能平滑流量控制。

通過這些算法，API能夠控制不同用戶或客戶端在指定時(shí)間內(nèi)發(fā)起的請求數(shù)量，確保系統(tǒng)的平穩(wěn)運(yùn)行。

2. Redis實(shí)現(xiàn)分布式速率限制

Redis是一個(gè)高性能的鍵值數(shù)據(jù)庫，廣泛用于緩存、消息隊(duì)列和分布式速率限制等場景。在分布式系統(tǒng)中，Redis提供了高效的數(shù)據(jù)存儲和共享機(jī)制，可以幫助不同服務(wù)器實(shí)例共享請求計(jì)數(shù)信息，從而實(shí)現(xiàn)跨服務(wù)器的速率限制。

Redis的實(shí)現(xiàn)思路

我們使用Redis的SETEX命令設(shè)置一個(gè)鍵值對，其中鍵為用戶標(biāo)識（例如IP或用戶ID），值為請求計(jì)數(shù)。每次用戶發(fā)起請求時(shí)，我們先檢查該鍵是否存在。如果存在，檢查其值是否超過限額；如果不存在，設(shè)置新的鍵并開始計(jì)數(shù)。通過設(shè)置鍵的過期時(shí)間，可以實(shí)現(xiàn)速率限制。

示例代碼

from fastapi import FastAPI, Request, HTTPException
import redis
import time

app = FastAPI()

# 連接Redis服務(wù)器
r = redis.Redis(host='localhost', port=6379, db=0)

# 限制參數(shù)
LIMIT = 100  # 每分鐘100次請求
TIME_WINDOW = 60  # 1分鐘

@app.middleware("http")
async def rate_limit(request: Request, call_next):
    ip_address = request.client.host
    current_time = int(time.time())
    
    # 構(gòu)造Redis的鍵
    redis_key = f"rate_limit:{ip_address}:{current_time // TIME_WINDOW}"
    
    # 使用Redis的INCR命令增加計(jì)數(shù)
    request_count = r.incr(redis_key)
    
    if request_count == 1:
        # 設(shè)置過期時(shí)間為60秒（時(shí)間窗口大?。?
        r.expire(redis_key, TIME_WINDOW)
    
    if request_count > LIMIT:
        raise HTTPException(status_code=429, detail="Too many requests")
    
    response = await call_next(request)
    return response

代碼解析

• r.incr(redis_key)：Redis的INCR命令可以原子性地遞增鍵的值。如果鍵不存在，它會(huì)先創(chuàng)建鍵并設(shè)置初值為1。
• r.expire(redis_key, TIME_WINDOW)：設(shè)置鍵的過期時(shí)間，使得計(jì)數(shù)在每個(gè)時(shí)間窗口內(nèi)自動(dòng)重置。
• 429 Too Many Requests：當(dāng)請求次數(shù)超過限制時(shí)，返回429狀態(tài)碼表示超出請求頻率限制。

這種方式可以有效防止單個(gè)IP地址在短時(shí)間內(nèi)發(fā)送過多請求，保障API的可用性與性能。

3. 防止DDoS攻擊的常見策略

DDoS（Distributed Denial of Service）攻擊通過大量惡意請求淹沒目標(biāo)服務(wù)器，導(dǎo)致系統(tǒng)不可用。為了防止這種攻擊，除了傳統(tǒng)的防火墻和負(fù)載均衡策略外，我們還需要在API層面實(shí)現(xiàn)防護(hù)。

常見的防御策略

1.IP黑名單/白名單

基于IP的訪問控制可以有效阻止已知攻擊源的流量。通過將惡意IP加入黑名單，可以防止這些IP的請求進(jìn)入系統(tǒng)。

2.請求速率限制

利用速率限制算法（如漏桶或令牌桶），控制請求頻率，避免單個(gè)來源發(fā)送過多請求。

3.行為分析與智能防護(hù)

通過分析請求的行為模式，識別并阻止異常流量。例如，檢測異常的請求頭、請求頻率、請求路徑等。

4.驗(yàn)證碼與身份驗(yàn)證

在用戶請求的關(guān)鍵環(huán)節(jié)，如登錄、注冊、支付等，加入驗(yàn)證碼或二次身份驗(yàn)證，防止惡意機(jī)器人自動(dòng)化攻擊。

示例代碼：基于IP的速率限制和驗(yàn)證碼

from fastapi import FastAPI, HTTPException, Request
from fastapi.responses import JSONResponse
from pydantic import BaseModel
import redis
import time
import random

app = FastAPI()

r = redis.Redis(host='localhost', port=6379, db=0)
LIMIT = 100
TIME_WINDOW = 60
CAPTCHA_THRESHOLD = 10

@app.post("/login")
async def login(request: Request, user: BaseModel):
    ip_address = request.client.host
    current_time = int(time.time())
    
    redis_key = f"rate_limit:{ip_address}:{current_time // TIME_WINDOW}"
    request_count = r.incr(redis_key)
    
    if request_count == 1:
        r.expire(redis_key, TIME_WINDOW)
    
    if request_count > LIMIT:
        # 啟動(dòng)驗(yàn)證碼機(jī)制
        captcha = random.randint(1000, 9999)
        return JSONResponse(content={"captcha_required": True, "captcha": captcha}, status_code=400)
    
    return {"message": "Login successful"}

代碼解析

當(dāng)請求頻率超過限制時(shí)，返回一個(gè)驗(yàn)證碼，用戶需要通過驗(yàn)證碼驗(yàn)證來繼續(xù)操作。

這種方式有效阻止了自動(dòng)化攻擊，減少了惡意請求的成功率。

4. 基于IP或用戶身份的訪問頻率控制

除了全局的速率限制外，還可以根據(jù)IP地址或用戶身份來單獨(dú)限制訪問頻率。通過這種方法，可以更精細(xì)化地控制API的訪問權(quán)限，避免某個(gè)特定用戶或IP占用過多資源。

示例代碼：基于用戶身份的訪問頻率控制

from fastapi import Depends, HTTPException, Request
from pydantic import BaseModel

@app.get("/user_dashboard")
async def user_dashboard(user_id: str, request: Request):
    user_limit_key = f"user:{user_id}:rate_limit"
    ip_limit_key = f"ip:{request.client.host}:rate_limit"
    
    # 用戶訪問頻率限制
    user_request_count = r.incr(user_limit_key)
    if user_request_count == 1:
        r.expire(user_limit_key, TIME_WINDOW)
    
    if user_request_count > LIMIT:
        raise HTTPException(status_code=429, detail="User request limit exceeded")
    
    # IP訪問頻率限制
    ip_request_count = r.incr(ip_limit_key)
    if ip_request_count == 1:
        r.expire(ip_limit_key, TIME_WINDOW)
    
    if ip_request_count > LIMIT:
        raise HTTPException(status_code=429, detail="IP request limit exceeded")
    
    return {"message": "Welcome to the user dashboard"}

代碼解析

user_id：每個(gè)用戶有獨(dú)立的請求計(jì)數(shù)，防止某個(gè)用戶濫用API。

request.client.host：IP地址的請求計(jì)數(shù)，防止同一個(gè)IP地址濫用API。

根據(jù)用戶和IP的訪問頻率分別設(shè)置限制，提高了控制精度。

到此這篇關(guān)于使用Redis實(shí)現(xiàn)請求限制與速率限制的文章就介紹到這了,更多相關(guān)Redis請求限制與速率限制內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

最新評論