Cache com Redis: padrões que funcionam

O problema real#

Consultar perfil de usuário no banco demora 80 ms (network latência + query). Sua API responde 100 requisições/segundo × 80 ms = 8 segundos de latência total. Cliente espera, browser carrega lento, usuário sai.

Aí coloca Redis. Mesma query agora demora 5 ms (rede local). Latência cai de 8s para 5s. Você acha que ganhou, mas o cache tá errado. Usuário muda nome, você não atualiza o cache. Ele vê nome velho por horas. Coloca TTL de 1 hora, achando que resolve. 100 mil usuários × 1 hora = você gerou 100 mil queries quando TTL expirou (thundering herd / cache stampede).

Realidade: cache é fácil de entender, difícil de fazer direito.

Cache-aside: o padrão mais comum#

Você lê de Redis. Se não tem (miss), lê do banco e salva em Redis:

import redis
import json

cache = redis.Redis(host='localhost', port=6379)

def get_user_profile(user_id):
    # Passo 1: tentar cache
    key = f"user:{user_id}:profile"
    cached = cache.get(key)
    
    if cached:
        print(f"Cache hit")
        return json.loads(cached)
    
    # Passo 2: cache miss, consulta banco
    print(f"Cache miss, querying DB")
    user = db.query(f"SELECT * FROM users WHERE id = {user_id}")
    
    # Passo 3: salva em Redis (com TTL)
    cache.setex(key, 3600, json.dumps(user))  # 3600 segundos = 1 hora
    
    return user

# Teste
user = get_user_profile(123)  # Cache miss (0 ms DB + 1 ms Redis) = 81 ms
user = get_user_profile(123)  # Cache hit (1 ms Redis) = 1 ms

Vantagem: simples, não precisa invalidar — TTL expira sozinho. Desvantagem: miss é caro (vai pro banco). Se 1 milhão de usuários acessam o mesmo perfil ao mesmo tempo e TTL expirou, 1 milhão de queries = cache stampede.

Write-through: garantir consistência#

Quando salva, escreve em cache E no banco:

def update_user_profile(user_id, new_data):
    # Passo 1: atualiza banco
    db.query(f"UPDATE users SET name = '{new_data['name']}' WHERE id = {user_id}")
    
    # Passo 2: atualiza cache
    key = f"user:{user_id}:profile"
    cache.setex(key, 3600, json.dumps(new_data))
    
    return new_data

# Teste
update_user_profile(123, {'name': 'João Silva'})  # Atualiza DB e cache
user = get_user_profile(123)  # Cache hit, novo nome

Vantagem: cache sempre consistente com banco. Desvantagem: lento — duas operações (DB + Redis) são seriais. Se uma falha, está inconsistente.

Write-behind: rapido mas arriscado#

Escreve em cache primeiro, depois no banco de forma assíncrona:

import asyncio

def update_user_profile_async(user_id, new_data):
    # Passo 1: salva em cache (rápido)
    key = f"user:{user_id}:profile"
    cache.setex(key, 3600, json.dumps(new_data))
    
    # Passo 2: agenda update do banco (assíncrono)
    asyncio.create_task(update_db_async(user_id, new_data))
    
    return {"status": "queued"}

async def update_db_async(user_id, new_data):
    # Simula delay de rede
    await asyncio.sleep(0.1)
    db.query(f"UPDATE users SET name = '{new_data['name']}' WHERE id = {user_id}")

# Teste
update_user_profile_async(123, {'name': 'Maria'})  # Retorna imediato (5 ms)

Vantagem: velocidade — retorna imediatamente. Desvantagem: risco — se aplicação crasha antes de sincronizar, perda de dados. Use com fila durável.

TTL e a escolha do tempo#

Quanto tempo cachear?

# ❌ TTL muito longo
cache.setex(key, 86400, data)  # 24 horas — usuário vê dados de ontem

# ✓ TTL apropriado
cache.setex(key, 300, data)  # 5 minutos — bom balanço pra maioria

# ❌ TTL muito curto
cache.setex(key, 10, data)  # 10 segundos — expira rápido, cachê é inútil

Regra empírica:

• Dados mutáveis frequentes (preço, estoque): 30 a 60 segundos
• Dados estáveis (perfil, foto): 1 a 24 horas
• Dados imutáveis (config, referência): sem TTL (PERSIST) ou dias

Invalidação: o problema difícil#

Você não quer esperar TTL expirar. Quando atualiza, deleta cache:

def update_user_email(user_id, new_email):
    # Atualiza banco
    db.query(f"UPDATE users SET email = '{new_email}' WHERE id = {user_id}")
    
    # ❌ PROBLEMA: qual chave deletar?
    # cache.delete(f"user:{user_id}:profile")  # deleta perfil
    # cache.delete(f"user:{user_id}:stats")    # deleta stats
    # cache.delete(f"user:{user_id}:*")        # precisa SCAN
    
    # ✓ SOLUÇÃO 1: usar versão de cache (cache versioning)
    cache.incr(f"user:{user_id}:version")  # incrementa versão
    
    # Query agora precisa ler versão
    def get_user_profile(user_id):
        version = cache.get(f"user:{user_id}:version") or "1"
        key = f"user:{user_id}:profile:v{version}"
        # ...

Estratégia por tag (em código da aplicação):

# Salva com metadados
user_data = {
    'id': 123,
    'name': 'João',
    'email': 'joao@example.com'
}

cache.hset(f"user:123", mapping={
    'data': json.dumps(user_data),
    'version': '1',
    'tags': 'user-active,user-premium'  # tags separadas por vírgula
})

