Cache — Redis para Acelerar Consultas Quentes

Q: Como um filtro de Bloom é implementado?

Usando o módulo redisbloom ou a biblioteca Python pybloom_live . Todos os IDs de produtos existentes são adicionados ao filtro de Bloom, e uma verificação é realizada antes de cada consulta. A taxa de falsos positivos é de aproximadamente 1%, o que é aceitável.

Q: Como monitoro a taxa de hit do cache?

O comando INFO stats do Redis retorna keyspace_hits e keyspace_misses ; a taxa de hit é calculada como hits / (hits + misses). Integre isso em um dashboard de monitoramento Prometheus + Grafana.

Um cache é como uma geladeira — você mantém os ingredientes que mais usa (dados quentes) na geladeira para que estejam sempre à mão, e não precisa ir ao supermercado (banco de dados) toda vez. Mas ingredientes expiram (TTL), e quando repõe o estoque, precisa atualizá-los (invalidação de cache).

1. O Que Você Vai Aprender

Gerenciamento de Conexão Redis: redis-py Integrando um Cliente Assíncrono com FastAPI
Estratégias de Cache: Padrão Cache-Aside, Expiração TTL, Design de Nomenclatura de Chave de Cache
Invalidação de Cache: Invalidação proativa durante operações de escrita; consistência de cache durante atualizações em lote de preços
Proteção contra Penetração/Rompimento/Avalanche de Cache: Filtros de Bloom, Locks de Exclusão Mútua, TTL Aleatório
Cenário Alice: Cache de Preços de Produtos Populares — latência P99 reduzida de 200 ms para 5 ms

2. A História Real da Alice

(1) Problema: Consultas Populares Sobrecarregam o Banco de Dados

O PriceTracker tem 100 produtos populares que representam 80% do volume de consultas, com 8.000 consultas por segundo. Cada consulta passa pelo PostgreSQL, fazendo a CPU do banco de dados subir para 90% e resultando em uma latência P99 de 200 ms. Charlie diz que adicionar uma instância de banco de dados custaria USD 500 por mês, mas 80% dos resultados de consulta não mudam em 5 minutos.

(2) Soluções para Cache com Redis

Armazene dados de preços de produtos populares no Redis. Para consultas, verifique o Redis primeiro (5 ms); se não houver hit, consulte o PostgreSQL (50 ms) e escreva o resultado de volta no Redis com um TTL de 5 minutos.

PYTHON

async def get_product_cached(product_id: int, db, redis):
    cached = await redis.get(f"product:{product_id}")
    if cached:
        return json.loads(cached)
    product = await db.execute(select(Product).where(Product.id == product_id))
    data = product.scalar_one_or_none()
    await redis.setex(f"product:{product_id}", 300, json.dumps(data))
    return data

(3) Resultado

Consultas populares atingem o cache Redis 80% das vezes; latência P99 caiu de 200 ms para 5 ms; QPS do banco de dados caiu de 8.000 para 1.600; instâncias de banco de dados do Charlie foram reduzidas de 4 para 1; economizando USD 1.500 por mês.

3. Gerenciamento de Conexão Redis

(1) Integração do Cliente Assíncrono

(1) ▶Exemplo: Dependência de Conexão Redis

PYTHON

import redis.asyncio as aioredis
from fastapi import FastAPI, Depends

REDIS_URL = "redis://localhost:6379/0"

# Pool de conexão Redis em nível de aplicação
redis_pool = aioredis.ConnectionPool.from_url(REDIS_URL, max_connections=20)

async def get_redis() -> aioredis.Redis:
    """Cliente Redis com escopo de requisição a partir do pool de conexão"""
    client = aioredis.Redis(connection_pool=redis_pool)
    try:
        yield client
    finally:
        await client.aclose()

# Ciclo de vida startup/shutdown
app = FastAPI()

@app.on_event("startup")
async def startup():
    global redis_pool
    redis_pool = aioredis.ConnectionPool.from_url(REDIS_URL, max_connections=20)

@app.on_event("shutdown")
async def shutdown():
    await redis_pool.disconnect()

Saída:

TEXT

# Função definida com sucesso

(2) Selecionando Estruturas de Dados do Redis

