Redis: mehr als ein Cache (Pub/Sub, Queues, Rate Limiting)
Inhalt
Ich betreibe Redis in fast jedem Projekt — und nutze es selten nur als Cache. Redis ist ein In-Memory-Datenstruktur-Store mit erstaunlicher Vielseitigkeit.
Redis mit Python verbinden
pip install redis[hiredis] # hiredis: C-Erweiterung für Speed
import redis
# Synchron
r = redis.Redis(host="localhost", port=6379, db=0, decode_responses=True)
# Asynchron (für asyncio/FastAPI)
import redis.asyncio as aioredis
ar = aioredis.Redis(host="localhost", port=6379, decode_responses=True)
# Connection Pool (Produktion)
pool = redis.ConnectionPool(host="localhost", port=6379, max_connections=50)
r = redis.Redis(connection_pool=pool)Pub/Sub-Messaging
Redis Pub/Sub ermöglicht Echtzeit-Kommunikation zwischen Prozessen — z.B. für Live-Notifications:
# Publisher
import redis
r = redis.Redis(decode_responses=True)
def publish_order_event(order_id: int, event: str):
r.publish(
f"orders:{order_id}",
json.dumps({"event": event, "order_id": order_id, "ts": time.time()})
)
# Subscriber (in separatem Prozess/Thread)
pubsub = r.pubsub()
pubsub.subscribe("orders:*") # Pattern-Subscription
for message in pubsub.listen():
if message["type"] == "pmessage":
data = json.loads(message["data"])
handle_order_event(data)Job Queues ohne Celery
Für einfache Queue-Anforderungen reicht Redis ohne Celery:
import redis, json, time
r = redis.Redis(decode_responses=True)
# Producer: Job einreihen
def enqueue_job(queue: str, payload: dict, priority: int = 5):
job = {"id": str(uuid.uuid4()), "payload": payload, "created": time.time()}
r.zadd(queue, {json.dumps(job): priority}) # Sorted Set = Prioritäts-Queue
# Consumer: Job abholen (BZPOPMIN = blockierend, höchste Priorität)
def consume_job(queue: str, timeout: int = 5):
result = r.bzpopmin(queue, timeout=timeout)
if result:
_, job_json, _ = result
return json.loads(job_json)
return None
# Verwendung
enqueue_job("email_queue", {"to": "user@example.com", "subject": "Test"}, priority=10)
job = consume_job("email_queue")Rate Limiting
Elegant mit Sliding Window:
import redis, time
r = redis.Redis(decode_responses=True)
def is_rate_limited(user_id: str, limit: int = 100, window: int = 3600) -> bool:
key = f"rate_limit:{user_id}"
now = time.time()
pipe = r.pipeline()
# Alte Einträge entfernen
pipe.zremrangebyscore(key, 0, now - window)
# Aktuellen Request hinzufügen
pipe.zadd(key, {str(now): now})
# Anzahl der Requests zählen
pipe.zcard(key)
# TTL setzen
pipe.expire(key, window)
_, _, count, _ = pipe.execute()
return count > limit
# Im Django/FastAPI View:
if is_rate_limited(str(request.user.id)):
raise RateLimitExceeded("Zu viele Anfragen")Leaderboards mit Sorted Sets
import redis
r = redis.Redis(decode_responses=True)
# Punkte hinzufügen
r.zadd("leaderboard:2026", {"user:42": 1500, "user:7": 2300, "user:99": 800})
r.zincrby("leaderboard:2026", 100, "user:42") # +100 Punkte
# Top 10 abrufen
top10 = r.zrevrange("leaderboard:2026", 0, 9, withscores=True)
# [("user:7", 2300.0), ("user:42", 1600.0), ...]
# Rang eines Users
rank = r.zrevrank("leaderboard:2026", "user:42") # 0-basiert
print(f"Rang: {rank + 1}")Sessions und Tokens
import redis, secrets, json
from datetime import timedelta
r = redis.Redis(decode_responses=True)
def create_session(user_id: int, ttl: int = 86400) -> str:
token = secrets.token_urlsafe(32)
session_data = {
"user_id": user_id,
"created_at": time.time(),
}
r.setex(f"session:{token}", ttl, json.dumps(session_data))
return token
def get_session(token: str) -> dict | None:
data = r.get(f"session:{token}")
return json.loads(data) if data else None
def invalidate_session(token: str):
r.delete(f"session:{token}")