redis-patterns

Redis Patterns

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

So funktioniert es

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

Wann aktivieren

• Hinzufügen von Caching zu einer Anwendung
• Implementierung von Ratenbegrenzung oder Drosselung
• Erstellen verteilter Sperren oder Koordination
• Einrichten von Sitzungs- oder Token-Speicher
• Verwenden von Pub/Sub oder Redis Streams für Messaging
• Konfigurieren von Redis in Produktion (Pooling, Eviction, Clustering)

Spickzettel für Datenstrukturen

Anwendungsfall	Struktur	Beispiel-Schlü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
Ereignisstrom	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 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 – gruppiere 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 (jetzt + Sequenz), 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)

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

Pub/Sub & Streams

Pub/Sub (Feuern und Vergessen)

# Publisher
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 (Dauerhafte Warteschlange)

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

# Verbrauchergruppe – garantiert Mindestens-einmal-Zustellung
versuchen:
    r.xgroup_create('events:orders', 'processor', id='0', mkstream=True)
außer Exception:
    bestanden  # Gruppe existiert bereits

def consume(stream: str, group: str, consumer: str):
    während True:
        messages = r.xreadgroup(group, consumer, {stream: '>'}, count=10, block=2000)
        für _, entries in (messages or []):
            für 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	Vorgeschlagenes TTL
Benutzersitzung	24h (86400)
API-Antwort-Cache	5–15 Minuten
Ratenbegrenzungsfenster	Fenstergröße entsprechen
Kurzlebige Tokens	5–10 Minuten
Bestenliste	1h–24h
Statische/Referenzdaten	1h–1 Woche

Setzen Sie immer ein TTL. 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 (Hochverfü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)

Eviction-Richtlinien

Richtlinie	Verhalten	Am besten geeignet für
noeviction	Fehler beim Schreiben, wenn voll	Warteschlangen / kritische Daten
allkeys-lru	Am wenigsten kürzlich verwendet entfernen	Allgemeiner Cache
volatile-lru	LRU nur bei Schlüsseln mit TTL	Gemischter Datenspeicher
allkeys-lfu	Am wenigsten häufig verwendet entfernen	Schiefe Zugriffsmuster
volatile-ttl	Nächstablaufend entfernen	Priorisiere langlebige Daten

Setzen Sie via redis.conf: maxmemory-policy allkeys-lru

Anti-Patterns

Anti-Pattern	Problem	Behebung
Schlüssel ohne TTL	Speicher wächst unbegrenzt	Immer TTL setzen
KEYS * in Produktion	Blockiert den Server (O(N))	SCAN-Cursor verwenden
Große Blobs speichern (>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
Verbindungspool-Limits ignorieren	Verbindungserschöpfung unter Last	Pool auf Workload abstimmen
Cache-Miss-Stampede nicht behandeln	Thundering Herd bei Kaltstart	Sperren oder probabilistisches vorzeitiges Ablaufen verwenden
FLUSHALL ohne Nachdenken	Löscht gesamte Instanz	Löschungen auf Schlüsselmuster beschränken

Cache-Miss-Stampede-Prävention

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)  # Erneut prüfen nach Sperrerwerb
        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 "Verteilte Sperren".

Beispiele

Django/Flask-API-Endpunkt mit Caching versehen: Verwenden Sie Cache-Aside mit setex und einer TTL von 5 Minuten für die Antwort. Schlüsseln Sie auf die Anforderungsparameter.

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 Ratenbegrenzung pro Benutzer.

Hintergrundjob über mehrere Worker 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 verteilen: Verwenden Sie Pub/Sub für "Feuern und Vergessen". 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
Ratenbegrenzung mit gleitendem Fenster	Genaue Ratenbegrenzung pro Benutzer
Redis Streams	Dauerhafte Ereigniswarteschlange mit Verbrauchergruppen
Pub/Sub	Broadcast ohne Zustellungsgarantien
Bestenliste mit sortiertem Set	Rangbewertung, Paginierung
HyperLogLog	Ungefähre eindeutige Zählung bei geringem Speicherverbrauch

Verwandt

• Fertigkeit: postgres-patterns — Relationale Datenmuster
• Fertigkeit: backend-patterns — API- und Dienstschichtmuster
• Fertigkeit: database-migrations — Schema-Versionierung
• Fertigkeit: django-patterns — Django-Cache-Framework-Integration
• Agent: database-reviewer — Vollständiger Datenbank-Überprüfungs-Workflow