Introdução

Caching é uma técnica essencial para melhorar a performance e escalabilidade de aplicações. Em sistemas modernos, é comum que bancos de dados se tornem gargalos, principalmente quando recebem muitas requisições repetidas para os mesmos dados. O cache entra em cena para aliviar essa pressão, guardando os dados mais acessados em uma camada intermediária de acesso mais rápido — normalmente, na memória RAM. Ao receber uma requisição, a aplicação verifica primeiro no cache; se o dado estiver lá (cache hit), retorna imediatamente. Caso contrário (cache miss), busca no banco de dados, armazena no cache e então retorna ao cliente.

O Redis é uma das ferramentas mais populares para esse tipo de tarefa. Neste tutorial, vamos entender os conceitos fundamentais de caching, aprender como o Redis funciona e, finalmente, implementá-lo na prática em um projeto Python real com acesso a um banco de dados PostgreSQL via SQLAlchemy.

Redis

O Redis é uma ferramenta de código aberto, extremamente rápida, usada para armazenar dados em memória. Isso o torna ideal para aplicações que exigem acesso a informações com baixa latência, como sistemas de recomendação em tempo real ou APIs de alta demanda. o Redis suporta o armazenamento de estruturas como strings, listas, hashes e conjuntos, e oferece recursos avançados como expiração automática, replicação e clusterização.

Implementando Caching com Redis

Vamos usar Python e Redis, mas a abordagem vale para diversas linguagens (veja https://redis.io/learn/develop)

1. Implementação Simples

Antes de mais nada, precisamos de uma classe que consulte o banco e armazene no cache:

from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.orm import declarative_base, sessionmaker
import redis

# Configuração do banco
DATABASE_URL = "postgresql://usuario:senha@localhost:5432/meubanco"
engine = create_engine(DATABASE_URL)
SessionLocal = sessionmaker(bind=engine)
Base = declarative_base()

# User Model
class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True)
    name = Column(String)

Base.metadata.create_all(engine)

# Cache simples no Redis
class RedisCache:
    def __init__(self, url="redis://localhost:6379/0"):
        self.redis = redis.Redis.from_url(url, decode_responses=True)
    def get(self, key):
        return self.redis.get(key)
    def set(self, key, value):
        self.redis.set(key, value)

class UserService:
    def __init__(self, db_session, cache):
        self.db = db_session
        self.cache = cache
    def get_user(self, user_id):
        key = f"user:{user_id}"
        cached = self.cache.get(key)
        if cached:
            print("Cache hit")  # Dado já no cache
            return json.loads(cached)
        print("Cache miss: buscando no banco")
        user = self.db.query(User).filter(User.id==user_id).first()
        if user:
            data = {"id": user.id, "name": user.name}
            self.cache.set(key, data)
            return data
        return None

Note que, ao armazenar dados no cache, utilizamos uma convenção para as chaves no formato "user:{user_id}". Esse padrão é uma prática recomendada no uso do Redis, pois ajuda a organizar melhor os dados e evita colisões entre diferentes tipos de informações. Ao prefixar a chave com o tipo da entidade (user), deixamos claro o que está sendo armazenado e facilitamos operações futuras, como a invalidação de cache (mais sobre isso nas próximas sessões).

Essa implementação simples funciona, mas logo surge uma pergunta: e se o cache crescer indefinidamente?

2. TTL: Tempo de Vida das Chaves

Para limitar o uso de memória, podemos atribuir um Time To Live (TTL) a cada chave. Ao expirar, o Redis descarta automaticamente a entrada.

# Ajuste na classe RedisCache
class RedisCache:
    # ... init como antes ...
    def set(self, key, value, ttl=None):
        if ttl:
            # TTL em segundos: aqui, 120s (2 minutos)
            self.redis.setex(key, ttl, value)
        else:
            self.redis.set(key, value)

# No UserService:
self.cache.set(key, data, ttl=120)

