Trong quá trình xây dựng hệ thống AI gateway cho HolySheep AI, tôi đã thử nghiệm và so sánh hầu hết các thuật toán rate limiting phổ biến. Bài viết này sẽ chia sẻ kinh nghiệm thực chiến, giúp bạn chọn đúng thuật toán và triển khai hiệu quả cho dự án của mình.

Mục lục

Bảng so sánh nhanh các thuật toán Rate Limiting

Thuật toánĐộ chính xácMemoryBurst supportĐộ phức tạpUse case tốt nhất
Token BucketCaoThấpCóTrung bìnhAPI Gateway, chat apps
Leaky BucketCaoThấpKhôngTrung bìnhVideo streaming, IoT
Sliding Window LogRất caoCaoCóCaoTài chính, thanh toán
Sliding Window CounterTrung bình-caoThấpCóTrung bìnhGeneral API
Fixed WindowThấpThấpCóThấpPrototype, đơn giản
AdaptiveCaoTrung bìnhCóCaoMulti-tenant systems

Từ kinh nghiệm triển khai HolySheep API với hơn 50.000 request mỗi ngày, tôi khuyên dùng Token Bucket cho hầu hết trường hợp. Nếu bạn cần độ chính xác cao hơn và có budget cho Redis cluster, hãy chọn Sliding Window Counter.

Token Bucket - Thuật toán phổ biến nhất

Token Bucket hoạt động như một cái xô chứa tokens. Mỗi request cần lấy 1 token từ xô. Xô được refill với tốc độ cố định. Điểm mạnh: cho phép burst (bùng nổ) traffic mà vẫn giới hạn được tổng lượng request trung bình.

Nguyên lý hoạt động

┌─────────────────────────────────────────────────────────┐
│                    TOKEN BUCKET                         │
├─────────────────────────────────────────────────────────┤
│                                                         │
│    refill_rate = 10 tokens/giây                         │
│    bucket_capacity = 30 tokens                          │
│                                                         │
│    ┌─────────┐                                          │
│    │ ░░░░░░░ │ ← Bucket đầy (30 tokens)                │
│    │ ░░░░░░░ │                                          │
│    │ ░░░░░░░ │   Request 1 → Lấy 1 token               │
│    │ ░░░░░░░ │   Request 2 → Lấy 1 token               │
│    │ ░░░░░░░ │   ...                                    │
│    └─────────┘                                          │
│         ↓                                               │
│    refill liên tục theo thời gian                       │
│                                                         │
└─────────────────────────────────────────────────────────┘
class TokenBucket:
    """
    Token Bucket Rate Limiter - Triển khai đơn giản với Python
    Tốc độ refill: 10 tokens/giây
    Dung tích bucket: 30 tokens (cho phép burst)
    """
    
    def __init__(self, capacity: int = 30, refill_rate: float = 10.0):
        self.capacity = capacity
        self.refill_rate = refill_rate
        self.tokens = float(capacity)
        self.last_refill_time = time.time()
        self.lock = threading.Lock()
    
    def _refill(self):
        """Tự động refill tokens dựa trên thời gian trôi qua"""
        now = time.time()
        elapsed = now - self.last_refill_time
        
        # Tính tokens cần thêm vào
        tokens_to_add = elapsed * self.refill_rate
        self.tokens = min(self.capacity, self.tokens + tokens_to_add)
        self.last_refill_time = now
    
    def allow_request(self, tokens: int = 1) -> bool:
        """
        Kiểm tra và lấy token cho request
        Returns: True nếu request được phép, False nếu bị reject
        """
        with self.lock:
            self._refill()
            
            if self.tokens >= tokens:
                self.tokens -= tokens
                return True
            return False
    
    def get_wait_time(self, tokens: int = 1) -> float:
        """Tính thời gian chờ để có đủ tokens"""
        with self.lock:
            self._refill()
            if self.tokens >= tokens:
                return 0.0
            return (tokens - self.tokens) / self.refill_rate



