redis-patterns

Redis Patterns

Kurzreferenz für Redis-Best-Practices in gängigen Backend-Anwendungsfällen.

Funktionsweise

Redis ist ein In-Memory-Datenstrukturspeicher, der Strings, Hashes, Listen, Sets, sortierte Sets, Streams und mehr unterstützt. Einzelne Redis-Befehle sind auf einer einzelnen Instanz atomar; mehrschrittige Workflows erfordern Lua-Skripte, MULTI/EXEC-Transaktionen oder explizite Synchronisation, um atomar zu bleiben. Daten werden optional über RDB-Snapshots oder AOF-Protokolle persistiert. Clients kommunizieren über TCP mit dem RESP-Protokoll; Verbindungspools sind unerlässlich, um den Overhead des Handshakes pro Anfrage zu vermeiden.

Wann zu aktivieren

• Hinzufügen von Caching zu einer Anwendung
• Implementieren von Ratenbegrenzung oder Drosselung
• Erstellen verteilter Sperren oder Koordination
• Einrichten von Sitzungs- oder Token-Speicher
• Verwenden von Pub/Sub oder Redis Streams für Nachrichtenübermittlung
• Konfigurieren von Redis in der Produktion (Pooling, Räumung, Clustering)

Spickzettel für Datenstrukturen

Anwendungsfall	Struktur	Beispielschlüssel
Einfacher Cache	String	product:123
Benutzersitzung	Hash	session:abc
Bestenliste	Sortiertes Set	scores:weekly
Einzigartige Besucher	Set	visitors:2024-01-01
Aktivitätsfeed	Liste	feed:user:456
Ereignisstream	Stream	events:orders
Zähler / Ratenbegrenzungen	String (INCR)	ratelimit:user:123
Bloom-Filter / HLL	HyperLogLog	hll:pageviews

Kernmuster

Cache-Aside (Lazy Loading)

import redis
import json

r = redis.Redis(host='localhost', port=6379, decode_responses=True)

def get_product(product_id: int):
    cache_key = f"product:{product_id}"
    cached = r.get(cache_key)

    if cached:
        return json.loads(cached)

    product = db.query("SELECT * FROM products WHERE id = %s", product_id)
    r.setex(cache_key, 3600, json.dumps(product))  # TTL: 1 Stunde
    return product

Write-Through-Cache

def update_product(product_id: int, data: dict):
    # Zuerst in die DB schreiben
    db.execute("UPDATE products SET ... WHERE id = %s", product_id)

    # Cache sofort aktualisieren
    cache_key = f"product:{product_id}"
    r.setex(cache_key, 3600, json.dumps(data))

Cache-Invalidierung

# Tag-basierte Invalidierung – gruppierte verwandte Schlüssel unter einem Set
def cache_product(product_id: int, category_id: int, data: dict):
    key = f"product:{product_id}"
    tag = f"tag:category:{category_id}"
    pipe = r.pipeline(transaction=True)
    pipe.setex(key, 3600, json.dumps(data))
    pipe.sadd(tag, key)
    pipe.expire(tag, 3600)
    pipe.execute()

def invalidate_category(category_id: int):
    tag = f"tag:category:{category_id}"
    keys = r.smembers(tag)
    if keys:
        r.delete(*keys)
    r.delete(tag)

Sitzungsspeicher

import time
import uuid

def create_session(user_id: int, ttl: int = 86400) -> str:
    session_id = str(uuid.uuid4())
    key = f"session:{session_id}"
    pipe = r.pipeline(transaction=True)
    pipe.hset(key, mapping={
        "user_id": user_id,
        "created_at": int(time.time()),
    })
    pipe.expire(key, ttl)
    pipe.execute()
    return session_id

def get_session(session_id: str) -> dict | None:
    data = r.hgetall(f"session:{session_id}")
    return data if data else None

def delete_session(session_id: str):
    r.delete(f"session:{session_id}")

Ratenbegrenzung

Festes Fenster (Einfach)

def is_rate_limited(user_id: int, limit: int = 100, window: int = 60) -> bool:
    key = f"ratelimit:{user_id}:{int(time.time()) // window}"
    pipe = r.pipeline(transaction=True)
    pipe.incr(key)
    pipe.expire(key, window)
    count, _ = pipe.execute()
    return count > limit

Gleitendes Fenster (Lua – Atomar)

-- sliding_window.lua
local key = KEYS[1]
local now = tonumber(ARGV[1])
local window = tonumber(ARGV[2])
local limit = tonumber(ARGV[3])

redis.call('ZREMRANGEBYSCORE', key, 0, now - window)
local count = redis.call('ZCARD', key)