Por que TTL? Dentro de um período típico de 2 minutos, dados populares são recarregados várias vezes, mantendo o cache sem nunca crescer acima do necessário.

3. Políticas de Despejo (Eviction)

Mesmo com TTL, pode haver cenários em que o cache atinja limites de memória antes das chaves expirarem. O Redis permite configuração via `maxmemory` e `maxmemory-policy`:

# Exemplo de configuração dinâmica
conn = redis.Redis()
conn.config_set("maxmemory", "100mb")
conn.config_set("maxmemory-policy", "allkeys-lfu")

Opções de política:'

allkeys-lru: remove a chave menos usada recentemente
allkeys-lfu: remove a chave menos acessada no total
volatile-ttl: remove a chave com menor TTL restante
noeviction: retorna erro ao exceder memória

Nesse caso, escolhemos allkeys-lfu. Porém, a melhor política varia muito de caso a caso.

4. Invalidação de Cache

A invalidação de cache é o processo pelo qual dados armazenados temporariamente em um cache são removidos ou atualizados para garantir que reflitam as informações mais recentes da fonte original, como um banco de dados. Esse mecanismo é essencial para evitar que usuários recebam dados desatualizados, o que pode comprometer a integridade e a confiabilidade do sistema. Existem diversas estratégias para realizar a invalidação de cache, incluindo a remoção manual de entradas específicas, a atualização automática baseada em eventos e a definição de um tempo de vida (TTL) para os dados em cache. A escolha da estratégia adequada depende das necessidades específicas da aplicação, equilibrando a necessidade de dados atualizados com o desempenho do sistema. Implementar uma política eficaz de invalidação de cache é crucial para manter a consistência dos dados e proporcionar uma experiência de usuário confiável

4.1. Solução com KEYS

Uma ideia simples é usar `KEYS` para listar e remover chaves com um dado prefixo:

def invalidate_with_keys(conn, user_id):
    pattern = f"user:{user_id}*"
    for key in conn.keys(pattern):
        conn.delete(key)

Por que não usar? O comando KEYS varre todo o dataset, bloqueando o Redis durante a operação, o que causa lentidão para outros clientes. Isso é um problema por que o Redis é single-threaded. Além disso, é comum que vários serviços compartilhem a mesma instância do redis. Então, todos os serviços parariam sempre que um deles tivesse que fazer uma invalidação de cache.

4.2. Solução Eficiente com SCAN_ITER

Para evitar bloqueios, usamos SCAN_ITER.

def invalidate_with_scan_iter(conn, user_id):
    pattern = f"user:{user_id}*"
    for key in conn.scan_iter(match=pattern):
        conn.delete(key)

A varredura incremental evita bloqueios prolongados, mantendo a alta disponibilidade do cache. O SCAN_ITER divide o banco inteiro em buckets, e analisa eles um a um, permitindo que outras tarefas executem nesses intervalos.

4.3 Mais controle com SCAN

Similar ao SCAN_ITER, temos o SCAN:

CHUNK_SIZE = 5000

def invalidate_with_scan(conn, user_id):
    pattern = f"user:{user_id}*"
    cursor, keys = conn.scan(cursor=0, match=pattern, count=CHUNK_SIZE)
    if keys:
        conn.delete(*keys)
    
    while cursor != 0:
        cursor, keys = conn.scan(cursor=cursor, match=pattern, count=CHUNK_SIZE)
        if keys:
            conn.delete(*keys)

O comando SCAN funciona de forma semelhante ao SCAN_ITER, mas com mais controle sobre a iteração. Ele permite definir quantos elementos devem ser retornados por vez (com o parâmetro count) — o que pode ser útil para otimizar a performance de acordo com sua aplicação.

Além disso, o SCAN utiliza um cursor para controlar a posição na iteração. A primeira chamada deve iniciar com o cursor 0, e o processo de varredura termina quando o Redis retorna novamente 0 como cursor, indicando que todas as partições de chaves foram examinadas.