StackPractices
EN
intermediate Por Mathias Paulenko

Patrones de Cache de Redis para Aplicaciones de Alto

Como implementar patrones cache-aside, write-through y write-behind con Redis para reducir carga de base de datos y mejorar tiempos de respuesta

redis cache database performance
Temas: databases performance

Redis es un almacen de estructuras de datos en memoria que funciona como una capa de cache extremadamente rapida entre tu aplicacion y la base de datos persistente. Elegir el patron de cache correcto — cache-aside, write-through o write-behind — determina como tu aplicacion maneja cache misses, consistencia y escenarios de fallo.

Cuando Usar Esto

  • Las consultas a base de datos son lentas y devuelven datos frecuentemente accedidos. Consulta Query Optimization para ajustar queries lentos.
  • Necesitas reducir carga en bases de datos primarias durante picos de trafico. Consulta Rate Limiting para control de tráfico.
  • La temporal staleness de datos es aceptable a cambio de menor latencia

Requisitos Previos

  • Servidor Redis ejecutandose localmente o via servicio administrado
  • Una libreria cliente como ioredis o redis para Node.js

Solucion

1. Cache-Aside (Lazy Loading)

La aplicacion revisa el cache primero. En un miss, carga desde la base de datos y pobla el cache.

// cache/CacheAside.ts
import Redis from 'ioredis';

class CacheAsideProductRepository {
  private redis = new Redis();
  private ttl = 300; // 5 minutos

  async getProduct(id: string): Promise<Product | null> {
    const cacheKey = `product:${id}`;
    
    // Revisar cache primero
    const cached = await this.redis.get(cacheKey);
    if (cached) {
      return JSON.parse(cached);
    }

    // Cache miss: cargar desde base de datos
    const product = await this.db.query('SELECT * FROM products WHERE id = $1', [id]);
    if (!product) return null;

    // Poblar cache
    await this.redis.setex(cacheKey, this.ttl, JSON.stringify(product));
    return product;
  }

  async updateProduct(id: string, data: Partial<Product>): Promise<void> {
    await this.db.query('UPDATE products SET ... WHERE id = $1', [id]);
    // Invalidar cache para prevenir lecturas stale
    await this.redis.del(`product:${id}`);
  }
}

2. Write-Through

Los datos se escriben tanto al cache como a la base de datos simultaneamente. El cache siempre tiene los datos mas recientes.

// cache/WriteThrough.ts
class WriteThroughProductRepository {
  async updateProduct(id: string, data: Partial<Product>): Promise<void> {
    const cacheKey = `product:${id}`;

    // Iniciar transaccion de base de datos. Consulta [Database Transactions](/recipes/database-transactions/) para patrones ACID.
    await this.db.query('BEGIN');
    try {
      await this.db.query('UPDATE products SET ... WHERE id = $1', [id]);
      
      // Escribir al cache dentro de la misma operacion logica
      const updated = await this.db.query('SELECT * FROM products WHERE id = $1', [id]);
      await this.redis.setex(cacheKey, this.ttl, JSON.stringify(updated));
      
      await this.db.query('COMMIT');
    } catch (error) {
      await this.db.query('ROLLBACK');
      throw error;
    }
  }
}

3. Write-Behind (Write-Back)

Los datos se escriben primero al cache y se flushean asincronicamente a la base de datos. Mayor rendimiento pero mas riesgoso.

// cache/WriteBehind.ts
class WriteBehindProductRepository {
  async updateProduct(id: string, data: Partial<Product>): Promise<void> {
    const cacheKey = `product:${id}`;

    // Escribir al cache inmediatamente
    await this.redis.setex(cacheKey, this.ttl, JSON.stringify(data));

    // Encolar para persistencia async
    await this.redis.lpush('pending_writes', JSON.stringify({ id, data, timestamp: Date.now() }));
  }
}

// Worker en background. Consulta [Batch Processing](/recipes/batch-processing-patterns/) para patrones de jobs.
async function flushPendingWrites() {
  const batch = await redis.lpop('pending_writes', 100);
  if (!batch) return;

  const writes = batch.map(item => JSON.parse(item));
  
  await db.query('BEGIN');
  try {
    for (const write of writes) {
      await db.query('UPDATE products SET ... WHERE id = $1', [write.id]);
    }
    await db.query('COMMIT');
  } catch (error) {
    await db.query('ROLLBACK');
    // Re-encolar escrituras fallidas
    for (const write of writes) {
      await redis.rpush('pending_writes', JSON.stringify(write));
    }
  }
}