Estrutura	Comando	Caso de Uso	Caso de Uso PriceTracker
String	SET/GET	Cache de Valor Único	Cache de Preço de Produto
Hash	HSET/HGET	Cache de Objeto	Cache de Detalhes de Produto
List	LPUSH/LRANGE	Série Temporal	Log de Mudanças de Preço
Set	SADD/SMEMBERS	Conjunto Sem Duplicatas	IDs de Preços Enviados
ZSet	ZADD/ZRANGE	Ranking	Produtos Mais Vendidos

4. Modo Cache-Aside

(1) Processo de Leitura/Escrita

flowchart TD
    Request[Requisição de Leitura] --> CheckRedis{Cache Hit?}
    CheckRedis -->|Sim| Return[Retornar Dados em Cache]
    CheckRedis -->|Não| QueryDB[Consultar Banco de Dados]
    QueryDB --> WriteCache[Escrever no Redis com TTL]
    WriteCache --> Return2[Retornar Dados]
    
    WriteRequest[Requisição de Escrita] --> UpdateDB[Atualizar Banco de Dados]
    UpdateDB --> Invalidate[Invalidar Cache]
    Invalidate --> Return3[Retornar Sucesso]

(1) ▶Exemplo: Consulta de Produto Cache-Aside

PYTHON

import json
from fastapi import FastAPI, Depends
from redis.asyncio import Redis

app = FastAPI()

@app.get("/api/v1/products/{product_id}")
async def get_product(
    product_id: int,
    db: AsyncSession = Depends(get_db),
    redis: Redis = Depends(get_redis),
):
    # Passo 1: Verificar cache Redis
    cache_key = f"product:{product_id}"
    cached = await redis.get(cache_key)
    if cached:
        return json.loads(cached)
    
    # Passo 2: Cache miss - consultar banco de dados
    repo = ProductRepository(db)
    product = await repo.get_by_id(product_id)
    if not product:
        raise HTTPException(status_code=404, detail="Product not found")
    
    # Passo 3: Escrever no cache com TTL
    product_data = ProductResponse.model_validate(product).model_dump()
    await redis.setex(cache_key, 300, json.dumps(product_data))  # TTL de 5 min
    
    return product_data

Saída:

TEXT

# Função definida com sucesso

(2) ▶Exemplo: Invalidação proativa durante operações de escrita

PYTHON

@app.put("/api/v1/products/{product_id}")
async def update_product(
    product_id: int,
    update: ProductUpdate,
    db: AsyncSession = Depends(get_db),
    redis: Redis = Depends(get_redis),
    user=Depends(get_current_user),
):
    repo = ProductRepository(db)
    product = await repo.update(product_id, update.model_dump(exclude_unset=True))
    
    # Invalidar cache após escrita
    await redis.delete(f"product:{product_id}")
    
    return ProductResponse.model_validate(product).model_dump()

Saída:

TEXT

# Função definida com sucesso

(2) Convenções de Nomenclatura de Chave de Cache

Modo	Nome do Elemento	TTL	Descrição
Cache em Memória	`product:{id}`	5 min	Detalhes de Produto Único
Cache de Lista	`products:cat:{category}:p:{page}`	2 min	Lista Paginada
Contador	`count:products:cat:{category}`	1 min	Contagem de Produtos
Top Picks	`ranking:products:hot`	10 min	Rankings ZSet
Contagem de Limitação de Taxa	`ratelimit:{ip}`	60 s	Limitação de Taxa da API

5. Proteção contra Penetração/Rompimento/Avalanche de Cache

(1) Três Principais Problemas de Cache

Problema	Causa	Estratégia de Mitigação
Penetração	Verificação de Dados Inexistentes, Bypassando o Cache para Acessar o DB Diretamente	Filtros de Bloom, Cache de Valores Nulos
Sobrecarga	Surta massiva de requisições atinge o banco de dados no momento em que uma chave quente expira	Locks de exclusão mútua, chaves sem expiração + atualizações assíncronas
Avalanche	Expiração em Massa de Chaves	Offset Aleatório de TTL

(1) ▶Exemplo: Cache de Valores Nulos para Prevenir Penetração

PYTHON