=== SỬ DỤNG VỚI HOLYSHEEP AI API ===

import aiohttp
import asyncio

async def call_holysheep_with_rate_limit(prompt: str, bucket: TokenBucket):
    """Gọi HolySheep AI với rate limiting bằng Token Bucket"""
    
    # Đợi cho đến khi có token
    wait_time = bucket.get_wait_time()
    if wait_time > 0:
        await asyncio.sleep(wait_time)
    
    # Kiểm tra và lấy token
    if not bucket.allow_request():
        raise Exception("Rate limit exceeded - vượt quota")
    
    # Gọi HolySheep API
    base_url = "https://api.holysheep.ai/v1"
    headers = {
        "Authorization": f"Bearer YOUR_HOLYSHEEP_API_KEY",
        "Content-Type": "application/json"
    }
    payload = {
        "model": "gpt-4.1",
        "messages": [{"role": "user", "content": prompt}]
    }
    
    async with aiohttp.ClientSession() as session:
        async with session.post(
            f"{base_url}/chat/completions",
            headers=headers,
            json=payload
        ) as response:
            return await response.json()



Rate limit: 10 requests/giây, burst được 30 requests

limiter = TokenBucket(capacity=30, refill_rate=10.0)

Leaky Bucket - Ổn định đầu ra

Khác với Token Bucket cho phép burst, Leaky Bucket đảm bảo output rate cố định. Mọi request được đưa vào queue và drip ra với tốc độ đều đặn. Hoàn hảo cho video streaming hoặc IoT devices gửi data.

class LeakyBucket:
    """
    Leaky Bucket Rate Limiter
    - Mọi request được xử lý với rate cố định
    - Không cho phép burst
    - Queue có giới hạn
    """
    
    def __init__(self, leak_rate: float = 10.0, bucket_size: int = 100):
        self.leak_rate = leak_rate  # requests/giây
        self.bucket_size = bucket_size
        self.queue = []
        self.last_leak_time = time.time()
        self.lock = threading.Lock()
    
    def _leak(self):
        """Xử lý (drip) requests khỏi bucket theo thời gian"""
        now = time.time()
        elapsed = now - self.last_leak_time
        
        # Tính số requests có thể drip
        leaked = int(elapsed * self.leak_rate)
        if leaked > 0:
            self.queue = self.queue[leaked:]
            self.last_leak_time = now
    
    def allow_request(self) -> tuple[bool, float]:
        """
        Thêm request vào bucket
        Returns: (được phép, thời gian chờ)
        """
        with self.lock:
            self._leak()
            
            if len(self.queue) < self.bucket_size:
                self.queue.append(time.time())
                return True, 0.0
            else:
                # Tính thời gian để request đầu tiên được drip
                wait_time = (len(self.queue) - self.bucket_size + 1) / self.leak_rate
                return False, wait_time
    
    def process_request(self) -> float:
        """Blocking - đợi và xử lý request"""
        while True:
            allowed, wait = self.allow_request()
            if allowed:
                return 0.0
            time.sleep(wait)



Ví dụ: API rate limit 10 requests/giây

leaky = LeakyBucket(leak_rate=10.0, bucket_size=100)

Sliding Window - Độ chính xác cao nhất

Sliding Window chia thời gian thành các windows nhỏ và đếm requests trong window hiện tại + window trước. Đây là thuật toán được dùng trong nhiều payment gateway vì độ chính xác cao.