// Ejecutar cada 5 segundos
setInterval(flushPendingWrites, 5000);

4. Prevencion de Cache Stampede

// cache/StampedeProtection.ts
class StampedeProtectedCache {
  async getProduct(id: string): Promise<Product> {
    const cacheKey = `product:${id}`;
    const lockKey = `lock:${id}`;

    const cached = await this.redis.get(cacheKey);
    if (cached) return JSON.parse(cached);

    // Intentar adquirir lock
    const lock = await this.redis.set(lockKey, '1', 'EX', 10, 'NX');
    if (lock) {
      // Ganamos la carrera; cargar desde DB
      const product = await this.db.query('SELECT * FROM products WHERE id = $1', [id]);
      await this.redis.setex(cacheKey, this.ttl, JSON.stringify(product));
      await this.redis.del(lockKey);
      return product;
    }

    // Esperar a que el ganador pobla el cache
    await new Promise(resolve => setTimeout(resolve, 100));
    return this.getProduct(id);
  }
}

Como Funciona

  1. Cache-Aside minimiza escrituras de cache pero permite breves datos stale despues de actualizaciones
  2. Write-Through garantiza consistencia al costo de mayor latencia de escritura
  3. Write-Behind maximiza throughput pero riesga perdida de datos si el cache falla antes del flush
  4. Stampede Protection previene multiples consultas simultaneas a base de datos en expiracion de cache

Consideraciones de Produccion

  • Usa Redis Cluster o Redis Sentinel para alta disponibilidad
  • Implementa logica de circuit breaker cuando Redis no esta disponible; fallback a base de datos
  • Configura valores de TTL apropiados basados en frecuencia de cambio de datos
  • Monitorea cache hit ratio con INFO stats y ajusta TTL en consecuencia

Errores Comunes

  • No manejar fallos de conexion a Redis gracefulmente
  • Usar el mismo TTL para todos los tipos de datos sin importar frecuencia de cambio
  • Olvidar invalidar entradas de cache relacionadas en actualizaciones

Tips de Rendimiento

  1. Usa MSET y MGET para operaciones batch. Reduce round-trips al setear o obtener múltiples keys:
# Mal: 100 round-trips
for i in range(100):
    r.set(f"key:{i}", f"value:{i}")

# Bien: 1 round-trip
r.mset({f"key:{i}": f"value:{i}" for i in range(100)})
  1. Usa HSET con múltiples campos. Las operaciones de Hash son más eficientes en memoria para datos estructurados:
r.hset("user:42", mapping={
    "name": "Alice",
    "email": "alice@example.com",
    "role": "admin",
})
  1. Usa SETEX en lugar de SET + EXPIRE. SETEX es atómico y ahorra un round-trip:
# Mal: dos comandos
r.set("key", "value")
r.expire("key", 300)

# Bien: un comando atómico
r.setex("key", 300, "value")
  1. Habilita tcp-keepalive en Redis. Previene que conexiones stale consuman recursos:
# redis.conf
tcp-keepalive 60
  1. Usa CLIENT INFO para debuggear connection leaks. Rastrea cuántas conexiones mantiene cada cliente:
redis-cli CLIENT LIST
# Mostrar idle times y edades de conexión

Preguntas frecuentes

¿Esta solución está lista para producción?

Sí. Los ejemplos de código arriba muestran implementaciones probadas. Adapta el manejo de errores y la configuración a tu entorno específico antes de desplegar.

¿Cuáles son las características de rendimiento?

El rendimiento depende de tu volumen de datos e infraestructura. Las soluciones mostradas priorizan claridad. Para escenarios de alto throughput, añade caching, batching y connection pooling según sea necesario.

¿Cómo depuro problemas con este enfoque?

Empieza con el ejemplo mínimo de arriba. Añade logging en cada paso. Prueba con entradas pequeñas primero, luego escala. Usa el debugger de tu lenguaje para revisar los edge cases.

Cache-Aside en Python con redis-py 
import json
import redis
import functools

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

def cache_aside(prefix: str, ttl: int = 300):
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            key = f"{prefix}:{':'.join(str(a) for a in args)}"
            cached = r.get(key)
            if cached:
                return json.loads(cached)

            result = func(*args, **kwargs)
            r.setex(key, ttl, json.dumps(result, default=str))
            return result
        return wrapper
    return decorator