if count < limit then
    -- Eindeutiges Mitglied verwenden (now + sequence), um Kollisionen innerhalb derselben Millisekunde zu vermeiden
    local seq_key = key .. ':seq'
    local seq = redis.call('INCR', seq_key)
    redis.call('EXPIRE', seq_key, math.ceil(window / 1000))
    redis.call('ZADD', key, now, now .. '-' .. seq)
    redis.call('EXPIRE', key, math.ceil(window / 1000))
    return 1
end
return 0

sliding_window = r.register_script(open('sliding_window.lua').read())

def allow_request(user_id: int) -> bool:
    key = f"ratelimit:sliding:{user_id}"
    now = int(time.time() * 1000)
    return bool(sliding_window(keys=[key], args=[now, 60000, 100]))

Verteilte Sperren

Verteilte Sperre (Einzelner Knoten – SET NX PX)

import uuid

def acquire_lock(resource: str, ttl_ms: int = 5000) -> str | None:
    lock_key = f"lock:{resource}"
    token = str(uuid.uuid4())
    acquired = r.set(lock_key, token, px=ttl_ms, nx=True)
    return token if acquired else None

def release_lock(resource: str, token: str) -> bool:
    release_script = """
    if redis.call('get', KEYS[1]) == ARGV[1] then
        return redis.call('del', KEYS[1])
    else
        return 0
    end
    """
    result = r.eval(release_script, 1, f"lock:{resource}", token)
    return bool(result)

# Verwendung
token = acquire_lock("order:payment:123")
if token:
    try:
        process_payment()
    finally:
        release_lock("order:payment:123", token)

Verwenden Sie für Multi-Node-Setups die Bibliothek redlock-py, die den vollständigen Redlock-Algorithmus implementiert.

Pub/Sub & Streams

Pub/Sub (Feuer-und-Vergiss)

# Herausgeber
def publish_event(channel: str, payload: dict):
    r.publish(channel, json.dumps(payload))

# Abonnent (blockierend – in separatem Thread/Prozess ausführen)
def subscribe_events(channel: str):
    pubsub = r.pubsub()
    pubsub.subscribe(channel)
    for message in pubsub.listen():
        if message['type'] == 'message':
            handle(json.loads(message['data']))

Redis Streams (Langlebige Warteschlange)

# Produzent
def emit(stream: str, event: dict):
    r.xadd(stream, event, maxlen=10000)  # Stream-Länge begrenzen

# Verbrauchergruppe – garantiert mindestens einmalige Zustellung
try:
    r.xgroup_create('events:orders', 'processor', id='0', mkstream=True)
except Exception:
    pass  # Gruppe existiert bereits

def consume(stream: str, group: str, consumer: str):
    while True:
        messages = r.xreadgroup(group, consumer, {stream: '>'}, count=10, block=2000)
        for _, entries in (messages or []):
            for msg_id, data in entries:
                process(data)
                r.xack(stream, group, msg_id)

Bevorzugen Sie Streams gegenüber Pub/Sub, wenn Sie Zustellungsgarantien, Verbrauchergruppen oder Wiederholung benötigen.

Schlüsseldesign

Namenskonventionen

# Muster: resource:id:field
user:123:profile
order:456:status
cache:product:789

# Muster: namespace:resource:id
myapp:session:abc123
myapp:ratelimit:user:123

# Muster: resource:date (zeitgebundene Schlüssel)
stats:pageviews:2024-01-01

TTL-Strategie

Datentyp	Empfohlene TTL
Benutzersitzung	24h (86400)
API-Antwort-Cache	5–15 Min
Ratenbegrenzungsfenster	Fenstergröße anpassen
Kurzlebige Token	5–10 Min
Bestenliste	1h–24h
Statische/Referenzdaten	1h–1 Woche

Legen Sie immer eine TTL fest. Schlüssel ohne TTL sammeln sich auf unbestimmte Zeit an und verursachen Speicherdruck.

Verbindungsverwaltung

Verbindungspooling

from redis import ConnectionPool, Redis

pool = ConnectionPool(
    host='localhost',
    port=6379,
    db=0,
    max_connections=20,
    decode_responses=True,
    socket_connect_timeout=2,
    socket_timeout=2,
)

r = Redis(connection_pool=pool)

Cluster-Modus

from redis.cluster import RedisCluster

r = RedisCluster(
    startup_nodes=[{"host": "redis-1", "port": 6379}],
    decode_responses=True,
    skip_full_coverage_check=True,
)

Sentinel (Hohe Verfügbarkeit)

from redis.sentinel import Sentinel

sentinel = Sentinel(
    [('sentinel-1', 26379), ('sentinel-2', 26379)],
    socket_timeout=0.5,
)
master = sentinel.master_for('mymaster', decode_responses=True)
replica = sentinel.slave_for('mymaster', decode_responses=True)