class SlidingWindowCounter:
    """
    Sliding Window Counter - Hybrid giữa Fixed Window và Sliding Window Log
    Độ chính xác cao với memory thấp
    """
    
    def __init__(self, max_requests: int = 100, window_size: int = 60):
        self.max_requests = max_requests
        self.window_size = window_size  # giây
        self.windows = {}  # {timestamp: count}
        self.lock = threading.Lock()
    
    def _clean_old_windows(self, current_time: float):
        """Xóa các windows đã hết hạn"""
        cutoff = current_time - self.window_size
        self.windows = {
            ts: count for ts, count in self.windows.items() 
            if ts > cutoff
        }
    
    def allow_request(self) -> bool:
        """Kiểm tra request có được phép không"""
        with self.lock:
            current_time = time.time()
            self._clean_old_windows(current_time)
            
            # Đếm tổng requests trong window
            total = sum(self.windows.values())
            
            if total < self.max_requests:
                # Lấy window hiện tại (làm tròn xuống)
                current_window = int(current_time)
                self.windows[current_window] = self.windows.get(current_window, 0) + 1
                return True
            return False
    
    def get_remaining(self) -> int:
        """Lấy số requests còn lại trong window"""
        with self.lock:
            current_time = time.time()
            self._clean_old_windows(current_time)
            
            total = sum(self.windows.values())
            return max(0, self.max_requests - total)



=== TRIỂN KHAI VỚI REDIS CHO DISTRIBUTED SYSTEMS ===

import redis

class RedisSlidingWindow:
    """
    Sliding Window với Redis - cho hệ thống phân tán nhiều servers
    Key format: ratelimit:{user_id}
    """
    
    def __init__(self, redis_client: redis.Redis, max_requests: int = 100, window_seconds: int = 60):
        self.redis = redis_client
        self.max_requests = max_requests
        self.window_seconds = window_seconds
    
    def allow_request(self, identifier: str) -> dict:
        """
        Kiểm tra và ghi nhận request
        Returns thông tin về rate limit
        """
        key = f"ratelimit:{identifier}"
        current_time = time.time()
        window_start = current_time - self.window_seconds
        
        pipe = self.redis.pipeline()
        
        # Xóa entries cũ
        pipe.zremrangebyscore(key, 0, window_start)
        
        # Đếm requests trong window
        pipe.zcard(key)
        
        # Thêm request hiện tại
        pipe.zadd(key, {str(current_time): current_time})
        
        # Set expiry cho key
        pipe.expire(key, self.window_seconds * 2)
        
        results = pipe.execute()
        current_count = results[1]
        
        return {
            "allowed": current_count < self.max_requests,
            "remaining": max(0, self.max_requests - current_count - 1),
            "reset_at": current_time + self.window_seconds
        }



Kết nối Redis

redis_client = redis.Redis(host='localhost', port=6379, db=0)
sliding_window = RedisSlidingWindow(redis_client, max_requests=100, window_seconds=60)


Sử dụng với HolySheep API

def call_with_sliding_window(user_id: str, prompt: str):
    result = sliding_window.allow_request(user_id)
    
    if not result["allowed"]:
        raise Exception(f"Rate limited. Retry after {result['reset_at'] - time.time():.1f}s")
    
    # Gọi HolySheep API
    response = requests.post(
        "https://api.holysheep.ai/v1/chat/completions",
        headers={
            "Authorization": "Bearer YOUR_HOLYSHEEP_API_KEY",
            "Content-Type": "application/json"
        },
        json={
            "model": "claude-sonnet-4.5",
            "messages": [{"role": "user", "content": prompt}]
        }
    )
    
    return response.json()

Fixed Window - Đơn giản nhưng có nhược điểm

Fixed Window chia thời gian thành các khoảng cố định và reset counter vào đầu mỗi window. Đơn giản nhưng có "boundary burst" - user có thể gửi 2x requests tại thời điểm giao giữa 2 windows.

