O que Todo Dev Deve Saber sobre ElastiCache: Redis e Memcached para Cache em Alta Performance Já leu

ElastiCache: Fundamentos e Arquitetura

ElastiCache é um serviço gerenciado da AWS que oferece cache distribuído em memória, permitindo reduzir latência e melhorar performance de aplicações. Ele suporta dois engines principais: Redis e Memcached, cada um com características distintas que exploraremos aqui. A escolha entre eles não é trivial — compreender suas diferenças é essencial para arquitetar soluções escaláveis.

O cache funciona armazenando dados frequentemente acessados em memória (muito mais rápida que disco), eliminando consultas repetidas ao banco de dados. Uma requisição típica que levaria 100ms no banco pode ser resolvida em 1-5ms no cache, transformando a experiência do usuário. AWS gerencia replicação, failover e backups automaticamente, deixando você focar na lógica da aplicação.

Redis vs Memcached: Quando Usar Cada Um

Redis é uma estrutura de dados avançada com persistência, suporta Strings, Listas, Sets, Sorted Sets e Hashes. Ideal quando você precisa de operações complexas, expiração granular de chaves ou sincronização entre instâncias. Memcached é mais simples — apenas key-value com strings — mas extremamente rápido e consume menos recursos. Use Redis para dados com lógica complexa; Memcached para cache simples e horizontal scaling massivo.

Implementação Prática com Redis

Redis é mais robusto e oferece mais possibilidades. Vamos implementar um cache real para dados de usuário com Python usando redis-py:

import redis
import json
from datetime import timedelta

# Conectar ao ElastiCache Redis (endpoint fornecido pela AWS)
r = redis.Redis(
    host='seu-cluster.abc123.ng.0001.use1.cache.amazonaws.com',
    port=6379,
    decode_responses=True
)

class UserCache:
    def __init__(self, redis_client):
        self.r = redis_client
        self.ttl = timedelta(hours=1)

    def get_user(self, user_id):
        # Tentar obter do cache
        cached = self.r.get(f"user:{user_id}")
        if cached:
            return json.loads(cached)

        # Se não existe, buscar do DB (simulado)
        user = self._fetch_from_db(user_id)

        # Armazenar no cache com TTL de 1 hora
        self.r.setex(
            f"user:{user_id}",
            self.ttl,
            json.dumps(user)
        )
        return user

    def invalidate_user(self, user_id):
        self.r.delete(f"user:{user_id}")

    def _fetch_from_db(self, user_id):
        # Simulação de consulta ao banco
        return {
            "id": user_id,
            "name": "João",
            "email": "joao@example.com"
        }

# Uso
cache = UserCache(r)
user = cache.get_user(123)  # Primeira chamada: busca DB e cacheia
user = cache.get_user(123)  # Segunda: retorna do cache instantaneamente
cache.invalidate_user(123)  # Limpar cache quando dados mudam

Redis permite operações mais sofisticadas. Implementar um leaderboard com Sorted Sets é trivial:

# Adicionar scores (jogadores e pontos)
r.zadd("leaderboard", {"player1": 1000, "player2": 1500, "player3": 900})

# Obter top 10
top_10 = r.zrevrange("leaderboard", 0, 9, withscores=True)
print(top_10)  # [('player2', 1500.0), ('player1', 1000.0), ...]

# Incrementar score
r.zincrby("leaderboard", 50, "player1")

# Contar players acima de 1000 pontos
above_1000 = r.zcount("leaderboard", 1000, "+inf")
print(f"Players acima de 1000: {above_1000}")

Estratégias de Cache e Padrões Comuns

Existem três padrões principais: Cache-Aside, Write-Through e Write-Behind. Cache-Aside é o mais comum — seu código verifica o cache, se miss busca do DB e popula. Write-Through garante consistência escrevendo simultaneamente no cache e DB. Write-Behind (Lazy Write) melhora performance escrevendo assincronamente, mas risco de perda de dados existe.

Implementar invalidação eficiente é crítico. Usar cache tags, versioning ou event-driven invalidation previne stale data. Exemplo com invalidação por padrão:

class CacheManager:
    def __init__(self, redis_client):
        self.r = redis_client

    def cache_with_tag(self, key, value, tags=None, ttl=3600):
        """Cache com suporte a tags para invalidação em lote"""
        self.r.setex(key, ttl, json.dumps(value))

        if tags:
            for tag in tags:
                self.r.sadd(f"tag:{tag}", key)

    def invalidate_by_tag(self, tag):
        """Invalidar todas as chaves com uma tag"""
        keys = self.r.smembers(f"tag:{tag}")
        if keys:
            self.r.delete(*keys)
        self.r.delete(f"tag:{tag}")

# Uso
cm = CacheManager(r)
cm.cache_with_tag("user:123", user_data, tags=["user", "user:123"])
cm.cache_with_tag("product:456", product_data, tags=["product", "category:electronics"])

# Invalidar todos os usuários
cm.invalidate_by_tag("user")

# Invalidar todos os eletrônicos
cm.invalidate_by_tag("category:electronics")

Otimização e Monitoramento em Produção

Monitorar hit ratio, eviction rate e memory usage é fundamental. CloudWatch integra-se nativamente com ElastiCache. Configure alarmes para quando hit ratio cair abaixo de 80% (indicador de cache pequeno ou TTL curto demais) ou quando memory usage aproximar do limite.

Implementar circuit breaker é essencial — se cache falhar, sua aplicação não deve quebrar. Use timeout curtos e fallbacks:

import redis
from redis.exceptions import ConnectionError, TimeoutError as RedisTimeout

class ResilientCache:
    def __init__(self, redis_client, db_fallback):
        self.r = redis_client
        self.db = db_fallback
        self.timeout = 0.5  # 500ms

    def get(self, key, fetch_fn):
        try:
            result = self.r.get(key)
            if result:
                return json.loads(result)
        except (ConnectionError, RedisTimeout):
            # Cache indisponível, usar DB diretamente
            pass

        # Cache miss ou indisponível
        data = fetch_fn()

        # Tentar cachear sem bloquear
        try:
            self.r.setex(key, 3600, json.dumps(data))
        except:
            pass  # Se cache falha, continua normalmente

        return data

# Uso
resilient = ResilientCache(r, db_connection)
user = resilient.get("user:123", lambda: db.fetch_user(123))

Conclusão

ElastiCache é ferramenta imprescindível para aplicações em escala. Três pontos-chave: (1) Redis oferece versatilidade com múltiplos tipos de dados e persistência, enquanto Memcached é minimalista e ultra-rápido — escolha conforme complexidade do seu caso; (2) Implementar padrões corretos de invalidação (Cache-Aside com tags ou event-driven) evita dados obsoletos; (3) Resiliência é crítica — sempre tenha fallbacks para quando cache falha, monitore métricas continuamente e dimensione adequadamente para seu hit ratio esperado.

Referências

• AWS ElastiCache Official Documentation
• Redis Documentation
• redis-py Client Library
• AWS ElastiCache Best Practices
• High Performance in-memory Caching Patterns