patrones-redis

Patrones de Redis

Referencia rápida de las mejores prácticas de Redis para casos de uso comunes en backend.

Cómo funciona

Redis es un almacén de estructuras de datos en memoria que admite cadenas, hashes, listas, conjuntos, conjuntos ordenados, streams y más. Los comandos individuales de Redis son atómicos en una sola instancia; los flujos de trabajo de varios pasos requieren scripts Lua, transacciones MULTI/EXEC o sincronización explícita para mantener la atomicidad. Los datos se persisten opcionalmente mediante instantáneas RDB o registros AOF. Los clientes se comunican a través de TCP utilizando el protocolo RESP; los pools de conexiones son esenciales para evitar la sobrecarga del protocolo de enlace por solicitud.

Cuándo activar

• Agregar caché a una aplicación
• Implementar limitación de tasa o control de acceso
• Construir bloqueos distribuidos o coordinación
• Configurar almacenamiento de sesiones o tokens
• Usar Pub/Sub o Redis Streams para mensajería
• Configurar Redis en producción (pooling, desalojo, clustering)

Hoja de referencia de estructuras de datos

Caso de uso	Estructura	Clave de ejemplo
Caché simple	String	producto:123
Sesión de usuario	Hash	sesion:abc
Tabla de clasificación	Conjunto ordenado	puntuaciones:semanal
Visitantes únicos	Conjunto	visitantes:2024-01-01
Feed de actividad	Lista	feed:usuario:456
Flujo de eventos	Stream	eventos:pedidos
Contadores / límites de tasa	String (INCR)	limitetasa:usuario:123
Filtro Bloom / HLL	HyperLogLog	hll:vistaspagina

Patrones principales

Cache-Aside (Carga diferida)

import redis
import json

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

def obtener_producto(producto_id: int):
    clave_cache = f"producto:{producto_id}"
    cacheado = r.get(clave_cache)

    if cacheado:
        return json.loads(cacheado)

    producto = db.consulta("SELECT * FROM productos WHERE id = %s", producto_id)
    r.setex(clave_cache, 3600, json.dumps(producto))  # TTL: 1 hora
    return producto

Caché de escritura directa

def actualizar_producto(producto_id: int, datos: dict):
    # Escribir en la base de datos primero
    db.ejecutar("UPDATE productos SET ... WHERE id = %s", producto_id)

    # Actualizar caché inmediatamente
    clave_cache = f"producto:{producto_id}"
    r.setex(clave_cache, 3600, json.dumps(datos))

Invalidación de caché

# Invalidación basada en etiquetas: agrupar claves relacionadas bajo un conjunto
def cachear_producto(producto_id: int, categoria_id: int, datos: dict):
    clave = f"producto:{producto_id}"
    etiqueta = f"etiqueta:categoria:{categoria_id}"
    pipe = r.pipeline(transaction=True)
    pipe.setex(clave, 3600, json.dumps(datos))
    pipe.sadd(etiqueta, clave)
    pipe.expire(etiqueta, 3600)
    pipe.execute()

def invalidar_categoria(categoria_id: int):
    etiqueta = f"etiqueta:categoria:{categoria_id}"
    claves = r.smembers(etiqueta)
    if claves:
        r.delete(*claves)
    r.delete(etiqueta)

Almacenamiento de sesiones

import time
import uuid

def crear_sesion(usuario_id: int, ttl: int = 86400) -> str:
    sesion_id = str(uuid.uuid4())
    clave = f"sesion:{sesion_id}"
    pipe = r.pipeline(transaction=True)
    pipe.hset(clave, mapping={
        "usuario_id": usuario_id,
        "creado_en": int(time.time()),
    })
    pipe.expire(clave, ttl)
    pipe.execute()
    return sesion_id

def obtener_sesion(sesion_id: str) -> dict | None:
    datos = r.hgetall(f"sesion:{sesion_id}")
    return datos if datos else None

def eliminar_sesion(sesion_id: str):
    r.delete(f"sesion:{sesion_id}")

Limitación de tasa

Ventana fija (Simple)

def esta_limitado(usuario_id: int, limite: int = 100, ventana: int = 60) -> bool:
    clave = f"limitetasa:{usuario_id}:{int(time.time()) // ventana}"
    pipe = r.pipeline(transaction=True)
    pipe.incr(clave)
    pipe.expire(clave, ventana)
    contador, _ = pipe.execute()
    return contador > limite

