von Thomas Brenner, Senior Backend Engineer bei HolySheep AI
Einleitung: Wenn der Black Friday zum Albtraum wird
Es war der 27. November 2024, 23:47 Uhr. Unser E-Commerce-Kunde – ein deutsches Fashion-Unternehmen mit 2,3 Millionen aktiven Nutzern – stand kurz vor dem größten Sale-Event des Jahres. Plötzlich meldete das Monitoring: API-Response-Zeiten von 12 Sekunden, Fehlerraten bei 34%, und im Chat-System stapelten sich 8.400 unzustellbare Nachrichten. Der Grund? Ein konkurrierender Dienst hatte unbemerkt die Rate seiner KI-Chat-Integration erhöht, und unser primitives Fixed-Window-Limiter knickte unter der Last ein.
Dieser Vorfall kostete uns drei Stunden Reaktionszeit und etwa 2.400 € an verlorenen Conversion-Umsätzen. Die Lektion war klar: Rate Limiting ist keine Nebensache – es ist die Lebensversicherung jeder skalierbaren API-Infrastruktur. In diesem Tutorial zeige ich Ihnen alle vier klassischen Algorithmen mit praxistauglichem Code und erkläre, warum wir bei HolySheep AI einen Hybriden aus Token Bucket und Sliding Window einsetzen.
Was ist API Rate Limiting und warum ist es kritisch?
Rate Limiting (Ratenbegrenzung) kontrolliert, wie viele Anfragen ein Client in einem bestimmten Zeitraum stellen darf. Die Hauptziele sind:
- Systemstabilität: Verhindern von Überlastung bei Lastspitzen
- Kostenkontrolle: Begrenzung des Ressourcenverbrauchs bei Cloud-APIs
- Fairness: Gleichmäßige Verteilung der Kapazität zwischen Nutzern
- Missbrauchsschutz: Abwehr von DDoS-Angriffen und Fehlern in Client-Software
Bei HolySheep AI sehen wir täglich über 180 Millionen API-Aufrufe. Ohne effektives Rate Limiting wäre unser Dienst innerhalb von Minuten überlastet – besonders während der Spitzenzeiten um 9:00 Uhr (MEZ) und 14:00 Uhr (MEZ), wenn europäische Entwickler gleichzeitig auf unsere KI-Modelle zugreifen.
Die vier Grundstrategien im Detail
1. Fixed Window Limiter (Feste Fenster)
Der Fixed Window Algorithmus ist der einfachste Ansatz: Wir teilen die Zeit in gleichmäßige Fenster (z.B. 60 Sekunden) und erlauben maximal N Anfragen pro Fenster.
Funktionsweise:
Fenster 1: 00:00 - 00:59 |████████████░░░░░| 8/10 Requests ✓
Fenster 2: 01:00 - 01:59 |░░░░░░░░░░░░░░░░| 0/10 Requests (Limit erreicht um 00:47)
Fenster 3: 02:00 - 02:59 |████████████| 10/10 Requests ✓Python-Implementierung:
"""Fixed Window Rate Limiter mit Redis"""
import time
import redis
from datetime import datetime
class FixedWindowLimiter:
def __init__(self, max_requests: int, window_seconds: int, redis_client=None):
self.max_requests = max_requests
self.window_seconds = window_seconds
self.redis = redis_client or redis.Redis(host='localhost', port=6379, decode_responses=True)
def is_allowed(self, client_id: str) -> tuple[bool, dict]:
"""Prüft ob Request erlaubt ist und aktualisiert Zähler"""
current_window = int(time.time() // self.window_seconds)
key = f"ratelimit:fixed:{client_id}:{current_window}"
current_count = self.redis.get(key)
if current_count is None:
# Neues Fenster starten
self.redis.setex(key, self.window_seconds * 2, 1)
return True, {"remaining": self.max_requests - 1, "reset": self.window_seconds}
current_count = int(current_count)
if current_count >= self.max_requests:
ttl = self.redis.ttl(key)
return False, {
"error": "Rate limit exceeded",
"retry_after": ttl if ttl > 0 else self.window_seconds,
"limit": self.max_requests
}
self.redis.incr(key)
return True, {"remaining": self.max_requests - current_count - 1}
def get_usage(self, client_id: str) -> dict:
"""Aktuellen Verbrauch abrufen"""
current_window = int(time.time() // self.window_seconds)
key = f"ratelimit:fixed:{client_id}:{current_window}"
current = self.redis.get(key)
return {
"used": int(current) if current else 0,
"limit": self.max_requests,
"window_seconds": self.window_seconds
}
Beispiel-Nutzung mit HolySheep AI API
def call_holysheep_api(user_id: str, message: str):
limiter = FixedWindowLimiter(max_requests=100, window_seconds=60)
allowed, info = limiter.is_allowed(user_id)
if not allowed:
print(f"⚠️ Rate limit erreicht. Retry nach {info['retry_after']}s")
return None
import requests
response = requests.post(
"https://api.holysheep.ai/v1/chat/completions",
headers={
"Authorization": f"Bearer YOUR_HOLYSHEEP_API_KEY",
"Content-Type": "application/json"
},
json={
"model": "deepseek-v3.2",
"messages": [{"role": "user", "content": message}],
"max_tokens": 500
}
)
print(f"✅ Anfrage erfolgreich. Verbleibend: {info['remaining']}")
return response.json()2. Sliding Window Limiter (Schiebe Fenster)
Das Problem von Fixed Window: Ein Client kann 100 Requests um 23:59 Uhr und weitere 100 Requests um 00:01 Uhr senden – also 200 Requests in 2 Minuten, obwohl das Limit bei 100/min liegt. Der Sliding Window Algorithmus löst dies, indem er das Fenster kontinuierlich verschiebt.
Funktionsweise mit Zeitstempel-Liste:
Client-Anfrage um 12:00:30
┌─────────────────────────────────────────────────────────┐
│ Vergangene Requests (letzte 60 Sekunden): │
│ 11:59:45 │ 12:00:05 │ 12:00:15 │ 12:00:22 │ 12:00:28 │
│ ✓ │ ✓ │ ✓ │ ✓ │ ✓ │
│ │
│ Prüfung: 5 Requests inkl. aktuelle ≤ 10? → ERLAUBT ✓ │
└─────────────────────────────────────────────────────────┘Python-Implementierung mit Redis Sorted Sets:
"""Sliding Window Rate Limiter - Production Ready"""
import time
import redis
from typing import Optional
import json
class SlidingWindowLimiter:
"""
Implements sliding window log algorithm using Redis sorted sets.
More accurate than fixed window but slightly higher memory usage.
"""
def __init__(
self,
redis_url: str = "redis://localhost:6379",
max_requests: int = 100,
window_seconds: int = 60
):
self.redis = redis.from_url(redis_url)
self.max_requests = max_requests
self.window_seconds = window_seconds
def is_allowed(self, client_id: str) -> dict:
"""
Returns detailed rate limit information.
"""
now = time.time()
window_start = now - self.window_seconds
key = f"ratelimit:sliding:{client_id}"
pipe = self.redis.pipeline()
# 1. Alte Einträge außerhalb des Fensters löschen
pipe.zremrangebyscore(key, 0, window_start)
# 2. Aktuelle Anzahl im Fenster zählen
pipe.zcard(key)
# 3. Request hinzufügen (mit aktuellem Timestamp als Score)
pipe.zadd(key, {f"{now}": now})
# 4. TTL auf Fenstergröße setzen (Aufräumen)
pipe.expire(key, self.window_seconds + 1)
results = pipe.execute()
current_count = results[1] # Ergebnis von zcard
if current_count >= self.max_requests:
# Request ablehnen, aber den hinzugefügten Eintrag wieder entfernen
self.redis.zrem(key, f"{now}")
# Nächsten Request-Time berechnen
oldest = self.redis.zrange(key, 0, 0, withscores=True)
if oldest:
retry_after = int(oldest[0][1] + self.window_seconds - now) + 1
else:
retry_after = self.window_seconds
return {
"allowed": False,
"limit": self.max_requests,
"remaining": 0,
"retry_after": max(1, retry_after),
"reset": int(now + retry_after),
"total_in_window": current_count
}
return {
"allowed": True,
"limit": self.max_requests,
"remaining": self.max_requests - current_count - 1,
"reset": int(now + self.window_seconds),
"total_in_window": current_count + 1
}
def get_window_info(self, client_id: str) -> dict:
"""Detaillierte Fenster-Statistik"""
now = time.time()
window_start = now - self.window_seconds
key = f"ratelimit:sliding:{client_id}"
# Nur Einträge im aktuellen Fenster
current_entries = self.redis.zrangebyscore(
key, window_start, now, withscores=True
)
return {
"requests": len(current_entries),
"limit": self.max_requests,
"window_seconds": self.window_seconds,
"utilization_percent": round(len(current_entries) / self.max_requests * 100, 1),
"oldest_request": current_entries[0][1] if current_entries else None,
"newest_request": current_entries[-1][1] if current_entries else None
}
Production-Integration mit HolySheep API
class HolySheepAPIClient:
"""Rate-limited HolySheep AI Client"""
def __init__(self, api_key: str, requests_per_minute: int = 60):
self.base_url = "https://api.holysheep.ai/v1"
self.api_key = api_key
self.limiter = SlidingWindowLimiter(
max_requests=requests_per_minute,
window_seconds=60
)
self.session = requests.Session()
self.session.headers.update({
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
})
def chat(self, message: str, model: str = "deepseek-v3.2") -> Optional[dict]:
"""Chat Completions mit automatischem Rate Limiting"""
rate_info = self.limiter.is_allowed(f"chat-{model}")
if not rate_info["allowed"]:
print(f"⏳ Rate limit erreicht. Warte {rate_info['retry_after']}s...")
time.sleep(rate_info["retry_after"])
rate_info = self.limiter.is_allowed(f"chat-{model}")
response = self.session.post(
f"{self.base_url}/chat/completions",
json={
"model": model,
"messages": [{"role": "user", "content": message}],
"temperature": 0.7,
"max_tokens": 1000
},
timeout=30
)
return {
"data": response.json(),
"rate_limit": rate_info
}
Beispiel: Batch-Verarbeitung mit Ratelimit-Handhabung
def process_user_queries(queries: list, api_key: str):
"""Verarbeitet Queries mit intelligentem Rate Limiting"""
client = HolySheepAPIClient(api_key, requests_per_minute=120)
results = []
for i, query in enumerate(queries):
result = client.chat(query)
results.append(result)
# Progress-Logging alle 10 Requests
if (i + 1) % 10 == 0:
info = client.limiter.get_window_info(f"chat-deepseek-v3.2")
print(f"📊 Fortschritt: {i+1}/{len(queries)} | "
f"Auslastung: {info['utilization_percent']}%")
return results3. Leaky Bucket Limiter (漏桶算法)
Der Leaky Bucket (Undichtes Fass) arbeitet nach dem Prinzip einer Wasseruhr: Anfragen werden in einen virtuellen Eimer gegeben und sickern mit konstanter Rate durch. Der Vorteil: Konstante Output-Rate, unabhängig von Burst-Last.
Mathematisches Modell:
┌──────────────────────────────────────────────────────┐
│ Input-Rate variabel: Output-Rate konstant: │
│ │
│ ████ │
│ ████████████ ┌─────┐ ░░░░░░░░░ (5/s) │
│ ███████████████──┤Bucket├──►░░░░░░░░░ │
│ ████████████████└─────┘ ░░░░░░░░░ │
│ ████████ ░░░░░░░░░ │
│ ░░░░░░░░░ │
│ │
│ Bei Überlauf: Request wird verworfen │
└──────────────────────────────────────────────────────┘
Formel:
Letzter_Leak_Zeitpunkt + (Bucket_Füllung / Leak_Rate) = Max_Wartezeit
Bei 1000 Requests × 10ms pro Leak = 10 Sekunden max. VerzögerungPython-Implementierung:
"""Leaky Bucket Rate Limiter mit asyncio Support"""
import time
import asyncio
from typing import Optional
from dataclasses import dataclass
from collections import deque
import threading
@dataclass
class LeakyBucketConfig:
"""Konfiguration für den Leaky Bucket Algorithmus"""
capacity: int = 100 # Maximale Bucket-Größe
leak_rate: float = 10.0 # Leak-Rate in Requests/Sekunde
max_wait: float = 30.0 # Maximale Wartezeit in Sekunden
class LeakyBucketLimiter:
"""
Thread-safe Leaky Bucket Implementation.
Garantiert konstante Output-Rate, puffert Bursts.
"""
def __init__(self, config: LeakyBucketConfig = None):
self.config = config or LeakyBucketConfig()
self._lock = threading.RLock()
self._buckets: dict[str, float] = {} # client_id -> letzte Leak-Zeit
self._last_request_time: dict[str, float] = {}
def _get_bucket_level(self, client_id: str) -> float:
"""Berechnet aktuelles Bucket-Fülllevel in Requests"""
now = time.time()
if client_id not in self._buckets:
self._buckets[client_id] = 0.0
self._last_request_time[client_id] = now
return 0.0
last_time = self._last_request_time[client_id]
elapsed = now - last_time
# "Leaken": Füllstand um vergangene Zeit reduzieren
leaked = elapsed * self.config.leak_rate
self._buckets[client_id] = max(0.0, self._buckets[client_id] - leaked)
self._last_request_time[client_id] = now
return self._buckets[client_id]
def acquire(self, client_id: str, tokens: int = 1) -> tuple[bool, float]:
"""
Versucht Request zu akquirieren.
Returns:
(erfolgreich, wartezeit)
"""
with self._lock:
current_level = self._get_bucket_level(client_id)
if current_level + tokens <= self.config.capacity:
# Genug Platz: Request durchlassen
self._buckets[client_id] = current_level + tokens
return True, 0.0
# Nicht genug Platz: Wartezeit berechnen
overflow = (current_level + tokens) - self.config.capacity
wait_time = overflow / self.config.leak_rate
if wait_time > self.config.max_wait:
return False, wait_time
# Warten und dann akquirieren
time.sleep(wait_time)
self._buckets[client_id] = self.config.capacity
return True, wait_time
async def async_acquire(self, client_id: str, tokens: int = 1) -> tuple[bool, float]:
"""Async-Version für asyncio-basierte Anwendungen"""
loop = asyncio.get_event_loop()
with self._lock:
current_level = self._get_bucket_level(client_id)
if current_level + tokens <= self.config.capacity:
self._buckets[client_id] = current_level + tokens
return True, 0.0
overflow = (current_level + tokens) - self.config.capacity
wait_time = overflow / self.config.leak_rate
if wait_time > self.config.max_wait:
return False, wait_time
# Außerhalb des Locks warten
await asyncio.sleep(wait_time)
with self._lock:
self._buckets[client_id] = self.config.capacity
return True, wait_time
def get_status(self, client_id: str) -> dict:
"""Aktueller Bucket-Status"""
with self._lock:
level = self._get_bucket_level(client_id)
return {
"client_id": client_id,
"current_level": round(level, 2),
"capacity": self.config.capacity,
"leak_rate": self.config.leak_rate,
"utilization": round(level / self.config.capacity * 100, 1),
"estimated_empty_in": round(
level / self.config.leak_rate, 2
) if level > 0 else 0
}
Async Production-Example mit HolySheep
class AsyncHolySheepClient:
"""Hochleistungs-Async Client mit Leaky Bucket Rate Limiting"""
def __init__(self, api_key: str, rpm: int = 60):
self.base_url = "https://api.holysheep.ai/v1"
self.api_key = api_key
self.limiter = LeakyBucketLimiter(
config=LeakyBucketConfig(
capacity=rpm,
leak_rate=rpm / 60.0, # Requests pro Sekunde
max_wait=30.0
)
)
async def chat_async(self, message: str, model: str = "deepseek-v3.2") -> dict:
"""Asynchroner Chat-Aufruf mit automatischem Rate Limiting"""
success, wait_time = await self.limiter.async_acquire("global")
if not success:
raise Exception(f"Rate limit timeout: {wait_time:.1f}s")
async with aiohttp.ClientSession() as session:
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
async with session.post(
f"{self.base_url}/chat/completions",
headers=headers,
json={
"model": model,
"messages": [{"role": "user", "content": message}]
},
timeout=aiohttp.ClientTimeout(total=30)
) as response:
return await response.json()
Beispiel: Rate-limited Batch Processing
async def batch_process_queries(queries: list, api_key: str):
"""Parallele Batch-Verarbeitung mit Leaky Bucket Control"""
client = AsyncHolySheepClient(api_key, rpm=120)
tasks = []
for query in queries:
task = client.chat_async(query)
tasks.append(task)
# Semaphore für max. 10 gleichzeitige Requests
semaphore = asyncio.Semaphore(10)
async def bounded_task(task):
async with semaphore:
return await task
bounded_tasks = [bounded_task(t) for t in tasks]
results = await asyncio.gather(*bounded_tasks, return_exceptions=True)
return results4. Token Bucket Limiter (令牌桶)
Der Token Bucket ist der Algorithmus, den die meisten modernen APIs verwenden – einschließlich HolySheep AI. Das Prinzip: Ein virtueller Eimer enthält Tokens. Jeder Request "verbraucht" ein Token. Tokens werden mit konstanter Rate nachgefüllt. Der Clou: Auch Bursts sind erlaubt, solange Tokens vorhanden sind.
Unterschied zu Leaky Bucket:
Leaky Bucket: Feste Output-Rate, keine Bursts möglich
Token Bucket: Erlaubt Bursts bis zur Bucket-Kapazität
Beispiel bei 10 Tokens/sec, Bucket=100:
Szenario A (Burst):
100 Requests sofort → Alle sofort bedient (Buckets voll)
Danach 10 Requests/sec → Konstante Rate
Szenario B (Sustained):
10 Requests/sec → Funktioniert für immer (Tokens werden aufgefüllt)
┌────────────────────────────────────────────────────┐
│ Token Bucket bei HolySheep: │
│ • Bucket-Größe: 60 Tokens (1 Minute Puffer) │
│ • Refill-Rate: 1 Token/Sekunde │
│ • Erlaubt: 60/min normal, 120/min im Burst │
└────────────────────────────────────────────────────┘Production-Ready Python-Implementierung:
"""Token Bucket Rate Limiter - Der Industry Standard"""
import time
import threading
import math
from typing import Optional
from dataclasses import dataclass, field
@dataclass
class TokenBucketState:
"""Zustand eines einzelnen Token Buckets"""
tokens: float
last_update: float
locked: bool = False
class TokenBucketLimiter:
"""
Thread-safe Token Bucket Implementation.
Vorteile gegenüber anderen Algorithmen:
- Erlaubt Bursts (bis Bucket-Kapazität)
- Konstant-genaue Rate (keine Window-Artefakte)
- Speicher-effizient (nur 2 Werte pro Client)
"""
def __init__(
self,
bucket_size: int = 60,
refill_rate: float = 1.0, # Tokens pro Sekunde
initial_tokens: Optional[float] = None
):
self.bucket_size = bucket_size
self.refill_rate = refill_rate
self.initial_tokens = initial_tokens or bucket_size
self._buckets: dict[str, TokenBucketState] = {}
self._lock = threading.RLock()
def _get_state(self, client_id: str) -> TokenBucketState:
"""Holt oder erstellt Bucket-State für Client"""
if client_id not in self._buckets:
self._buckets[client_id] = TokenBucketState(
tokens=self.initial_tokens,
last_update=time.time()
)
return self._buckets[client_id]
def _refill(self, state: TokenBucketState) -> float:
"""Füllt Tokens basierend auf vergangener Zeit auf"""
now = time.time()
elapsed = now - state.last_update
new_tokens = min(
self.bucket_size,
state.tokens + (elapsed * self.refill_rate)
)
state.tokens = new_tokens
state.last_update = now
return new_tokens
def try_acquire(self, client_id: str, tokens: int = 1) -> bool:
"""
Versucht Tokens zu akquirieren ohne zu blockieren.
Returns:
True wenn akquiriert, False wenn nicht genug Tokens
"""
with self._lock:
state = self._get_state(client_id)
self._refill(state)
if state.tokens >= tokens:
state.tokens -= tokens
return True
return False
def acquire(
self,
client_id: str,
tokens: int = 1,
blocking: bool = True,
timeout: Optional[float] = None
) -> tuple[bool, float]:
"""
Akquiriert Tokens mit optionalem Warten.
Args:
client_id: Eindeutige Client-ID
tokens: Anzahl benötigter Tokens
blocking: Ob blockieren erlaubt ist
timeout: Maximale Wartezeit in Sekunden
Returns:
(erfolgreich, gewartete_zeit)
"""
start_time = time.time()
waited = 0.0
while True:
with self._lock:
state = self._get_state(client_id)
self._refill(state)
if state.tokens >= tokens:
state.tokens -= tokens
return True, waited
if not blocking:
return False, waited
# Berechne Wartezeit bis genug Tokens
tokens_needed = tokens - state.tokens
wait_time = math.ceil(tokens_needed / self.refill_rate * 1000) / 1000
# Prüfe Timeout
waited += wait_time
if timeout and waited >= timeout:
return False, waited
time.sleep(min(wait_time, 0.1)) # Max 100ms pro Sleep
def get_available_tokens(self, client_id: str) -> float:
"""Gibt verfügbare Tokens für Client zurück"""
with self._lock:
state = self._get_state(client_id)
return self._refill(state)
def get_status(self, client_id: str) -> dict:
"""Detaillierter Status für Monitoring"""
with self._lock:
state = self._get_state(client_id)
self._refill(state)
return {
"client_id": client_id,
"available_tokens": round(state.tokens, 2),
"bucket_size": self.bucket_size,
"refill_rate": self.refill_rate,
"utilization_percent": round(
(1 - state.tokens / self.bucket_size) * 100, 1
),
"tokens_per_minute": self.refill_rate * 60,
"burst_capacity": self.bucket_size
}
HolySheep-spezifische Implementation
class HolySheepTokenLimiter:
"""
Token Bucket speziell für HolySheep AI API konfiguriert.
Unterstützt verschiedene Tier-Level.
"""
TIERS = {
"free": {
"bucket_size": 60,
"refill_rate": 1.0, # 60/min
"models": ["deepseek-v3.2"]
},
"pro": {
"bucket_size": 600,
"refill_rate": 10.0, # 600/min
"models": ["deepseek-v3.2", "gpt-4.1", "claude-sonnet-4.5"]
},
"enterprise": {
"bucket_size": 6000,
"refill_rate": 100.0, # 6000/min
"models": ["*"] # Alle Modelle
}
}
def __init__(self, tier: str = "free"):
if tier not in self.TIERS:
raise ValueError(f"Unbekannte Tier: {tier}")
config = self.TIERS[tier]
self.limiter = TokenBucketLimiter(
bucket_size=config["bucket_size"],
refill_rate=config["refill_rate"]
)
self.tier = tier
self.models = config["models"]
def is_request_allowed(self, client_id: str, model: str) -> tuple[bool, dict]:
"""Prüft ob Request erlaubt ist und gibt Metriken zurück"""
if model not in self.models and "*" not in self.models:
return False, {
"error": f"Model {model} nicht verfügbar für Tier {self.tier}",
"required_tier": "pro" if model in self.TIERS["pro"]["models"] else "enterprise"
}
allowed, waited = self.limiter.acquire(client_id, tokens=1, blocking=False)
status = self.limiter.get_status(client_id)
return allowed, {
**status,
"waited_seconds": waited,
"tier": self.tier
}
Monitoring-Decorator für API-Endpunkte
def rate_limited(tier: str = "free"):
"""Decorator für rate-limited API-Endpunkte"""
limiter = HolySheepTokenLimiter(tier)
def decorator(func):
def wrapper(client_id: str, *args, **kwargs):
allowed, info = limiter.is_request_allowed(client_id, kwargs.get("model", "deepseek-v3.2"))
if not allowed:
return {
"error": "rate_limit_exceeded",
"message": info.get("error", "Zu viele Anfragen"),
"retry_after": math.ceil(
(1 - info.get("available_tokens", 0)) / limiter.limiter.refill_rate
)
}
result = func(*args, **kwargs)
return {
"data": result,
"rate_limit_info": info
}
return wrapper
return decorator
Beispiel: Production API Endpoint
@rate_limited(tier="pro")
def chat_endpoint(model: str, messages: list, client_id: str):
"""Beispiel für rate-limited Chat-Endpoint"""
import requests
response = requests.post(
"https://api.holysheep.ai/v1/chat/completions",
headers={"Authorization": f"Bearer YOUR_HOLYSHEEP_API_KEY"},
json={
"model": model,
"messages": messages,
"temperature": 0.7,
"max_tokens": 2000
},
timeout=30
)
return response.json()Vergleichstabelle: Alle vier Algorithmen
| Kriterium | Fixed Window | Sliding Window | Leaky Bucket | Token Bucket |
|---|---|---|---|---|
| Burst-Toleranz | ❌ Nein | ⚠️ Begrenzt | ❌ Nein | ✅ Ja (bis Kapazität) |
| Output-Rate | Variabel | Variabel | ✅ Konstant | ✅ Konstant (im Mittel) |
| Speicher-Komplexität | ✅ O(1) | ⚠️ O(n) | ✅ O(1) | ✅ O(1) |
| Rate-Genauigkeit | ⚠️ Fenster-Artefakte | ✅ Exakt | ✅ Exakt | ✅ Exakt |
| Implementierungs-Komplexität | ✅ Einfach | ⚠️ Mittel | ⚠️ Mittel | ⚠️ Mittel |
| Redis-Datentyp | String/Counter | Sorted Set | String + Lua | String + Lua |
| typische Latenz | <1ms | <2ms | <3ms | <2ms |
| Geeignet für | Einfache APIs, Development | Premium APIs, strenge Limits | Medien-Streaming, IoT | Standard für moderne APIs |
Geeignet / nicht geeignet für
✅ Fixed Window – Ideal für:
- Entwicklung und Prototyping mit einfachen Anforderungen
- APIs mit losen Rate-Limits (>100 Requests/Minute)
- Szenarien wo gelegentliche Grenzwertüberschreitung tolerierbar ist
- Internal Microservices mit vertrauenswürdigen Clients
❌ Fixed Window – Nicht geeignet für:
- Produktions-APIs mit strikten SLAs
- Szenarien mit Sicherheitsanforderungen (DDoS-Schutz)
- APIs mit variablen Mod
Verwandte Ressourcen
Verwandte Artikel