@app.get("/api/v1/products/{product_id}")
async def get_product_with_null_cache(
    product_id: int,
    db: AsyncSession = Depends(get_db),
    redis: Redis = Depends(get_redis),
):
    cache_key = f"product:{product_id}"
    cached = await redis.get(cache_key)
    
    if cached:
        data = json.loads(cached)
        if data.get("_null"):
            raise HTTPException(status_code=404, detail="Product not found")
        return data
    
    # Consultar DB
    repo = ProductRepository(db)
    product = await repo.get_by_id(product_id)
    
    if not product:
        # Cachear valor nulo com TTL curto para prevenir penetração de cache
        await redis.setex(cache_key, 60, json.dumps({"_null": True}))
        raise HTTPException(status_code=404, detail="Product not found")
    
    product_data = ProductResponse.model_validate(product).model_dump()
    await redis.setex(cache_key, 300, json.dumps(product_data))
    return product_data

Saída:

TEXT

# Função definida com sucesso

(2) ▶Exemplo: Prevenção de Rompimento com Locks de Exclusão Mútua

PYTHON

import asyncio

async def get_product_with_lock(
    product_id: int,
    db: AsyncSession,
    redis: Redis,
):
    cache_key = f"product:{product_id}"
    lock_key = f"lock:product:{product_id}"
    
    cached = await redis.get(cache_key)
    if cached:
        return json.loads(cached)
    
    # Tentar adquirir lock (apenas uma requisição reconstrói o cache)
    lock_acquired = await redis.set(lock_key, "1", nx=True, ex=10)
    
    if lock_acquired:
        try:
            # Esta requisição reconstrói o cache
            repo = ProductRepository(db)
            product = await repo.get_by_id(product_id)
            if product:
                data = ProductResponse.model_validate(product).model_dump()
                await redis.setex(cache_key, 300, json.dumps(data))
                return data
        finally:
            await redis.delete(lock_key)
    else:
        # Outras requisições aguardam brevemente e tentam o cache novamente
        await asyncio.sleep(0.1)
        cached = await redis.get(cache_key)
        if cached:
            return json.loads(cached)
    
    raise HTTPException(status_code=404, detail="Product not found")

Saída:

TEXT

# Função definida com sucesso

(3) ▶Exemplo: TTL Aleatório para Prevenir Avalanches

PYTHON

import random

def get_cache_ttl(base_ttl: int = 300, jitter: int = 60) -> int:
    """Adicionar jitter aleatório ao TTL para prevenir avalanche de cache"""
    return base_ttl + random.randint(-jitter, jitter)

# Uso
await redis.setex(cache_key, get_cache_ttl(300, 60), json.dumps(data))
# TTL: 240-360 segundos (5 min ± 1 min de jitter)

Saída:

TEXT

# Função definida com sucesso

(2) Comparação de Políticas de Coerência de Cache

Estratégia	Operação de Escrita	Vantagens	Desvantagens
Cache-Aside	Atualizar o DB primeiro, depois limpar o cache	Simples e confiável	Breve janela de inconsistência
Write-Through	Escreve simultaneamente no cache e no DB	Alta consistência	Alta latência de escrita
Write-Behind	Escreve no cache primeiro, depois assincronamente no banco de dados	Boa performance de escrita	Potencial perda de dados

❓Perguntas Frequentes

P O cache deve armazenar dados em JSON ou Pickle?

R Recomendamos JSON. É legível por humanos, multi-linguagem e seguro (Pickle tem vulnerabilidades de desserialização). orjson é 3 a 10 vezes mais rápido que json.

P Qual é a configuração de TTL apropriada?

R Depende da frequência com que os dados mudam. Dados de preço do PriceTracker são atualizados a cada 5 minutos (breves inconsistências são permitidas), detalhes de produto a cada 30 minutos, e rankings a cada 10 minutos. Um offset aleatório é adicionado para prevenir avalanche.

P Como devo configurar o tamanho do pool de conexão Redis?

R Combine com o número de Workers FastAPI. Cada Worker requer 5-10 conexões Redis. 4 Workers x 10 = 40 conexões; defina max_connections para 50 para permitir alguma folga.

P Como um filtro de Bloom é implementado?