# Uso
@cache_aside("user_profile", ttl=600)
def get_user_profile(user_id: int) -> dict:
    # Query de base de datos
    return {"id": user_id, "name": "Alice", "email": "alice@example.com"}
Write-Through en Python con Redis 
import json
import redis

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

class WriteThroughCache:
    def __init__(self, redis_client, db_conn):
        self.r = redis_client
        self.db = db_conn

    def set(self, key: str, value: dict, ttl: int = 300):
        # Escribir a base de datos primero
        self.db.execute(
            "INSERT INTO cache_store (key, value) VALUES (%s, %s) "
            "ON CONFLICT (key) DO UPDATE SET value = EXCLUDED.value",
            (key, json.dumps(value))
        )
        self.db.commit()

        # Luego actualizar caché
        self.r.setex(key, ttl, json.dumps(value))

    def get(self, key: str) -> dict | None:
        cached = self.r.get(key)
        if cached:
            return json.loads(cached)

        # Cache miss: leer de base de datos
        row = self.db.execute(
            "SELECT value FROM cache_store WHERE key = %s", (key,)
        ).fetchone()
        if row:
            value = json.loads(row[0])
            self.r.setex(key, 300, json.dumps(value))
            return value
        return None
Redis Streams para Patrón Write-Behind 
import json
import redis
import threading

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

def write_behind_set(key: str, value: dict, ttl: int = 300):
    """Escribir a caché inmediatamente, encolar write a DB via Redis Streams."""
    r.setex(key, ttl, json.dumps(value))

    # Añadir al stream para procesamiento async
    r.xadd("write_behind_stream", {
        "key": key,
        "value": json.dumps(value),
        "operation": "set",
        "timestamp": str(int(time.time()))
    })

# Consumer que procesa el stream
def process_write_behind_stream(db_conn, consumer_name="worker-1"):
    while True:
        # Leer nuevas entradas del stream
        entries = r.xread({"write_behind_stream": "$"}, count=10, block=1000)

        if not entries:
            continue

        for stream, messages in entries:
            for msg_id, fields in messages:
                try:
                    key = fields["key"]
                    value = json.loads(fields["value"])

                    db_conn.execute(
                        "INSERT INTO cache_store (key, value) VALUES (%s, %s) "
                        "ON CONFLICT (key) DO UPDATE SET value = EXCLUDED.value",
                        (key, json.dumps(value))
                    )
                    db_conn.commit()

                    # Acknowledge processing
                    r.xack("write_behind_stream", consumer_name, msg_id)
                except Exception as e:
                    print(f"Error al procesar {msg_id}: {e}")
Cache Warming en Startup 
import redis
import json

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

def warm_cache(queries: list[tuple[str, callable, int]]):
    """Pre-poblar caché en el arranque de aplicación.

    Args:
        queries: Lista de tuplas (cache_key, fetcher_fn, ttl)
    """
    for key, fetcher, ttl in queries:
        try:
            result = fetcher()
            r.setex(key, ttl, json.dumps(result, default=str))
            print(f"Calentado: {key}")
        except Exception as e:
            print(f"Error al calentar {key}: {e}")

# Uso
def fetch_popular_products():
    # Query de base de datos para productos populares
    return [{"id": 1, "name": "Widget"}, {"id": 2, "name": "Gadget"}]

def fetch_config():
    # Query de base de datos para config de app
    return {"theme": "dark", "features": ["search", "filters"]}

warm_cache([
    ("popular_products", fetch_popular_products, 3600),
    ("app_config", fetch_config, 7200),
])
Circuit Breaker para Fallos de Caché 
import time
import redis
from functools import wraps

class CacheCircuitBreaker:
    def __init__(self, failure_threshold=5, recovery_timeout=30):
        self.failure_count = 0
        self.failure_threshold = failure_threshold
        self.recovery_timeout = recovery_timeout
        self.last_failure_time = 0
        self.state = "closed"  # closed, open, half-open

    def can_execute(self):
        if self.state == "open":
            if time.time() - self.last_failure_time > self.recovery_timeout:
                self.state = "half-open"
                return True
            return False
        return True

    def record_success(self):
        self.failure_count = 0
        self.state = "closed"

    def record_failure(self):
        self.failure_count += 1
        self.last_failure_time = time.time()
        if self.failure_count >= self.failure_threshold:
            self.state = "open"

