Als leitender Backend-Architekt habe ich in den letzten Jahren zahlreiche Produktionssysteme entwickelt, die tausende von AI-API-Anfragen pro Sekunde verarbeiten müssen. Die größte Herausforderung dabei ist nicht die reine Kapazität, sondern die faire und effiziente Verteilung von Ressourcen nach Priorität. In diesem Tutorial zeige ich Ihnen eine battle-getestete Architektur, die ich bei HolySheep AI Jetzt registrieren selbst im Einsatz habe – mit echten Benchmark-Daten und Kostenanalysen.
Warum eine Priority Queue für AI-APIs?
Bei der Arbeit mit AI-APIs wie der HolySheep API treten typische Herausforderungen auf: unterschiedliche SLA-Anforderungen (interaktive Nutzer vs. Batch-Jobs), variable Request-Komplexität und Kostenexplosion bei unbeaufsichtigtem Throughput. Eine Priority Queue ermöglicht:
- Garantierte Latenz für kritische Anfragen (z.B. Chat-UI)
- Faire Ressourcenverteilung nach Business Value
- Automatische Backpressure-Mechanismen
- Kostenkontrolle durch dynamische Rate-Limiting
Architektur-Übersicht
Unsere Architektur basiert auf einem dreistufigen Pipeline-Modell mit Redis als zentralem Knotenpunkt. Die Prioritätsstufen werden als Integer-Werte definiert, wobei niedrigere Werte höhere Priorität bedeuten.
┌─────────────────────────────────────────────────────────────────┐
│ REQUEST INGESTION LAYER │
│ ┌─────────┐ ┌─────────┐ ┌─────────┐ ┌─────────┐ │
│ │ Priority│ │ Priority│ │ Priority│ │ Priority│ │
│ │ P0 │ │ P1 │ │ P2 │ │ P3 │ │
│ │ (Critical)│ │(Standard)│ │ (Batch) │ │ (Bulk) │ │
│ └────┬────┘ └────┬────┘ └────┬────┘ └────┬────┘ │
│ │ │ │ │ │
│ └────────────┴────────────┴────────────┘ │
│ │ │
│ ┌────▼────┐ │
│ │ Redis │ │
│ │Priority │ │
│ │ Queue │ │
│ └────┬────┘ │
└─────────────────────────┼───────────────────────────────────────┘
│
┌─────────────────────────┼───────────────────────────────────────┐
│ PROCESSING LAYER │
│ ┌────▼────┐ │
│ │ Workers │ │
│ │ Pool N │ │
│ └────┬────┘ │
│ │ │
│ ┌──────────┼──────────┐ │
│ │ │ │ │
│ ┌──────▼──┐ ┌─────▼────┐ ┌───▼─────┐ │
│ │ Worker 1│ │Worker 2 │ │Worker N │ │
│ │ (P0-P1) │ │ (P1-P2) │ │(P2-P3) │ │
│ └─────────┘ └──────────┘ └─────────┘ │
└──────────────────────────────────────────────────────────────────┘Redis-basierte Priority Queue Implementierung
Die Kernkomponente ist eine Redis Sorted Set-basierte Queue, die O(log N) Einfüge- und Extraktionsoperationen ermöglicht. Wir nutzen die Kombination aus Score (für Priorität) und Timestamp (für FIFO innerhalb derselben Priorität).
import redis
import json
import time
import asyncio
from typing import Optional, Dict, Any
from dataclasses import dataclass, asdict
from enum import IntEnum
import hashlib
class RequestPriority(IntEnum):
"""Prioritätsstufen: Niedrigere Werte = Höhere Priorität"""
CRITICAL = 0 # Interaktive UI, SLA < 500ms
HIGH = 1 # Premium-User, SLA < 2s
NORMAL = 2 # Standard-User
LOW = 3 # Batch-Verarbeitung
BULK = 4 # Hintergrund-Jobs
@dataclass
class QueuedRequest:
request_id: str
priority: int
payload: Dict[str, Any]
created_at: float
user_id: str
callback_url: Optional[str] = None
def to_json(self) -> str:
return json.dumps({
'request_id': self.request_id,
'priority': self.priority,
'payload': self.payload,
'created_at': self.created_at,
'user_id': self.user_id,
'callback_url': self.callback_url
})
@staticmethod
def from_json(data: str) -> 'QueuedRequest':
parsed = json.loads(data)
return QueuedRequest(**parsed)
class AIRequestQueue:
"""
Priority-basierte Queue für AI-API-Anfragen.
Verwendet Redis Sorted Sets für O(log N) Operationen.
"""
QUEUE_KEY = "ai:priority:queue"
PROCESSING_KEY = "ai:processing:{}" # Format: ai:processing:{worker_id}
METRICS_KEY = "ai:metrics:requests"
def __init__(self, redis_url: str = "redis://localhost:6379"):
self.redis = redis.from_url(redis_url, decode_responses=True)
self._metrics_script = self.redis.register_script("""
redis.call('HINCRBY', KEYS[1], ARGV[1], 1)
redis.call('EXPIRE', KEYS[1], 3600)
""")
def enqueue(
self,
payload: Dict[str, Any],
priority: RequestPriority,
user_id: str,
request_id: Optional[str] = None,
callback_url: Optional[str] = None
) -> str:
"""
Fügt einen Request zur Priority Queue hinzu.
Der Score berechnet sich aus: priority * 1_000_000_000 + timestamp
Dies ermöglicht 1 Milliarde Requests pro Prioritätsstufe
mit korrekter FIFO-Ordnung.
"""
if request_id is None:
request_id = hashlib.sha256(
f"{user_id}{time.time()}{json.dumps(payload)}".encode()
).hexdigest()[:16]
timestamp = time.time()
# Score = Priorität (wichtig!) + Zeitstempel (für FIFO)
score = (priority * 1_000_000_000) + timestamp
request = QueuedRequest(
request_id=request_id,
priority=priority,
payload=payload,
created_at=timestamp,
user_id=user_id,
callback_url=callback_url
)
pipe = self.redis.pipeline()
# Sorted Set mit Score für Priorität + Zeitstempel
pipe.zadd(self.QUEUE_KEY, {request.to_json(): score})
# Hash für O(1) Lookup nach ID
pipe.hset(f"ai:request:{request_id}", mapping={
'status': 'queued',
'priority': priority,
'created_at': timestamp
})
pipe.execute()
# Metriken aktualisieren
self._metrics_script(keys=[self.METRICS_KEY], args=[f"priority_{priority}"])
return request_id
def dequeue(
self,
worker_id: str,
max_priority: RequestPriority = RequestPriority.BULK,
timeout: int = 5
) -> Optional[QueuedRequest]:
"""
Entnimmt den höchstpriorisierten Request aus der Queue.
Blockiert bis zu timeout Sekunden wenn keine Requests verfügbar.
"""
# Transaktion für atomare Operation
pipe = self.redis.pipeline()
# Zuerst: Versuche höchste Priorität zu bekommen
# Wir holen die besten N Requests (einer für uns, Rest als Reserve)
for _ in range(3): # Max 3 Retry-Versuche
# Atomare Operation: Zunächst Range lesen, dann entfernen
candidates = self.redis.zrange(
self.QUEUE_KEY,
0,
0, # Nur der erste (höchste Priorität)
withscores=True
)
if not candidates:
time.sleep(0.1) # Kurze Pause bei leerer Queue
continue
request_json, score = candidates[0]
request = QueuedRequest.from_json(request_json)
# Nur Requests bis zur maximal erlaubten Priorität
if request.priority > max_priority:
return None
# Atomares Remove-and-Add: Nur wenn Score noch aktuell
removed = self.redis.zrem(self.QUEUE_KEY, request_json)
if removed:
# Request erfolgreich entnommen
pipe.hset(f"ai:request:{request.request_id}", mapping={
'status': 'processing',
'worker_id': worker_id,
'processing_started_at': time.time()
})
# Temporary Set für in-Progress Requests
pipe.zadd(
self.PROCESSING_KEY.format(worker_id),
{request_json: time.time()}
)
pipe.execute()
return request
return None
def complete(self, request_id: str, worker_id: str, success: bool = True):
"""Markiert einen Request als abgeschlossen."""
pipe = self.redis.pipeline()
pipe.hset(f"ai:request:{request_id}", mapping={
'status': 'completed' if success else 'failed',
'completed_at': time.time()
})
pipe.delete(self.PROCESSING_KEY.format(worker_id))
pipe.execute()
def get_queue_stats(self) -> Dict[str, Any]:
"""Liefert aktuelle Queue-Statistiken."""
pipe = self.redis.pipeline()
for p in RequestPriority:
pipe.zcount(self.QUEUE_KEY, p * 1_000_000_000, (p + 1) * 1_000_000_000 - 0.001)
counts = pipe.execute()
return {
'total_queued': sum(counts),
'by_priority': {
p.name: counts[p] for p in RequestPriority
},
'timestamp': time.time()
}Worker-Pool mit Concurrency Control
Der Worker-Pool implementiert ein semaphorenbasiertes Concurrency-Limit, das verhindert, dass wir die API-Rate-Limits überschreiten. Bei HolySheep AI beträgt das Standard-Limit 1000 Requests pro Minute – wir limitieren unsere Workers auf 800 concurrent Requests für Headroom.
import asyncio
import aiohttp
import logging
from typing import Optional
from concurrent.futures import ThreadPoolExecutor
import signal
import sys
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class AIWorkerPool:
"""
Worker-Pool für AI-API Requests mit automatischer Retry-Logik
und Connection Pooling.
"""
def __init__(
self,
api_key: str,
base_url: str = "https://api.holysheep.ai/v1",
max_concurrent: int = 800,
max_retries: int = 3,
timeout_seconds: int = 60
):
self.api_key = api_key
self.base_url = base_url
self.max_concurrent = max_concurrent
self.max_retries = max_retries
self.timeout = aiohttp.ClientTimeout(total=timeout_seconds)
# Semaphore für Concurrency-Control
self._semaphore = asyncio.Semaphore(max_concurrent)
# Connection Pool für effiziente HTTP-Verbindungen
self._connector = aiohttp.TCPConnector(
limit=max_concurrent,
limit_per_host=100,
keepalive_timeout=30
)
self._session: Optional[aiohttp.ClientSession] = None
# Metriken
self._stats = {
'processed': 0,
'failed': 0,
'retried': 0,
'total_latency_ms': 0
}
async def __aenter__(self):
self._session = aiohttp.ClientSession(
connector=self._connector,
timeout=self.timeout,
headers={
'Authorization': f'Bearer {self.api_key}',
'Content-Type': 'application/json'
}
)
return self
async def __aexit__(self, exc_type, exc_val, exc_tb):
if self._session:
await self._session.close()
async def process_request(
self,
request: QueuedRequest,
model: str = "deepseek-v3.2"
) -> Dict[str, Any]:
"""
Verarbeitet einen einzelnen AI-API Request mit Retry-Logik.
"""
async with self._semaphore: # Concurrency-Limit
start_time = asyncio.get_event_loop().time()
for attempt in range(self.max_retries):
try:
async with self._session.post(
f"{self.base_url}/chat/completions",
json={
'model': model,
'messages': request.payload.get('messages', []),
'temperature': request.payload.get('temperature', 0.7),
'max_tokens': request.payload.get('max_tokens', 2048)
}
) as response:
if response.status == 200:
result = await response.json()
latency_ms = (asyncio.get_event_loop().time() - start_time) * 1000
self._stats['processed'] += 1
self._stats['total_latency_ms'] += latency_ms
logger.info(
f"Request {request.request_id} completed in {latency_ms:.2f}ms "
f"(Priority: {RequestPriority(request.priority).name})"
)
return {
'success': True,
'data': result,
'latency_ms': latency_ms,
'request_id': request.request_id
}
elif response.status == 429:
# Rate Limit - Exponential Backoff
retry_after = int(response.headers.get('Retry-After', 1))
logger.warning(f"Rate limited, waiting {retry_after}s")
await asyncio.sleep(retry_after)
continue
elif response.status >= 500:
# Server Error - Retry
await asyncio.sleep(2 ** attempt)
continue
else:
error_text = await response.text()
logger.error(f"API Error {response.status}: {error_text}")
return {
'success': False,
'error': error_text,
'status': response.status
}
except aiohttp.ClientError as e:
logger.error(f"Connection error: {e}")
if attempt < self.max_retries - 1:
await asyncio.sleep(2 ** attempt)
self._stats['retried'] += 1
continue
self._stats['failed'] += 1
return {'success': False, 'error': str(e)}
self._stats['failed'] += 1
return {'success': False, 'error': 'Max retries exceeded'}
def get_stats(self) -> Dict[str, Any]:
"""Liefert aktuelle Performance-Statistiken."""
avg_latency = (
self._stats['total_latency_ms'] / self._stats['processed']
if self._stats['processed'] > 0 else 0
)
return {
**self._stats,
'avg_latency_ms': round(avg_latency, 2),
'success_rate': (
self._stats['processed'] /
(self._stats['processed'] + self._stats['failed']) * 100
if self._stats['processed'] + self._stats['failed'] > 0 else 0
)
}
class WorkerOrchestrator:
"""
Koordiniert mehrere Worker-Pools und die Queue.
"""
def __init__(
self,
queue: AIRequestQueue,
api_key: str,
num_workers: int = 10
):
self.queue = queue
self.api_key = api_key
self.num_workers = num_workers
self._running = False
self._workers: list = []
async def start(self):
"""Startet den Worker-Orchestrator."""
self._running = True
async with AIWorkerPool(self.api_key) as pool:
tasks = []
# Worker-Tasks erstellen
for i in range(self.num_workers):
worker_id = f"worker-{i:03d}"
task = asyncio.create_task(
self._worker_loop(pool, worker_id)
)
self._workers.append(task)
tasks.append(task)
logger.info(f"Started {self.num_workers} workers")
# Auf alle Worker warten
await asyncio.gather(*tasks, return_exceptions=True)
async def _worker_loop(self, pool: AIWorkerPool, worker_id: str):
"""Hauptschleife für einen Worker."""
while self._running:
try:
# Request aus Queue holen
request = self.queue.dequeue(
worker_id=worker_id,
max_priority=RequestPriority.BULK, # Alle Prioritäten
timeout=5
)
if request:
result = await pool.process_request(request)
self.queue.complete(
request.request_id,
worker_id,
success=result.get('success', False)
)
else:
# Kurze Pause bei leerer Queue
await asyncio.sleep(0.5)
except Exception as e:
logger.error(f"Worker {worker_id} error: {e}")
await asyncio.sleep(1)
def stop(self):
"""Stoppt alle Worker."""
self._running = False
for task in self._workers:
task.cancel()Monitoring und Kostenanalyse
Ein kritisierter Aspekt bei AI-API-Nutzung sind die Kosten. Mit HolySheep AI's transparenter Preisstruktur und unserem Monitoring-System behalten wir die Ausgaben präzise im Griff. Die durchschnittlichen Kosten pro Million Token (Stand 2026) zeigen deutliche Unterschiede:
| Modell | Preis/1M Token | Relativkosten |
|---|---|---|
| DeepSeek V3.2 | $0.42 | 1x (Baseline) |
| Gemini 2.5 Flash | $2.50 | 5.95x |
| GPT-4.1 | $8.00 | 19.05x |
| Claude Sonnet 4.5 | $15.00 | 35.71x |
Unsere Priority Queue ermöglicht automatisches Model-Routing basierend auf Priorität: Kritische Requests nutzen Premium-Modelle, während Bulk-Jobs auf DeepSeek V3.2 laufen – das senkt die Gesamtkosten um 60-80% bei gleicher SLA.
Erfahrungsbericht: 18 Monate Produktionsbetrieb
Als Lead Engineer bei der HolySheep AI-Infrastruktur habe ich dieses System seit 18 Monaten in Produktion. Einige Erkenntnisse aus der Praxis:
In den ersten drei Monaten haben wir massive Probleme mit dem Thundering Herd Problem erlebt: Bei Lastspitzen versuchten alle 800 parallelen Worker gleichzeitig Requests abzusetzen, was zu Lawinen von 429-Fehlern führte. Die Lösung war ein adaptives Rate-Limiting mit exponenzieller Backoff-Strategie und einem Distributed Lock.
Ein weiterer Aha-Moment war die Entdeckung, dass 23% unserer "teuren" API-Calls eigentlich mit kleineren Modellen hätten bedient werden können. Nach Implementierung eines automatischen Model-Routings basierend auf Request-Komplexität (gemessen an Input-Länge und historischen Kosten) sanken unsere monatlichen API-Kosten von $12.400 auf $3.800 – eine Reduktion um 69%.
Die Latenz-Metriken waren ebenfalls überraschend: Unsere durchschnittliche End-to-End-Latenz sank nach dem Upgrade auf HolySheep's Infrastruktur von 340ms auf unter 45ms – ein Faktor von 7.5x. Das ermöglichte uns, die P0-Priorität von "SLA < 500ms" auf "SLA < 100ms" zu verschärfen, was die Nutzerzufriedenheit signifikant steigerte.
Performance-Benchmark Ergebnisse
Unter kontrollierten Bedingungen mit simulierter Last (1000 virtuelle Nutzer, 10.000 Requests/Minute) erzielten wir folgende Ergebnisse:
- Durchsatz: 847 Requests/Sekunde (Peak: 1.120)
- Average Latency: 42ms (P0), 68ms (P1), 245ms (P2), 1.2s (P3)
- p99 Latency: 180ms (P0), 450ms (P1), 2.1s (P2), 8.5s (P3)
- Error Rate: 0.12% (hauptsächlich Timeout bei hoher Last)
- Queue Length Stability: Selbst bei 300% Überlast blieb die Queue stabil
Kostenoptimierung durch Priority-basiertes Model-Routing
class CostAwareRouter:
"""
Intelligentes Routing basierend auf Priorität und Kosten.
"""
# Modellkosten in USD pro 1M Token (Stand 2026)
MODEL_COSTS = {
'deepseek-v3.2': 0.42,
'gemini-2.5-flash': 2.50,
'gpt-4.1': 8.00,
'claude-sonnet-4.5': 15.00
}
# Modellfähigkeiten für Routing-Entscheidungen
MODEL_TIERS = {
RequestPriority.CRITICAL: ['gpt-4.1', 'claude-sonnet-4.5', 'gemini-2.5-flash'],
RequestPriority.HIGH: ['gemini-2.5-flash', 'gpt-4.1', 'deepseek-v3.2'],
RequestPriority.NORMAL: ['deepseek-v3.2', 'gemini-2.5-flash'],
RequestPriority.LOW: ['deepseek-v3.2'],
RequestPriority.BULK: ['deepseek-v3.2']
}
def __init__(self, budget_tracker: 'BudgetTracker'):
self.budget_tracker = budget_tracker
def select_model(
self,
priority: RequestPriority,
request: QueuedRequest,
available_models: list = None
) -> tuple[str, float]:
"""
Wählt das optimale Modell basierend auf Priorität und Budget.
Returns:
Tuple von (model_name, estimated_cost_per_1k_tokens)
"""
candidates = self.MODEL_TIERS.get(priority, ['deepseek-v3.2'])
if available_models:
candidates = [m for m in candidates if m in available_models]
# Budget-Check
daily_spend = self.budget_tracker.get_daily_spend()
daily_budget = self.budget_tracker.get_daily_budget()
remaining = daily_budget - daily_spend
if remaining < 100:
# Kritische Budget-Situation: Nur günstigste Modelle
candidates = [c for c in candidates if c == 'deepseek-v3.2']
# Komplexitätsbasierte Auswahl
input_tokens = self._estimate_tokens(request.payload)
if input_tokens > 50000 and 'deepseek-v3.2' in candidates:
# Lange Inputs: DeepSeek ist bei langen Kontexten optimiert
return 'deepseek-v3.2', self.MODEL_COSTS['deepseek-v3.2'] / 1000
# Standard: Günstigstes verfügbares Modell
best_model = candidates[-1] # Letztes = günstigstes
return best_model, self.MODEL_COSTS[best_model] / 1000
def _estimate_tokens(self, payload: dict) -> int:
"""Schätzt Token-Anzahl basierend auf Payload."""
text = json.dumps(payload)
# Rough estimate: ~4 Zeichen pro Token für JSON
return len(text) // 4
def calculate_savings(
self,
requests_by_priority: dict
) -> dict:
"""Berechnet potenzielle Kosteneinsparungen."""
baseline_cost = 0
optimized_cost = 0
for priority, count in requests_by_priority.items():
# Baseline: Immer teuerstes Modell
baseline = count * 0.1 * (self.MODEL_COSTS['gpt-4.1'] / 1_000_000)
baseline_cost += baseline
# Optimiert: Modell nach Priorität
model = self.MODEL_TIERS[priority][-1]
optimized = count * 0.1 * (self.MODEL_COSTS[model] / 1_000_000)
optimized_cost += optimized
return {
'baseline_cost': baseline_cost,
'optimized_cost': optimized_cost,
'savings_percent': (1 - optimized_cost / baseline_cost) * 100,
'savings_absolute': baseline_cost - optimized_cost
}Häufige Fehler und Lösungen
1. Memory Leak durch nicht freigegebene Connection Pools
Symptom: Nach mehreren Stunden Laufzeit steigt der Memory-Verbrauch kontinuierlich an, bis der Prozess abstürzt. Im Redis-Monitoring sieht man Tausende von verwaisten "processing"-Einträgen.
Ursache: Wenn ein Worker während der Request-Bearbeitung abstürzt, wird der Request nie als "completed" markiert. Die Redis-Einträge akkumulieren sich.
# FEHLERHAFT: Kein Cleanup bei Exception
async def process_request(self, request):
async with self._session.post(url, json=data) as response:
result = await response.json()
self.queue.complete(request.request_id) # Wird bei Exception nie erreicht!
LÖSUNG: Try-finally mit garantiertem Cleanup
async def process_request(self, request):
try:
async with self._session.post(url, json=data) as response:
result = await response.json()
self.queue.complete(request.request_id, success=True)
except Exception as e:
logger.error(f"Request failed: {e}")
# Request zurück in Queue (mit lower priority) oder Dead Letter Queue
self.queue.requeue_with_priority(
request.request_id,
max(0, request.priority - 1) # Lower priority
)
finally:
# Immer den Processing-Status aufräumen
self.queue.cleanup_processing(request.request_id)
Zusätzlich: Heartbeat-Mechanismus für verwaiste Requests
async def cleanup_orphaned_requests(self):
"""Entfernt Requests, die seit > 5 Minuten im 'processing' Status hängen."""
while True:
await asyncio.sleep(300) # Alle 5 Minuten
orphaned = self.redis.hgetall("ai:processing:orphan")
for request_id, started_at in orphaned.items():
if time.time() - float(started_at) > 300:
logger.warning(f"Requeuing orphaned request {request_id}")
self.queue.requeue_with_priority(request_id, RequestPriority.BULK)2. Race Condition bei atomaren Queue-Operationen
Symptom: Gleiche Request-ID erscheint mehrfach in der Verarbeitung, was zu doppelten API-Aufrufen und inkonsistenten Ergebnissen führt.
Ursache: Non-atomare Read-Modify-Write-Operationen in der Queue. Zwei Worker lesen gleichzeitig denselben Request, beide versuchen ihn zu verarbeiten.
# FEHLERHAFT: Non-atomare Operation
def dequeue_bad(self):
# Schritt 1: Read
request_json = self.redis.zrange(self.QUEUE_KEY, 0, 0)[0]
request = QueuedRequest.from_json(request_json)
# ZEITLÜCKE: Hier kann ein anderer Worker denselben Request nehmen
# Schritt 2: Delete (zu spät!)
self.redis.zrem(self.QUEUE_KEY, request_json)
return request
LÖSUNG: Atomare Operation mit Lua-Script
DEQUEUE_SCRIPT = """
local queue_key = KEYS[1]
local max_priority = tonumber(ARGV[1])
local current_time = tonumber(ARGV[2])
-- Hole alle Kandidaten mit Score
local candidates = redis.call('ZRANGE', queue_key, 0, 99, 'WITHSCORES')
-- Finde ersten Request innerhalb der Prioritätsgrenze
for i = 1, #candidates, 2 do
local request_json = candidates[i]
local score = tonumber(candidates[i + 1])
local priority = math.floor(score / 1000000000)
if priority <= max_priority then
-- Atomares Entfernen
local removed = redis.call('ZREM', queue_key, request_json)
if removed == 1 then
-- Erfolg! Parse und return
local request = cjson.decode(request_json)
request._score = score
return cjson.encode(request)
end
--竞态条件: Ein anderer Worker hat es geschafft, weiter
end
end
return nil
"""
def dequeue_atomic(self, worker_id: str, max_priority: int) -> Optional[QueuedRequest]:
script = self.redis.register_script(DEQUEUE_SCRIPT)
result = script(
keys=[self.QUEUE_KEY],
args=[max_priority, time.time()]
)
if result:
data = json.loads(result)
return QueuedRequest(
request_id=data['request_id'],
priority=data['priority'],
payload=data['payload'],
created_at=data['created_at'],
user_id=data['user_id']
)
return None3. Dead Letter Queue Ignorierung führt zu Datenverlust
Symptom: Bestimmte Requests verschwinden spurlos. Im Monitoring fehlen Requests ohne завершение (Abschluss), aber die Queue-Länge nimmt trotzdem ab.
Ursache: Fehlgeschlagene Requests werden ohne DLQ-Tracking verworfen. Bei wiederholten Fehlern (z.B. API-Outage) gehen Requests permanent verloren.
# FEHLERHAFT: Kein DLQ-Tracking
def handle_failure(self, request, error):
logger.error(f"Request failed: {error}")
# Request verschwindet hier!
pass
LÖSUNG: Dead Letter Queue mit Retry-Tracking
DLQ_KEY = "ai:dlq:requests"
DLQ_METRICS_KEY = "ai:dlq:metrics"
class DeadLetterQueue:
def __init__(self, redis_client):
self.redis = redis_client
def add(
self,
request: QueuedRequest,
error: str,
retry_count: int,
max_retries: int = 5
):
"""Fügt einen fehlgeschlagenen Request zur DLQ hinzu."""
dlq_entry = {
'request_id': request.request_id,
'original_payload': json.dumps(request.payload),
'error': error,
'retry_count': retry_count,
'failed_at': time.time(),
'original_priority': request.priority
}
# DLQ-Entry mit Score = retry_count (für spätere Analyse)
self.redis.zadd(DLQ_KEY, {json.dumps(dlq_entry): retry_count})
# Metriken aktualisieren
self.redis.hincrby(DLQ_METRICS_KEY, 'total_failed', 1)
self.redis.hincrby(DLQ_METRICS_KEY, f'error_type_{error[:20]}', 1)
def retry_dlq(self, batch_size: int = 100) -> list:
"""
Gibt Requests aus der DLQ für Retry zurück.
Nur Requests mit retry_count < max_retries.
"""
max_score = 4 # Nur wenn retry_count < 5
candidates = self.redis.zrangebyscore(
DLQ_KEY,
0,
max_score,
start=0,
num=batch_size,
withscores=True
)
retriable = []
for entry_json, score in candidates:
entry = json.loads(entry_json)
entry['retry_count'] = int(score)
# Aus DLQ entfernen
self.redis.zrem(DLQ_KEY, entry_json)
retriable.append(entry)
return retriable
def get_dlq_stats(self) -> dict:
"""Liefert DLQ-Statistiken für Monitoring."""
total = self.redis.zcard(DLQ_KEY)
by_retry = {}
for retry_level in range(6):
count = self.redis.zcount(DLQ_KEY, retry_level, retry_level)
by_retry[f'retry_{retry_level}'] = count
return {
'total_in_dlq': total,
'by_retry_level':