Räumungsrichtlinien

Richtlinie	Verhalten	Am besten geeignet für
noeviction	Fehler beim Schreiben, wenn voll	Warteschlangen / kritische Daten
allkeys-lru	Räumt am wenigsten kürzlich verwendet	Allgemeiner Cache
volatile-lru	LRU nur bei Schlüsseln mit TTL	Gemischter Datenspeicher
allkeys-lfu	Räumt am wenigsten häufig verwendet	Ungleichmäßige Zugriffsmuster
volatile-ttl	Räumt den am bald ablaufenden	Priorisiert langlebige Daten

Legen Sie über redis.conf fest: maxmemory-policy allkeys-lru

Anti-Patterns

Anti-Pattern	Problem	Lösung
Schlüssel ohne TTL	Speicher wächst unbegrenzt	Immer TTL setzen
KEYS * in der Produktion	Blockiert den Server (O(N))	SCAN-Cursor verwenden
Speichern großer Blobs (>100KB)	Langsame Serialisierung, Speicherdruck	Referenz speichern + aus Objektspeicher abrufen
Ein Redis für alles	Keine Isolation zwischen Cache und Warteschlange	Separate DBs oder Instanzen verwenden
Ignorieren von Verbindungspoolgrenzen	Verbindungserschöpfung unter Last	Pool an Workload anpassen
Nicht behandeln von Cache-Miss-Stampede	Herdenproblem bei Kaltstart	Sperren oder probabilistischen vorzeitigen Ablauf verwenden
FLUSHALL ohne Nachdenken	Löscht gesamte Instanz	Löschvorgänge nach Schlüsselmuster begrenzen

Verhinderung von Cache-Miss-Stampede

import threading

_locks: dict[str, threading.Lock] = {}
_locks_mutex = threading.Lock()

def get_with_lock(key: str, fetch_fn, ttl: int = 300):
    cached = r.get(key)
    if cached:
        return json.loads(cached)

    with _locks_mutex:
        if key not in _locks:
            _locks[key] = threading.Lock()
        lock = _locks[key]
    with lock:
        cached = r.get(key)  # Nach Erwerb der Sperre erneut prüfen
        if cached:
            return json.loads(cached)
        value = fetch_fn()
        r.setex(key, ttl, json.dumps(value))
        return value

Hinweis: Ersetzen Sie bei Multi-Prozess-Bereitstellungen die prozessinterne Sperre durch acquire_lock/release_lock aus dem obigen Abschnitt zu verteilten Sperren.

Beispiele

Caching zu einem Django/Flask-API-Endpunkt hinzufügen: Verwenden Sie Cache-Aside mit setex und einer TTL von 5 Minuten für die Antwort. Schlüssel auf die Anforderungsparameter setzen.

API nach Benutzer ratenbegrenzen: Verwenden Sie das feste Fenster mit pipeline(transaction=True) für Endpunkte mit geringem Datenverkehr; verwenden Sie das gleitende Fenster Lua für genaue Drosselung pro Benutzer.

Einen Hintergrundjob über mehrere Worker hinweg koordinieren: Verwenden Sie acquire_lock mit einer TTL, die die erwartete Jobdauer überschreitet. Geben Sie die Sperre immer in einem finally-Block frei.

Benachrichtigungen an mehrere Abonnenten weiterleiten: Verwenden Sie Pub/Sub für Feuer-und-Vergiss. Wechseln Sie zu Streams, wenn Sie eine garantierte Zustellung oder Wiederholung für späte Verbraucher benötigen.

Kurzreferenz

Muster	Wann verwenden
Cache-Aside	Leselastig, toleriert leichte Veraltung
Write-Through	Starke Konsistenz erforderlich
Verteilte Sperre	Gleichzeitigen Zugriff auf eine Ressource verhindern
Gleitendes Fenster Ratenbegrenzung	Genaue Drosselung pro Benutzer
Redis Streams	Langlebige Ereigniswarteschlange mit Verbrauchergruppen
Pub/Sub	Broadcast ohne Zustellungsgarantie erforderlich
Sortiertes Set Bestenliste	Bewertete Rangfolge, Paginierung
HyperLogLog	Ungefähre eindeutige Zählung bei niedrigem Speicherverbrauch

Verwandte

• Skill: postgres-patterns — Muster für relationale Daten
• Skill: backend-patterns — API- und Servicelayermuster
• Skill: database-migrations — Schema-Versionierung
• Skill: django-patterns — Integration des Django-Cache-Frameworks
• Agent: database-reviewer — vollständiger Datenbank-Review-Workflow