Redis Anwendungsfälle: Caching, Queues & Rate Limiting

Sie haben Redis zu Ihrem Stack hinzugefügt, aber verwenden es nur als einfachen Key-Value-Speicher. Redis bietet leistungsstarke Datenstrukturen und Befehle, mit denen sich Caching, Queuing und Rate Limiting elegant lösen lassen. Dieser Leitfaden zeigt Ihnen, wie Sie diese Muster effektiv implementieren.

Warum Redis für diese Anwendungsfälle?

Redis ist ein In-Memory-Datenspeicher, der für seine Geschwindigkeit und Vielseitigkeit bekannt ist. Es unterstützt Strings, Hashes, Listen, Sets, Sorted Sets und mehr. Seine atomaren Operationen und die integrierte Ablaufzeit machen es ideal für Caching, Queues und Rate Limiting. Zudem lässt es sich einfach in jede Sprache integrieren.

Caching mit Redis

Caching ist der häufigste Redis-Anwendungsfall. Sie speichern häufig abgerufene Daten in Redis, um die Datenbanklast zu reduzieren und Antworten zu beschleunigen.

Grundlegendes Cache-Aside-Muster

Beim Cache-Aside-Muster prüft Ihre Anwendung zuerst Redis. Wenn die Daten nicht vorhanden sind (Cache-Miss), werden sie aus der Datenbank abgerufen und dann für das nächste Mal in Redis gespeichert.

import redis
import json

r = redis.Redis(host='localhost', port=6379, db=0)

def get_user(user_id):
    cache_key = f'user:{user_id}'
    cached = r.get(cache_key)
    if cached:
        return json.loads(cached)
    user = fetch_user_from_db(user_id)  # your DB call
    r.setex(cache_key, 3600, json.dumps(user))  # cache for 1 hour
    return user

Best Practices:

• Legen Sie immer eine Ablaufzeit (TTL) fest, um veraltete Daten zu vermeiden.
• Verwenden Sie aussagekräftige Schlüsselnamen wie user:123:profile.
• Behandeln Sie die Cache-Invalidierung: Wenn sich Daten ändern, aktualisieren oder löschen Sie den Cache-Eintrag.
• Erwägen Sie die Verwendung von Redis-Hashes zum Speichern mehrerer Felder eines Objekts, um Speicher zu sparen.

Fortgeschrittene Caching-Muster

Für Anwendungen mit hohem Datenverkehr sollten Sie diese Muster in Betracht ziehen:

• Write-through: Cache und Datenbank gemeinsam aktualisieren.
• Write-behind: Zuerst den Cache aktualisieren, dann asynchron die Datenbank.
• Refresh-ahead: Cache-Einträge proaktiv aktualisieren, bevor sie ablaufen.

Queues mit Redis

Redis-Listen und -Streams eignen sich perfekt für den Aufbau einfacher Message-Queues. Verwenden Sie Listen für einfache Job-Queues und Streams für komplexere Ereignisverarbeitung.

Einfache Job-Queue mit Listen

Verwenden Sie LPUSH, um Jobs hinzuzufügen, und BRPOP, um sie zu verarbeiten. BRPOP blockiert, bis ein Job verfügbar ist, was es effizient für Worker macht.

# Producer
r.lpush('jobs', json.dumps({'task': 'send_email', 'to': 'user@example.com'}))

# Consumer (worker)
while True:
    _, job_data = r.brpop('jobs', timeout=30)
    job = json.loads(job_data)
    process_job(job)

Best Practices:

• Verwenden Sie separate Listen für verschiedene Job-Typen oder Prioritäten.
• Implementieren Sie eine Dead-Letter-Queue für fehlgeschlagene Jobs.
• Verwenden Sie BRPOPLPUSH (oder BLMOVE in Redis 6.2+), um Jobs atomar in eine Verarbeitungsliste zu verschieben und so die Zuverlässigkeit zu gewährleisten.
• Überwachen Sie die Warteschlangenlänge mit LLEN, um Worker zu skalieren.

Redis Streams für Event Sourcing

Redis Streams (eingeführt in Redis 5.0) bieten eine robustere Queue mit Consumer Groups, Bestätigungen und Replay. Verwenden Sie XADD, um Ereignisse hinzuzufügen, und XREADGROUP, um sie zu konsumieren.

# Add event
r.xadd('events', {'type': 'order_placed', 'order_id': '123'})

# Consumer group
r.xgroup_create('events', 'order_processors', id='0', mkstream=True)
while True:
    events = r.xreadgroup('order_processors', 'worker1', {'events': '>'}, count=10, block=5000)
    for event in events:
        process_event(event)
        r.xack('events', 'order_processors', event.id)

Rate Limiting mit Redis

Rate Limiting schützt Ihre APIs vor Missbrauch und gewährleistet eine faire Nutzung. Die atomaren Operationen von Redis machen es ideal zum Zählen von Anfragen.

Fixed Window Rate Limiting

Erhöhen Sie einen Zähler für jeden Benutzer und legen Sie eine Ablaufzeit fest. Wenn der Zähler das Limit überschreitet, blockieren Sie die Anfrage.

def is_allowed(user_id, limit=100, window=60):
    key = f'rate_limit:{user_id}'
    current = r.incr(key)
    if current == 1:
        r.expire(key, window)
    return current