R Usando o módulo redisbloom ou a biblioteca Python pybloom_live. Todos os IDs de produtos existentes são adicionados ao filtro de Bloom, e uma verificação é realizada antes de cada consulta. A taxa de falsos positivos é de aproximadamente 1%, o que é aceitável.

P O que devo fazer se houver inconsistência entre o cache e o banco de dados?

R No modo Cache-Aside, inconsistências temporárias são normais. Estratégia: Atualize o DB primeiro, depois delete o cache (em vez de atualizar o cache), e use uma dupla exclusão atrasada (delete cache → atualize DB → aguarde antes de deletar novamente) para minimizar a janela de inconsistência.

P Como monitoro a taxa de hit do cache?

R O comando INFO stats do Redis retorna keyspace_hits e keyspace_misses; a taxa de hit é calculada como hits / (hits + misses). Integre isso em um dashboard de monitoramento Prometheus + Grafana.

📖Resumo

Cliente Assíncrono Redis (redis.asyncio) + Pool de Conexão + Injeção de Dependência yield integrado com FastAPI
Modo Cache-Aside: Para leituras, verifique o cache primeiro; se não houver hit, consulte o banco de dados e escreva no cache; para escritas, atualize o banco de dados e delete a entrada de cache.
Chaves de cache são nomeadas segundo o padrão {entity}:{id}, com diferentes TTLs definidos para diferentes cenários
Cache de valores nulos para prevenir penetração, locks de exclusão mútua para prevenir rompimento, e TTL aleatório para prevenir avalanches — três principais estratégias de defesa
PriceTracker: Taxa de hit de cache para produtos populares é 80%; latência P99 caiu de 200 ms para 5 ms

📝Exercícios

Exercício Básico (Dificuldade ⭐): Configure a dependência do cliente assíncrono Redis, implemente endpoints de cache simples GET/SET, e verifique que os dados podem ser armazenados e recuperados. Dica: redis.asyncio + Depends(get_redis) + setex()
Problema Avançado (Dificuldade ⭐⭐): Implemente o padrão Cache-Aside para a busca de produtos do PriceTracker: Ler do cache → Se cache miss, consultar o DB → Escrever no cache (TTL 5 minutos); após a operação de escrita atualizar o DB, delete a entrada de cache. Dica: await redis.get(key) + await redis.delete(key)
Questão Desafio (Dificuldade ⭐⭐⭐): Implemente proteção abrangente de cache — cache de valores nulos para prevenir penetração (TTL curto de 60s), lock de exclusão mútua para prevenir rompimento (SET lock NX EX), e TTL aleatório para prevenir avalanches (300s ± 60s) — e escreva testes para verificar se a lógica de proteção está correta. Dica: redis.set(lock_key, "1", nx=True, ex=10) + random.randint()

---|

Equipe Técnica Web-Tutorial

Uma plataforma de tutoriais mantida por diversos desenvolvedores. Cada tutorial é escrito e revisado por profissionais da área correspondente. Trabalhamos para manter nosso conteúdo preciso e confiável — se encontrar algum problema, avise-nos.

Sobre Nós

404 Not Found

Cache — Redis para Acelerar Consultas Quentes

1. O Que Você Vai Aprender

2. A História Real da Alice

(1) Problema: Consultas Populares Sobrecarregam o Banco de Dados

(2) Soluções para Cache com Redis

(3) Resultado

3. Gerenciamento de Conexão Redis

(1) Integração do Cliente Assíncrono

(1) ▶Exemplo: Dependência de Conexão Redis

(2) Selecionando Estruturas de Dados do Redis

4. Modo Cache-Aside

(1) Processo de Leitura/Escrita

(1) ▶Exemplo: Consulta de Produto Cache-Aside

(2) ▶Exemplo: Invalidação proativa durante operações de escrita

(2) Convenções de Nomenclatura de Chave de Cache

5. Proteção contra Penetração/Rompimento/Avalanche de Cache

(1) Três Principais Problemas de Cache

(1) ▶Exemplo: Cache de Valores Nulos para Prevenir Penetração

(2) ▶Exemplo: Prevenção de Rompimento com Locks de Exclusão Mútua

(3) ▶Exemplo: TTL Aleatório para Prevenir Avalanches

(2) Comparação de Políticas de Coerência de Cache

❓Perguntas Frequentes

📖Resumo

📝Exercícios