redis-patterns

Redis Patterns

Référence rapide des bonnes pratiques Redis pour les cas d'usage backend courants.

Comment ça fonctionne

Redis est un magasin de données en mémoire qui prend en charge les chaînes, les hachages, les listes, les ensembles, les ensembles triés, les flux et plus encore. Les commandes Redis individuelles sont atomiques sur une seule instance ; les workflows en plusieurs étapes nécessitent des scripts Lua, des transactions MULTI/EXEC ou une synchronisation explicite pour rester atomiques. Les données sont éventuellement persistées via des instantanés RDB ou des journaux AOF. Les clients communiquent via TCP en utilisant le protocole RESP ; les pools de connexions sont essentiels pour éviter la surcharge de négociation par requête.

Quand activer

• Ajouter un cache à une application
• Mettre en œuvre la limitation de débit ou le throttling
• Construire des verrous distribués ou de la coordination
• Mettre en place un stockage de sessions ou de jetons
• Utiliser Pub/Sub ou Redis Streams pour la messagerie
• Configurer Redis en production (pooling, éviction, clustering)

Aide-mémoire des structures de données

Cas d'usage	Structure	Exemple de clé
Cache simple	String	product:123
Session utilisateur	Hash	session:abc
Classement	Sorted Set	scores:weekly
Visiteurs uniques	Set	visitors:2024-01-01
Flux d'activité	List	feed:user:456
Flux d'événements	Stream	events:orders
Compteurs / limites de débit	String (INCR)	ratelimit:user:123
Filtre de Bloom / HLL	HyperLogLog	hll:pageviews

Motifs principaux

Cache-Aside (Chargement paresseux)

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 heure
    return product

Cache Write-Through

def update_product(product_id: int, data: dict):
    # Écrire d'abord dans la base
    db.execute("UPDATE products SET ... WHERE id = %s", product_id)

    # Mettre à jour le cache immédiatement
    cache_key = f"product:{product_id}"
    r.setex(cache_key, 3600, json.dumps(data))

Invalidation de cache

# Invalidation par tag — regrouper les clés associées sous un ensemble
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)

Stockage de sessions

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}")

Limitation de débit

Fenêtre fixe (simple)

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

Fenêtre glissante (Lua — Atomique)

-- 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
    -- Utiliser un membre unique (now + séquence) pour éviter les collisions dans la même milliseconde
    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]))

Verrous distribués

Verrou distribué (nœud unique — 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)

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

Pour les configurations multi-nœuds, utilisez la bibliothèque redlock-py qui implémente l'algorithme Redlock complet.

Pub/Sub & Streams

Pub/Sub (Fire-and-Forget)

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

# Abonné (bloquant — exécuter dans un thread/processus séparé)
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 (File d'attente durable)

# Producteur
def emit(stream: str, event: dict):
    r.xadd(stream, event, maxlen=10000)  # Limiter la longueur du flux

# Groupe de consommateurs — garantie de livraison au moins une fois
try:
    r.xgroup_create('events:orders', 'processor', id='0', mkstream=True)
except Exception:
    pass  # Le groupe existe déjà

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)

Préférez les Streams à Pub/Sub lorsque vous avez besoin de garanties de livraison, de groupes de consommateurs ou de rejeu.

Conception des clés

Conventions de nommage

# Motif : ressource:id:champ
user:123:profile
order:456:status
cache:product:789

# Motif : espace de noms:ressource:id
myapp:session:abc123
myapp:ratelimit:user:123

# Motif : ressource:date (clés limitées dans le temps)
stats:pageviews:2024-01-01

Stratégie TTL

Type de données	TTL suggéré
Session utilisateur	24h (86400)
Cache de réponse API	5–15 min
Fenêtre de limitation de débit	Correspondre à la taille de la fenêtre
Jetons éphémères	5–10 min
Classement	1h–24h
Données statiques/de référence	1h–1 semaine

Définissez toujours un TTL. Les clés sans TTL s'accumulent indéfiniment et causent une pression mémoire.

Gestion des connexions

Pool de connexions

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)

Mode cluster

from redis.cluster import RedisCluster

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

Sentinel (Haute disponibilité)

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)

Politiques d'éviction

Politique	Comportement	Meilleur pour
noeviction	Erreur en écriture quand plein	Files d'attente / données critiques
allkeys-lru	Éviction du moins récemment utilisé	Cache général
volatile-lru	LRU uniquement parmi les clés avec TTL	Magasin de données mixte
allkeys-lfu	Éviction du moins fréquemment utilisé	Modèles d'accès asymétriques
volatile-ttl	Éviction du plus proche de l'expiration	Prioriser les données à longue durée de vie

Définir via redis.conf : maxmemory-policy allkeys-lru

Anti-patrons

Anti-patron	Problème	Correctif
Clés sans TTL	La mémoire croît indéfiniment	Toujours définir un TTL
KEYS * en production	Bloque le serveur (O(N))	Utiliser le curseur SCAN
Stocker de gros blobs (>100 Ko)	Sérialisation lente, pression mémoire	Stocker une référence + récupérer depuis un stockage d'objets
Utiliser Redis pour tout	Pas d'isolation entre cache & file d'attente	Utiliser des bases ou instances séparées
Ignorer les limites du pool de connexions	Épuisement des connexions sous charge	Dimensionner le pool pour la charge de travail
Ne pas gérer la ruée sur l'absence de cache	Effet de troupeau au démarrage à froid	Utiliser des verrous ou une expiration précoce probabiliste
FLUSHALL sans réflexion	Efface toute l'instance	Limiter les suppressions par motif de clé

Prévention de la ruée sur l'absence de cache

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)  # Revérifier après avoir acquis le verrou
        if cached:
            return json.loads(cached)
        value = fetch_fn()
        r.setex(key, ttl, json.dumps(value))
        return value

Remarque : pour les déploiements multi-processus, remplacez le verrou intra-processus par acquire_lock/release_lock de la section Verrous distribués ci-dessus.

Exemples

Ajouter un cache à un point de terminaison API Django/Flask : Utilisez le cache-aside avec setex et un TTL de 5 minutes sur la réponse. Clé sur les paramètres de la requête.

Limiter le débit d'une API par utilisateur : Utilisez la fenêtre fixe avec pipeline(transaction=True) pour les points de terminaison à faible trafic ; utilisez le Lua à fenêtre glissante pour un throttling précis par utilisateur.

Coordonner un travail en arrière-plan entre plusieurs workers : Utilisez acquire_lock avec un TTL qui dépasse la durée prévue du travail. Libérez toujours dans un bloc finally.

Diffuser des notifications à plusieurs abonnés : Utilisez Pub/Sub pour le fire-and-forget. Passez aux Streams si vous avez besoin d'une livraison garantie ou d'un rejeu pour les consommateurs tardifs.

Référence rapide

Motif	Quand l'utiliser
Cache-aside	Lecture intensive, tolère une légère obsolescence
Write-through	Cohérence forte requise
Verrou distribué	Empêcher l'accès concurrent à une ressource
Limitation de débit fenêtre glissante	Throttling précis par utilisateur
Redis Streams	File d'attente d'événements durable avec groupes de consommateurs
Pub/Sub	Diffusion sans garantie de livraison nécessaire
Classement par ensemble trié	Notation classée, pagination
HyperLogLog	Comptage approximatif unique à faible mémoire

Liens connexes

• Skill : postgres-patterns — motifs de données relationnelles
• Skill : backend-patterns — motifs de couche API et service
• Skill : database-migrations — versionnement de schéma
• Skill : django-patterns — intégration du cache Django
• Agent : database-reviewer — processus complet de révision de base de données