Ventana deslizante (Lua — Atómico)

-- sliding_window.lua
local clave = KEYS[1]
local ahora = tonumber(ARGV[1])
local ventana = tonumber(ARGV[2])
local limite = tonumber(ARGV[3])

redis.call('ZREMRANGEBYSCORE', clave, 0, ahora - ventana)
local cuenta = redis.call('ZCARD', clave)

if cuenta < limite then
    -- Usar miembro único (ahora + secuencia) para evitar colisiones en el mismo milisegundo
    local clave_sec = clave .. ':seq'
    local sec = redis.call('INCR', clave_sec)
    redis.call('EXPIRE', clave_sec, math.ceil(ventana / 1000))
    redis.call('ZADD', clave, ahora, ahora .. '-' .. sec)
    redis.call('EXPIRE', clave, math.ceil(ventana / 1000))
    return 1
end
return 0

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

def permitir_solicitud(usuario_id: int) -> bool:
    clave = f"limitetasa:deslizante:{usuario_id}"
    ahora = int(time.time() * 1000)
    return bool(ventana_deslizante(keys=[clave], args=[ahora, 60000, 100]))

Bloqueos distribuidos

Bloqueo distribuido (Nodo único — SET NX PX)

import uuid

def adquirir_bloqueo(recurso: str, ttl_ms: int = 5000) -> str | None:
    clave_bloqueo = f"bloqueo:{recurso}"
    token = str(uuid.uuid4())
    adquirido = r.set(clave_bloqueo, token, px=ttl_ms, nx=True)
    return token si adquirido else None

def liberar_bloqueo(recurso: str, token: str) -> bool:
    script_liberar = """
    if redis.call('get', KEYS[1]) == ARGV[1] then
        return redis.call('del', KEYS[1])
    else
        return 0
    end
    """
    resultado = r.eval(script_liberar, 1, f"bloqueo:{recurso}", token)
    return bool(resultado)

# Uso
token = adquirir_bloqueo("pedido:pago:123")
if token:
    try:
        procesar_pago()
    finally:
        liberar_bloqueo("pedido:pago:123", token)

Para configuraciones de varios nodos, use la biblioteca redlock-py que implementa el algoritmo Redlock completo.

Pub/Sub y Streams

Pub/Sub (Disparar y olvidar)

# Publicador
def publicar_evento(canal: str, carga: dict):
    r.publish(canal, json.dumps(carga))

# Suscriptor (bloqueante — ejecutar en hilo/proceso separado)
def suscribir_eventos(canal: str):
    pubsub = r.pubsub()
    pubsub.subscribe(canal)
    for mensaje in pubsub.listen():
        if mensaje['type'] == 'message':
            manejar(json.loads(mensaje['data']))

Redis Streams (Cola duradera)

# Productor
def emitir(stream: str, evento: dict):
    r.xadd(stream, evento, maxlen=10000)  # Limitar longitud del stream

# Grupo de consumidores — garantiza entrega al menos una vez
try:
    r.xgroup_create('eventos:pedidos', 'procesador', id='0', mkstream=True)
except Exception:
    pass  # El grupo ya existe

def consumir(stream: str, grupo: str, consumidor: str):
    while True:
        mensajes = r.xreadgroup(grupo, consumidor, {stream: '>'}, count=10, block=2000)
        for _, entradas in (mensajes or []):
            for msg_id, datos in entradas:
                procesar(datos)
                r.xack(stream, grupo, msg_id)

Prefiera Streams sobre Pub/Sub cuando necesite garantías de entrega, grupos de consumidores o reproducción.

Diseño de claves

Convenciones de nomenclatura

# Patrón: recurso:id:campo
usuario:123:perfil
pedido:456:estado
cache:producto:789

# Patrón: espacio_nombres:recurso:id
miapp:sesion:abc123
miapp:limitetasa:usuario:123

# Patrón: recurso:fecha (claves con límite de tiempo)
estadisticas:vistaspagina:2024-01-01

Estrategia de TTL

Tipo de datos	TTL sugerido
Sesión de usuario	24h (86400)
Caché de respuesta API	5–15 min
Ventana de límite de tasa	Coincidir con el tamaño de la ventana
Tokens de corta duración	5–10 min
Tabla de clasificación	1h–24h
Datos estáticos/de referencia	1h–1 semana