# Invalidação por tag
def invalidate_by_tag(tag):
    # Buscar todas as chaves com essa tag
    keys = cache.scan_iter(match=f"*")
    for key in keys:
        tags = cache.hget(key, 'tags')
        if tags and tag.encode() in tags:
            cache.delete(key)

Cache stampede: as 3 defesas#

Problema: TTL expira, 1.000 requisições simultâneas, todas vão pro banco.

Defesa 1: Lock (Mutex)#

Apenas 1 thread recomputa, outros esperam:

import threading

def get_user_with_lock(user_id):
    key = f"user:{user_id}:profile"
    
    # Passo 1: tentar cache
    cached = cache.get(key)
    if cached:
        return json.loads(cached)
    
    # Passo 2: cache miss, tomar lock distribuído
    lock_key = f"user:{user_id}:lock"
    lock = cache.lock(lock_key, timeout=2)  # lock por 2 segundos
    
    if lock.acquire(blocking=False):
        try:
            # Passo 3: verificar novamente (outro thread pode ter preenchido)
            cached = cache.get(key)
            if cached:
                return json.loads(cached)
            
            # Passo 4: computar (apenas este thread)
            user = db.query(f"SELECT * FROM users WHERE id = {user_id}")
            cache.setex(key, 3600, json.dumps(user))
            return user
        finally:
            lock.release()
    else:
        # Outro thread pegou lock, espera que ele preencha cache
        time.sleep(0.1)
        cached = cache.get(key)
        return json.loads(cached)

Defesa 2: Jitter no TTL#

Variar TTL evita expiração em massa:

import random

# ❌ Todos expiram no mesmo segundo
cache.setex(key, 3600, data)

# ✓ TTL com jitter (aleatório)
ttl = 3600 + random.randint(-300, 300)  # 3.600 ± 5 minutos
cache.setex(key, ttl, data)

# Resultado: expiram distribuídas ao longo de 10 minutos

Defesa 3: Recomputação antecipada#

Antes de expirar, recomputa em background:

def get_user_smart(user_id):
    key = f"user:{user_id}:profile"
    ttl_key = f"user:{user_id}:ttl"
    
    cached = cache.get(key)
    if cached:
        # Verificar se tá perto de expirar
        ttl_remaining = cache.ttl(key)
        if ttl_remaining and ttl_remaining < 300:  # < 5 minutos
            # Recomputa em background (não bloqueia)
            asyncio.create_task(recompute_cache(user_id))
        
        return json.loads(cached)
    
    # Sem cache, computa normalmente
    user = db.query(f"SELECT * FROM users WHERE id = {user_id}")
    cache.setex(key, 3600, json.dumps(user))
    return user

async def recompute_cache(user_id):
    # Thread separada recomputa
    user = db.query(f"SELECT * FROM users WHERE id = {user_id}")
    cache.setex(f"user:{user_id}:profile", 3600, json.dumps(user))

O que NÃO cachear#

# ❌ Passwords e tokens (nunca!)
cache.set(f"token:{user_id}", password)

# ❌ Dados com regras de conformidade (LGPD, GDPR)
# Pode violar direito ao esquecimento

# ❌ Valores computados rapidinho < 5 ms
# Cache custa 1 ms, não vale

# ✓ Queries que demoram > 50 ms
# ✓ Dados imutáveis ou raramente mutáveis
# ✓ Computações custosas (agregações, joins)

Medindo hit rate e custo#

class MonitoredCache:
    def __init__(self, redis_conn):
        self.cache = redis_conn
        self.hits = 0
        self.misses = 0
    
    def get(self, key):
        value = self.cache.get(key)
        if value:
            self.hits += 1
        else:
            self.misses += 1
        return value
    
    def report(self):
        total = self.hits + self.misses
        hit_rate = (self.hits / total * 100) if total > 0 else 0
        print(f"Hit Rate: {hit_rate:.1f}% ({self.hits} hits, {self.misses} misses)")
        
        # Custo: cache miss = DB query (80 ms)
        # Cache hit = 1 ms
        time_saved = self.hits * (80 - 1) + self.misses * 80
        print(f"Tempo economizado: {time_saved / 1000:.1f}s")

cache = MonitoredCache(redis.Redis())
# Usar cache normalmente
cache.report()
# Hit Rate: 92.3% (923 hits, 77 misses)
# Tempo economizado: 72.5s

Quando NÃO usar Redis#

Volume pequeno: se DB queries demoram 5 ms, Redis (1 ms) economiza pouco. Gasto com operação é maior que ganho.

Dados muito mutáveis: conta bancária muda a cada segundo. Cache inconsistente = bugs graves. Não vale a complexidade.

Infraestrutura sem DevOps: Redis precisa monitorar, backup, failover. Managed Redis (AWS ElastiCache) é mais simples, mas caro.

Armadilhas comuns#

1. Evicção de dados por falta de memória:

WARNING: Memory used by Redis is above maxmemory policy

Configure limite e política (LRU vs LFU vs random):

# redis.conf
maxmemory 2gb
maxmemory-policy allkeys-lru  # remove chaves menos usadas

2. Sem replicação/backup: Redis tá em memória. Falha, perde tudo. Sempre use replicação:

# docker-compose.yml
redis-primary:
  image: redis:7
  command: redis-server --port 6379

redis-replica:
  image: redis:7
  command: redis-server --port 6380 --slaveof redis-primary 6379

Próximos passos#

Combine cache com observabilidade: Observabilidade em aplicações com LLM — monitora hit rate em tempo real.

Deploy Redis em produção com infra como código: Terraform na prática: sua primeira infra versionada.

Teste: cria Redis local (Docker), escreve app que faz GET/SET com monitoramento, mede hit rate com diferentes TTLs. Você vai ver empiricamente qual TTL melhor serve sua carga.