Redis na prática: estratégias de cache para aliviar seu banco principal

1. Por que usar Redis como camada de cache?

Bancos relacionais como MySQL e PostgreSQL são excelentes para persistência e consistência de dados, mas sofrem com gargalos de leitura quando o volume de requisições cresce. Cada consulta ao disco, mesmo com índices otimizados, leva milissegundos. Em aplicações com milhares de acessos simultâneos, esse tempo se acumula e degrada a experiência do usuário.

O Redis opera inteiramente em memória RAM, com latências típicas de microssegundos — até 100x mais rápido que um banco tradicional. Ao armazenar dados frequentemente acessados em cache, você reduz drasticamente a carga no banco principal, economiza CPU e I/O de disco, e melhora o tempo de resposta da aplicação.

2. Padrões fundamentais de cache com Redis

Cache-Aside (leitura sob demanda)

O padrão mais comum. A aplicação verifica o cache primeiro; se não encontrar (cache miss), busca no banco, armazena no Redis e retorna.

function buscarUsuario(id):
    chave = "usuario:" + id
    dados = redis.get(chave)
    if dados is None:
        dados = banco.buscarUsuario(id)
        redis.setex(chave, 3600, dados)  # TTL de 1 hora
    return dados

Write-Through (escrita síncrona)

Ao atualizar um dado, a aplicação escreve simultaneamente no cache e no banco. Garante que o cache esteja sempre atualizado, mas aumenta a latência da escrita.

function atualizarUsuario(id, novosDados):
    banco.atualizarUsuario(id, novosDados)
    chave = "usuario:" + id
    redis.set(chave, novosDados)

Write-Behind (escrita assíncrona)

A aplicação escreve apenas no cache, e um processo em background persiste os dados no banco posteriormente. Oferece baixa latência de escrita, mas requer cuidado com perda de dados em falhas.

3. Estratégias de expiração e invalidação

TTL (Time-To-Live)

Definir um tempo de vida para cada chave evita dados obsoletos e libera memória automaticamente.

redis.setex("relatorio:vendas:diario", 300, dadosRelatorio)

Invalidação manual

Quando o dado original é alterado, remova ou atualize a chave correspondente no cache.

function deletarProduto(id):
    banco.deletarProduto(id)
    redis.delete("produto:" + id)

Evitando Cache Stampede

Quando um cache popular expira, milhares de requisições podem bater no banco simultaneamente. Soluções:
- Usar SET NX com lock distribuído para que apenas um processo recarregue o cache
- Adicionar jitter (variação aleatória) no TTL para evitar expirações em massa

4. Estruturas de dados do Redis para cenários reais

Strings e Hashes para objetos

Para perfis de usuário, use hashes — mais eficientes que strings grandes.

redis.hset("usuario:456", "nome", "Maria")
redis.hset("usuario:456", "email", "maria@email.com")
dados = redis.hgetall("usuario:456")

Sorted Sets para rankings

Ideal para leaderboards e contagens em tempo real.

redis.zadd("ranking:jogadores", {"joao": 1500, "maria": 2300})
top10 = redis.zrevrange("ranking:jogadores", 0, 9, withscores=True)

Lists e Streams para filas

Útil para cache de páginas ou processamento assíncrono.

redis.rpush("fila:emails", json.dumps({"para": "user@email.com", "assunto": "Bem-vindo"}))
tarefa = redis.lpop("fila:emails")

5. Cache de queries e consultas complexas

Resultados de queries SQL agregadas ou relatórios podem ser cacheados com chaves compostas.

function buscarPedidosRecentes(usuarioId, dias):
    chave = f"usuario:{usuarioId}:pedidos:ultimos{dias}dias"
    dados = redis.get(chave)
    if dados is None:
        dados = banco.executar(f"SELECT * FROM pedidos WHERE usuario_id = ? AND data > NOW() - INTERVAL ? DAY", [usuarioId, dias])
        redis.setex(chave, 1800, dados)  # 30 minutos
    return dados

6. Cache distribuído: lidando com clusters e replicação

  • Redis Sentinel: Oferece alta disponibilidade com failover automático. Ideal para ambientes que toleram pequenas pausas.
  • Redis Cluster: Distribui dados automaticamente entre múltiplos nós, permitindo escalabilidade horizontal e maior capacidade de memória.
  • Consistência eventual: Em cache distribuído, aceite que dados possam estar levemente desatualizados entre nós.
  • Client-side caching: Redis 6+ suporta cache local no cliente, reduzindo ainda mais a latência para dados muito acessados.

7. Monitoramento e boas práticas de operação

Métricas essenciais

  • Hit rate: Percentual de requisições atendidas pelo cache (ideal acima de 90%)
  • Miss rate: Requisições que foram ao banco
  • Latência média: Tempo de resposta do Redis
  • Memória usada: Evite ultrapassar 75% da RAM disponível

Evitando hot keys e big keys

  • Hot keys: Chaves muito acessadas podem sobrecarregar um único nó. Distribua a carga usando replicação de leitura ou cache local.
  • Big keys: Chaves com grandes volumes de dados (ex.: listas com milhões de itens) degradam a performance. Use estruturas mais granulares.

Use SCAN em vez de KEYS

O comando KEYS bloqueia o Redis e pode travar o servidor em produção. Prefira SCAN para iterar sobre chaves.

cursor = 0
while True:
    cursor, chaves = redis.scan(cursor, match="usuario:*", count=100)
    for chave in chaves:
        print(chave)
    if cursor == 0:
        break

8. Exemplo prático: implementando cache em uma aplicação real

Fluxo completo de cache-aside com fallback para o banco

import redis
import mysql.connector
import json

# Conexões
cache = redis.Redis(host='localhost', port=6379, decode_responses=True)
banco = mysql.connector.connect(host='localhost', user='app', password='secret', database='ecommerce')

def buscar_produto_com_cache(produto_id):
    chave = f"produto:{produto_id}"

    # 1. Tentar cache
    produto_json = cache.get(chave)
    if produto_json:
        print("Cache HIT")
        return json.loads(produto_json)

    # 2. Cache MISS - buscar no banco
    print("Cache MISS - consultando banco")
    cursor = banco.cursor(dictionary=True)
    cursor.execute("SELECT id, nome, preco, estoque FROM produtos WHERE id = %s", (produto_id,))
    produto = cursor.fetchone()
    cursor.close()

    if not produto:
        return None

    # 3. Armazenar no cache com TTL de 10 minutos
    cache.setex(chave, 600, json.dumps(produto, default=str))

    return produto

# Teste de performance
import time

# Primeira chamada (sem cache)
inicio = time.time()
produto = buscar_produto_com_cache(1)
print(f"Tempo sem cache: {time.time() - inicio:.4f} segundos")

# Segunda chamada (com cache)
inicio = time.time()
produto = buscar_produto_com_cache(1)
print(f"Tempo com cache: {time.time() - inicio:.4f} segundos")

Resultado esperado

Cache MISS - consultando banco
Tempo sem cache: 0.0234 segundos
Cache HIT
Tempo com cache: 0.0008 segundos

A diferença é gritante: 29x mais rápido com cache. Em produção, com milhares de requisições por segundo, essa economia se traduz em servidores mais leves, menor custo de infraestrutura e usuários mais satisfeitos.

Referências