Siempre establezca un TTL. Las claves sin TTL se acumulan indefinidamente y causan presión de memoria.

Gestión de conexiones

Pool de conexiones

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)

Modo clúster

from redis.cluster import RedisCluster

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

Sentinel (Alta disponibilidad)

from redis.sentinel import Sentinel

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

Políticas de desalojo

Política	Comportamiento	Mejor para
noeviction	Error al escribir cuando está lleno	Colas / datos críticos
allkeys-lru	Desalojar el usado menos recientemente	Caché general
volatile-lru	LRU solo entre claves con TTL	Almacén de datos mixto
allkeys-lfu	Desalojar el usado con menos frecuencia	Patrones de acceso sesgados
volatile-ttl	Desalojar el que expira más pronto	Priorizar datos de larga duración

Establecer en redis.conf: maxmemory-policy allkeys-lru

Antipatrones

Antipatrón	Problema	Solución
Claves sin TTL	La memoria crece sin límite	Siempre establecer TTL
KEYS * en producción	Bloquea el servidor (O(N))	Usar cursor SCAN
Almacenar grandes blobs (>100KB)	Serialización lenta, presión de memoria	Almacenar referencia + obtener desde almacén de objetos
Un solo Redis para todo	Sin aislamiento entre caché y cola	Usar bases de datos o instancias separadas
Ignorar límites del pool de conexiones	Agotamiento de conexiones bajo carga	Dimensionar el pool según la carga de trabajo
No manejar estampida de fallos de caché	Avalancha en inicio en frío	Usar bloqueos o caducidad temprana probabilística
FLUSHALL sin pensar	Limpia toda la instancia	Limitar eliminaciones por patrón de clave

Prevención de estampida de fallos de caché

import threading

_bloqueos: dict[str, threading.Lock] = {}
_mutex_bloqueos = threading.Lock()

def obtener_con_bloqueo(clave: str, funcion_obtener, ttl: int = 300):
    cacheado = r.get(clave)
    if cacheado:
        return json.loads(cacheado)

    with _mutex_bloqueos:
        if clave not in _bloqueos:
            _bloqueos[clave] = threading.Lock()
        bloqueo = _bloqueos[clave]
    with bloqueo:
        cacheado = r.get(clave)  # Volver a verificar después de adquirir el bloqueo
        if cacheado:
            return json.loads(cacheado)
        valor = funcion_obtener()
        r.setex(clave, ttl, json.dumps(valor))
        return valor

Nota: para implementaciones multiproceso, reemplace el bloqueo en proceso con adquirir_bloqueo/liberar_bloqueo de la sección de Bloqueos distribuidos anterior.

Ejemplos

Agregar caché a un endpoint de API Django/Flask: Use cache-aside con setex y un TTL de 5 minutos en la respuesta. Clave basada en los parámetros de la solicitud.

Limitar la tasa de una API por usuario: Use ventana fija con pipeline(transaction=True) para endpoints de bajo tráfico; use Lua de ventana deslizante para una limitación precisa por usuario.

Coordinar un trabajo en segundo plano entre workers: Use adquirir_bloqueo con un TTL que supere la duración esperada del trabajo. Siempre libere en un bloque finally.

Distribuir notificaciones a múltiples suscriptores: Use Pub/Sub para disparar y olvidar. Cambie a Streams si necesita entrega garantizada o reproducción para consumidores tardíos.

Referencia rápida

Patrón	Cuándo usar
Cache-aside	Carga pesada de lectura, tolera ligera desactualización
Escritura directa	Se requiere consistencia fuerte
Bloqueo distribuido	Prevenir acceso concurrente a un recurso
Límite de tasa de ventana deslizante	Limitación precisa por usuario
Redis Streams	Cola de eventos duradera con grupos de consumidores
Pub/Sub	Transmisión sin necesidad de garantías de entrega
Tabla de clasificación con conjunto ordenado	Puntuación clasificada, paginación
HyperLogLog	Conteo único aproximado con bajo consumo de memoria

Relacionados

• Skill: patrones-postgres — patrones de datos relacionales
• Skill: patrones-backend — patrones de API y capa de servicio
• Skill: migraciones-base-datos — versionado de esquemas
• Skill: patrones-django — integración del framework de caché de Django
• Agente: revisor-base-datos — flujo de trabajo completo de revisión de bases de datos