redis-patterns

Modèles Redis

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

Fonctionnement

Redis est un magasin de structures 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, etc. Les commandes Redis individuelles sont atomiques sur une instance unique ; 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 la négociation par requête.

Quand activer

• Ajout de cache à une application
• Implémentation de la limitation de débit ou du throttling
• Construction de verrous distribués ou de coordination
• Mise en place de stockage de sessions ou de jetons
• Utilisation de Pub/Sub ou de Redis Streams pour la messagerie
• Configuration de Redis en production (pooling, éviction, clustering)

Aide-mémoire des structures de données

Cas d'utilisation	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 Bloom / HLL	HyperLogLog	hll:pageviews

Modèles 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 de données
    db.execute("UPDATE products SET ... WHERE id = %s", product_id)

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

Invalidation de cache

# Invalidation basée sur les tags — regrouper les clés connexes 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)

# Utilisation
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 & Flux

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']))

Flux Redis (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 — garantit une 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)

Privilégiez les Flux au Pub/Sub lorsque vous avez besoin de garanties de livraison, de groupes de consommateurs ou de relecture.

Conception des clés

Conventions de nommage

# Modèle : resource:id:field
user:123:profile
order:456:status
cache:product:789

# Modèle : namespace:resource:id
myapp:session:abc123
myapp:ratelimit:user:123

# Modèle : resource: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 de courte durée	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 entraînent une pression sur la 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 lorsque la mémoire est pleine	Files d'attente / données critiques
allkeys-lru	Éviction des moins récemment utilisées	Cache général
volatile-lru	LRU uniquement parmi les clés avec TTL	Magasin de données mixte
allkeys-lfu	Éviction des moins fréquemment utilisées	Modèles d'accès asymétriques
volatile-ttl	Éviction des clés avec la plus courte durée de vie restante	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 et récupérer depuis un magasin d'objets
Un seul Redis pour tout	Pas d'isolation entre cache et file d'attente	Utiliser des bases de données ou instances séparées
Ignorer les limites du pool de connexions	Épuisement des connexions sous charge	Dimensionner le pool en fonction de la charge de travail
Ne pas gérer une ruée vers le cache en cas d'absence	Ruée sur un cache froid	Utiliser des verrous ou une expiration précoce probabiliste
FLUSHALL sans réflexion	Efface toute l'instance	Limiter les suppressions par modèle de clé

Prévention de la ruée en cas d'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 d'API Django/Flask : Utilisez cache-aside avec setex et un TTL de 5 minutes sur la réponse. Clé basée sur les paramètres de la requête.

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

Coordonner un travail en arrière-plan entre les 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 Flux si vous avez besoin d'une livraison garantie ou d'une relecture pour les consommateurs tardifs.

Référence rapide

Modèle	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 par fenêtre glissante	Throttling précis par utilisateur
Flux Redis	File d'attente d'événements durable avec groupes de consommateurs
Pub/Sub	Diffusion sans garantie de livraison
Classement par ensemble trié	Score classé, pagination
HyperLogLog	Comptage unique approximatif à faible empreinte mémoire

Connexe

• Compétence : postgres-patterns — modèles de données relationnelles
• Compétence : backend-patterns — modèles de couche API et service
• Compétence : database-migrations — versionnement de schéma
• Compétence : django-patterns — intégration du framework de cache Django
• Agent : database-reviewer — workflow complet de révision de base de données