class FixedWindowRateLimiter:
    """
    Fixed Window Rate Limiter - Đơn giản nhất
    ⚠️ Có nhược điểm: boundary burst (2x limit tại thời điểm reset)
    """
    
    def __init__(self, max_requests: int = 100, window_seconds: int = 60):
        self.max_requests = max_requests
        self.window_seconds = window_seconds
        self.windows = {}  # {window_id: count}
    
    def _get_window_id(self) -> int:
        """Lấy ID của window hiện tại"""
        return int(time.time() // self.window_seconds)
    
    def allow_request(self) -> bool:
        window_id = self._get_window_id()
        
        if window_id != self.windows.get("_current_window"):
            # Window mới - reset
            self.windows = {"_current_window": window_id, "count": 0}
        else:
            self.windows["count"] = self.windows.get("count", 0)
        
        if self.windows["count"] < self.max_requests:
            self.windows["count"] += 1
            return True
        return False



=== VÍ DỤ SỬ DỤNG ===


Rate limit: 100 requests/phút

limiter = FixedWindowRateLimiter(max_requests=100, window_seconds=60)

for i in range(105):
    if limiter.allow_request():
        print(f"Request {i+1}: ✅ Allowed")
    else:
        print(f"Request {i+1}: ❌ Rejected")
        # Gọi HolySheep khi được phép
        # call_holysheep_api(prompt)

Adaptive Rate Limiting - Tự động điều chỉnh

Adaptive rate limiting thay đổi limit dựa trên tải hệ thống. Khi server load cao, tự động giảm rate limit. Khi load thấp, cho phép burst nhiều hơn. HolySheep sử dụng thuật toán này để đảm bảo SLA.

class AdaptiveRateLimiter:
    """
    Adaptive Rate Limiter - Tự động điều chỉnh theo tải hệ thống
    Sử dụng trong multi-tenant systems
    """
    
    def __init__(
        self,
        base_rate: int = 100,
        min_rate: int = 10,
        max_rate: int = 500,
        window_seconds: int = 60,
        target_latency_ms: float = 100.0
    ):
        self.base_rate = base_rate
        self.min_rate = min_rate
        self.max_rate = max_rate
        self.window_seconds = window_seconds
        self.target_latency = target_latency_ms
        
        self.current_rate = base_rate
        self.requests_in_window = 0
        self.latencies = []
        self.window_start = time.time()
        self.lock = threading.Lock()
    
    def _adjust_rate(self, current_latency: float):
        """Điều chỉnh rate dựa trên latency hiện tại"""
        # Tăng rate nếu latency thấp
        if current_latency < self.target_latency * 0.5:
            self.current_rate = min(self.max_rate, self.current_rate * 1.2)
        # Giảm rate nếu latency cao
        elif current_latency > self.target_latency:
            self.current_rate = max(self.min_rate, self.current_rate * 0.8)
    
    def _reset_window(self):
        """Reset counter cho window mới"""
        self.window_start = time.time()
        self.requests_in_window = 0
        self.latencies = []
    
    def record_request(self, latency_ms: float):
        """Ghi nhận latency của request vừa xử lý"""
        with self.lock:
            self.latencies.append(latency_ms)
            
            # Tính latency trung bình sau mỗi 10 requests
            if len(self.latencies) >= 10:
                avg_latency = sum(self.latencies) / len(self.latencies)
                self._adjust_rate(avg_latency)
                self.latencies = []
    
    def allow_request(self) -> tuple[bool, dict]:
        """Kiểm tra request và trả về thông tin rate limit"""
        with self.lock:
            elapsed = time.time() - self.window_start
            
            # Reset window nếu hết giờ
            if elapsed >= self.window_seconds:
                self._reset_window()
            
            # Tính effective rate (proportional)
            effective_limit = int(self.current_rate * (1 - elapsed / self.window_seconds))
            effective_limit = max(effective_limit, self.min_rate)
            
            if self.requests_in_window < effective_limit:
                self.requests_in_window += 1
                return True, {
                    "current_rate": self.current_rate,
                    "remaining": effective_limit - self.requests_in_window,
                    "reset_in": self.window_seconds - elapsed
                }
            
            return False, {
                "current_rate": self.current_rate,
                "remaining": 0,
                "reset_in": self.window_seconds - elapsed
            }



=== TRIỂN KHAI VỚI ASYNC ===

import asyncio
import time as sync_time

class AsyncAdaptiveLimiter:
    """Phiên bản async cho high-throughput systems"""
    
    def __init__(self, base_rate: int = 100, window_seconds: int = 60):
        self.base_rate = base_rate
        self.window_seconds = window_seconds
        self.current_rate = base_rate
        self.tokens = float(base_rate)
        self.last_update = sync_time.time()
        self.lock = asyncio.Lock()
    
    async def acquire(self) -> bool:
        """Lấy token cho request (blocking)"""
        async with self.lock:
            now = sync_time.time()
            elapsed = now - self.last_update
            
            # Refill tokens
            self.tokens = min(self.base_rate, self.tokens + elapsed * (self.base_rate / self.window_seconds))
            self.last_update = now
            
            if self.tokens >= 1:
                self.tokens -= 1
                return True
            
            return False
    
    async def wait_and_acquire(self):
        """Đợi cho đến khi có token"""
        while not await self.acquire():
            wait_time = 1.0 / (self.base_rate / self.window_seconds)
            await asyncio.sleep(wait_time)



Sử dụng với async HolySheep API calls

async def process_with_holysheep(limiter: AsyncAdaptiveLimiter, prompts: list):
    results = []
    
    for prompt in prompts:
        await limiter.wait_and_acquire()  # Đợi nếu cần
        
        start = sync_time.time()
        response = await call_holysheep_async(prompt)
        latency = (sync_time.time() - start) * 1000
        
        results.append(response)
    
    return results

Triển khai Complete Rate Limiter Middleware

Đây là code production-ready tôi dùng trong production của HolySheep, hỗ trợ Redis cluster cho distributed deployment.

"""
Complete Rate Limiter Middleware cho FastAPI
Hỗ trợ: Token Bucket, Sliding Window, Multi-tenant
"""
from fastapi import FastAPI, Request, HTTPException
from fastapi.responses import JSONResponse
from redis.cluster import RedisCluster
from typing import Optional
import json
import time

app = FastAPI()


Redis Cluster configuration

REDIS_NODES = [
    {"host": "redis-1.holysheep.ai", "port": 6379},
    {"host": "redis-2.holysheep.ai", "port": 6379},
]
redis_cluster = RedisCluster(startup_nodes=REDIS_NODES, decode_responses=True)

class DistributedRateLimiter:
    """
    Rate Limiter phân tán với Redis Cluster
    Hỗ trợ Token Bucket algorithm với Lua script atomic
    """
    
    LUA_SCRIPT = """
    local key = KEYS[1]
    local capacity = tonumber(ARGV[1])
    local refill_rate = tonumber(ARGV[2])
    local requested = tonumber(ARGV[3])
    local now = tonumber(ARGV[4])
    
    -- Lấy state hiện tại
    local bucket = redis.call('HMGET', key, 'tokens', 'last_update')
    local tokens = tonumber(bucket[1]) or capacity
    local last_update = tonumber(bucket[2]) or now
    
    -- Tính tokens cần refill
    local elapsed = now - last_update
    local new_tokens = math.min(capacity, tokens + (elapsed * refill_rate))
    
    -- Kiểm tra và lấy tokens
    if new_tokens >= requested then
        new_tokens = new_tokens - requested
        redis.call('HMSET', key, 'tokens', new_tokens, 'last_update', now)
        redis.call('EXPIRE', key, 3600)
        return {1, new_tokens}  -- allowed
    else
        redis.call('HMSET', key, 'tokens', new_tokens, 'last_update', now)
        redis.call('EXPIRE', key, 3600)
        return {0, new_tokens}  -- rejected
    end
    """
    
    def __init__(self, redis_client):
        self.redis = redis_client
        self.script = redis_client.register_script(self.LUA_SCRIPT)
    
    async def check_rate_limit(
        self,
        identifier: str,
        capacity: int = 100,
        refill_rate: float = 10.0,
        requested: int = 1
    ) -> dict:
        """Kiểm tra rate limit cho identifier"""
        key = f"ratelimit:token_bucket:{identifier}"
        now = time.time()
        
        result = self.script(
            keys=[key],
            args=[capacity, refill_rate, requested, now]
        )
        
        allowed = bool(result[0])
        remaining_tokens = float(result[1])
        
        return {
            "allowed": allowed,
            "remaining": int(remaining_tokens),
            "limit": capacity,
            "reset_in": (capacity - remaining_tokens) / refill_rate if not allowed else 0
        }



Khởi tạo rate limiter

rate_limiter = DistributedRateLimiter(redis_cluster)


Tenant configurations

TENANT_CONFIGS = {
    "free_tier": {"capacity": 60, "refill_rate": 1.0},      # 60 requests/phút
    "pro_tier": {"capacity": 600, "refill_rate": 10.0},     # 600 requests/phút  
    "enterprise": {"capacity": 6000, "refill_rate": 100.0},  # 6000 requests/phút
}

@app.middleware("http")
async def rate_limit_middleware(request: Request, call_next):
    """Middleware kiểm tra rate limit cho mọi request"""
    
    # Lấy API key từ header
    api_key = request.headers.get("Authorization", "").replace("Bearer ", "")
    
    if not api_key or api_key == "YOUR_HOLYSHEEP_API_KEY":
        return await call_next(request)
    
    # Xác định tier từ API key (trong thực tế, query từ database)
    tier = "free_tier"
    config = TENANT_CONFIGS.get(tier, TENANT_CONFIGS["free_tier"])
    
    # Check rate limit
    result = await rate_limiter.check_rate_limit(
        identifier=api_key,
        capacity=config["capacity"],
        refill_rate=config["refill_rate"]
    )
    
    # Thêm headers vào response
    response = await call_next(request)
    response.headers["X-RateLimit-Limit"] = str(config["capacity"])
    response.headers["X-RateLimit-Remaining"] = str(result["remaining"])
    response.headers["X-RateLimit-Reset"] = str(int(time.time() + result["reset_in"]))
    
    if not result["allowed"]:
        return JSONResponse(
            status_code=429,
            content={
                "error": "Too Many Requests",
                "message": "Rate limit exceeded",
                "retry_after": int(result["reset_in"]) + 1
            },
            headers=response.headers
        )
    
    return response


@app.post("/v1/chat/completions")
async def chat_completions(request: Request):
    """Proxy endpoint cho HolySheep AI - đã có rate limiting"""
    
    body = await request.json()
    
    # Forward request đến HolySheep
    async with aiohttp.ClientSession() as session:
        async with session.post(
            "https://api.holysheep.ai/v1/chat/completions",
            headers={
                "Authorization": f"Bearer YOUR_HOLYSHEEP_API_KEY",
                "Content-Type": "application/json"
            },
            json=body
        ) as resp:
            return await resp.json()


if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

So sánh HolySheep vs Dịch vụ khác

Tiêu chíHolySheep AIAPI chính thứcRelay service ARelay service B
GPT-4.1$8/MTok$60/MTok$15/MTok$20/MTok
Claude Sonnet 4.5$15/MTok$18/MTok$16/MTok$17/MTok
Gemini 2.5 Flash$2.50/MTok$2.50/MTok$3.00/MTok$4.00/MTok
DeepSeek V3.2$0.42/MTokKhông hỗ trợ$0.80/MTok$1.00/MTok
Thanh toánWeChat/Alipay/VisaChỉ VisaVisa/PayPalChỉ PayPal
Độ trễ trung bình<50ms100-300ms80-150ms120-200ms
Rate limit mặc định100 RPM3 RPM (free)50 RPM30 RPM
Tín dụng miễn phí✅ Có❌ Không❌ Không❌ Không
Tỷ giá¥1 ≈ $1Quy đổi USDUSDUSD
Tiết kiệm vs chính thức85%+Baseline60-75%50-70%

Phù hợp với ai

Giá và ROI

Với volume sử dụng trung bình

Tài nguyên liên quan

Bài viết liên quan