patrones-redis

Patrones de Redis

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

Cómo Funciona

Redis es un almacén de estructuras de datos en memoria que admite strings, 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 múltiples pasos requieren scripts Lua, transacciones MULTI/EXEC o sincronización explícita para mantenerse atómicos. Los datos se persisten opcionalmente mediante instantáneas RDB o registros AOF. Los clientes se comunican a través de TCP usando el protocolo RESP; los pools de conexiones son esenciales para evitar la sobrecarga de establecer una conexión por cada petición.

Cuándo Activar

• Agregar caché a una aplicación
• Implementar limitación de tasa o estrangulamiento
• 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	Sorted Set	puntuaciones:semanal
Visitantes únicos	Set	visitantes:2024-01-01
Feed de actividad	List	feed:usuario:456
Flujo de eventos	Stream	eventos:pedidos
Contadores / límites de tasa	String (INCR)	limite_tasa:usuario:123
Filtro Bloom / HLL	HyperLogLog	hll:paginas_vistas

Patrones Principales

Caché-Aside (Carga Diferida)

import redis
import json

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

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

    if en_cache:
        return json.loads(en_cache)

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

Caché Write-Through

def actualizar_producto(producto_id: int, datos: dict):
    # Escribir en la BD primero
    db.execute("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: agrupa claves relacionadas bajo un conjunto
def cache_producto(producto_id: int, categoria_id: int, datos: dict):
    clave = f"producto:{producto_id}"
    etiqueta = f"etiqueta:categoria:{categoria_id}"
    tuberia = r.pipeline(transaction=True)
    tuberia.setex(clave, 3600, json.dumps(datos))
    tuberia.sadd(etiqueta, clave)
    tuberia.expire(etiqueta, 3600)
    tuberia.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}"
    tuberia = r.pipeline(transaction=True)
    tuberia.hset(clave, mapping={
        "usuario_id": usuario_id,
        "creado_en": int(time.time()),
    })
    tuberia.expire(clave, ttl)
    tuberia.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 se_limita_tasa(usuario_id: int, limite: int = 100, ventana: int = 60) -> bool:
    clave = f"limite_tasa:{usuario_id}:{int(time.time()) // ventana}"
    tuberia = r.pipeline(transaction=True)
    tuberia.incr(clave)
    tuberia.expire(clave, ventana)
    cuenta, _ = tuberia.execute()
    return cuenta > 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 seq = redis.call('INCR', clave_sec)
    redis.call('EXPIRE', clave_sec, math.ceil(ventana / 1000))
    redis.call('ZADD', clave, ahora, ahora .. '-' .. seq)
    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"limite_tasa: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 if 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 multi-nodo use la biblioteca redlock-py que implementa el algoritmo Redlock completo.

Pub/Sub y Streams

Pub/Sub (Fire-and-Forget)

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

# Suscriptor (bloqueante — ejecutar en hilo/proceso separado)
def suscribirse_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 Durable)

# 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:limite_tasa:usuario:123

# Patrón: recurso:fecha (claves con límite temporal)
estadisticas:paginas_vistas:2024-01-01

Estrategia de TTL

Tipo de Dato	TTL Sugerido
Sesión de usuario	24h (86400)
Caché de respuesta de 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

Establezca siempre un TTL. Las claves sin TTL se acumulan indefinidamente y causan presión sobre la 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('mymaestro', decode_responses=True)
replica = sentinel.slave_for('mymaestro', 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	Desaloja el menos usado recientemente	Caché general
volatile-lru	LRU solo entre claves con TTL	Almacén de datos mixto
allkeys-lfu	Desaloja el menos usado frecuentemente	Patrones de acceso sesgado
volatile-ttl	Desaloja la que expira antes	Priorizar datos de larga duración

Configure mediante redis.conf: maxmemory-policy allkeys-lru

Antipatrones

Antipatrón	Problema	Solución
Claves sin TTL	La memoria crece sin límite	Establezca siempre TTL
KEYS * en producción	Bloquea el servidor (O(N))	Use el cursor SCAN
Almacenar blobs grandes (>100KB)	Serialización lenta, presión de memoria	Almacene referencia + obtenga del almacén de objetos
Redis único para todo	Sin aislamiento entre caché y cola	Use BD o instancias separadas
Ignorar límites del pool de conexiones	Agotamiento de conexiones bajo carga	Ajuste el pool a la carga de trabajo
No manejar estampida de fallos de caché	Embudo en arranque en frío	Use bloqueos o caducidad temprana probabilística
FLUSHALL sin pensarlo	Borra toda la instancia	Limite las eliminaciones por patrón de clave

Prevención de Estampida de Fallos de Caché

import threading

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

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

    with _bloqueos_mutex:
        if clave not in _bloqueos:
            _bloqueos[clave] = threading.Lock()
        bloqueo = _bloqueos[clave]
    with bloqueo:
        en_cache = r.get(clave)  # Revisar de nuevo después de adquirir el bloqueo
        if en_cache:
            return json.loads(en_cache)
        valor = funcion_obtener()
        r.setex(clave, ttl, json.dumps(valor))
        return valor

Nota: para despliegues 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 caché-aside con setex y un TTL de 5 minutos en la respuesta. Clave según 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 estrangulamiento preciso por usuario.

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

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

Referencia Rápida

Patrón	Cuándo Usar
Caché-aside	Carga pesada de lectura, tolera ligera obsolescencia
Write-through	Se requiere consistencia fuerte
Bloqueo distribuido	Evitar acceso concurrente a un recurso
Límite de tasa de ventana deslizante	Estrangulamiento preciso por usuario
Redis Streams	Cola de eventos durable con grupos de consumidores
Pub/Sub	Difusión sin garantías de entrega necesarias
Tabla de clasificación con Sorted Set	Puntuación clasificada, paginación
HyperLogLog	Conteo aproximado de únicos con bajo consumo de memoria

Relacionados

• Skill: patrones-postgres — patrones de datos relacionales
• Skill: patrones-backend — patrones de capa de API y 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 base de datos