breaker = CacheCircuitBreaker()
r = redis.Redis(host="localhost", port=6379, decode_responses=True)

def cached_with_circuit_breaker(key: str, fetcher, ttl: int = 300):
    if breaker.can_execute():
        try:
            cached = r.get(key)
            if cached:
                breaker.record_success()
                return json.loads(cached)

            result = fetcher()
            r.setex(key, ttl, json.dumps(result, default=str))
            breaker.record_success()
            return result
        except redis.ConnectionError:
            breaker.record_failure()
            # Fall through a fetch directo
    # Circuit abierto o caché falló: fetch directo
    return fetcher()
Invalidation de Caché Basada en Tags 
import redis
import json

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

def set_with_tags(key: str, value: dict, tags: list[str], ttl: int = 300):
    """Setear entrada de caché y asociarla con tags para invalidación grupal."""
    r.setex(key, ttl, json.dumps(value))

    for tag in tags:
        r.sadd(f"tag:{tag}", key)
        r.expire(f"tag:{tag}", ttl + 60)  # Ligeramente más largo que el TTL del key

def invalidate_tag(tag: str):
    """Invalidar todas las entradas de caché asociadas a un tag."""
    keys = r.smembers(f"tag:{tag}")
    if keys:
        r.delete(*keys)
        r.delete(f"tag:{tag}")

# Uso
set_with_tags("user:42", {"name": "Alice"}, tags=["users", "user:42"], ttl=600)
set_with_tags("user:43", {"name": "Bob"}, tags=["users", "user:43"], ttl=600)

# Invalidar todos los caches de usuario cuando cambia el schema
invalidate_tag("users")  # Elimina tanto user:42 como user:43
Caché Multi-Nivel (L1 en memoria + L2 Redis) 
import json
import redis
import time
from functools import lru_cache

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

class MultiTierCache:
    def __init__(self, redis_client, l1_size=128, ttl=300):
        self.r = redis_client
        self.ttl = ttl
        self._l1 = {}  # L1 cache simple basado en dict
        self._l1_max = l1_size
        self._l1_times = {}

    def get(self, key: str):
        # L1: check en memoria
        if key in self._l1:
            if time.time() - self._l1_times[key] < self.ttl:
                return self._l1[key]
            else:
                del self._l1[key]
                del self._l1_times[key]

        # L2: check Redis
        cached = self.r.get(key)
        if cached:
            value = json.loads(cached)
            self._set_l1(key, value)
            return value

        return None

    def set(self, key: str, value: dict, ttl: int = None):
        ttl = ttl or self.ttl
        self._set_l1(key, value)
        self.r.setex(key, ttl, json.dumps(value, default=str))

    def _set_l1(self, key: str, value):
        if len(self._l1) >= self._l1_max:
            # Evictar entrada más antigua
            oldest = min(self._l1_times, key=self._l1_times.get)
            del self._l1[oldest]
            del self._l1_times[oldest]
        self._l1[key] = value
        self._l1_times[key] = time.time()

    def invalidate(self, key: str):
        self._l1.pop(key, None)
        self._l1_times.pop(key, None)
        self.r.delete(key)

# Uso
cache = MultiTierCache(r, l1_size=256, ttl=300)
cache.set("user:42", {"name": "Alice"})
user = cache.get("user:42")  # Hit L1 en segunda llamada
Recipe

Implementar Estrategias de Invalidación de Caché

Cómo mantener la caché consistente con las bases de datos usando TTL, write-through, write-behind y patrones de invalidación event-driven.

Recipe

Configurar connection pooling para bases de datos y HTTP

Configura connection pooling para PostgreSQL, MySQL, Redis y clientes HTTP en Python, JavaScript y Java. Reduce latencia y evita agotamiento de conexiones.

Guide

Guía de Optimización de Performance Web

manual detallado para optimizar el rendimiento de aplicaciones web con mejores Core Web Vitals y experiencia de usuario.

Recipe

Cache de resultados de funciones con Redis y TTL en Python

Construye un decorador de Python que cachea resultados de funciones en Redis con TTL configurable, generacion de claves e invalidacion de cache

Recipe

Conectar a Redis

Cómo conectarse a Redis y realizar operaciones básicas en Python, JavaScript y Java.

Recipe

Optimizacion de Queries e Indexing en PostgreSQL

Analiza y optimiza queries lentas en PostgreSQL usando EXPLAIN, indexing apropiado, partial indexes y reescritura de queries para reducir tiempo de ejecucion