<h2>ElastiCache: Fundamentos e Arquitetura</h2>
<p>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.</p>
<p>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.</p>
<h3>Redis vs Memcached: Quando Usar Cada Um</h3>
<p><strong>Redis</strong> é 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. <strong>Memcached</strong> é 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.</p>
<div class="table-wrap"><table><thead><tr><th>Característica</th><th>Redis</th><th>Memcached</th></tr></thead><tbody><tr><td>Persistência</td><td>Sim (RDB/AOF)</td><td>Não</td></tr><tr><td>Tipos de Dados</td><td>Múltiplos</td><td>String apenas</td></tr><tr><td>Replicação</td><td>Sim</td><td>Não (cluster)</td></tr><tr><td>TTL por chave</td><td>Sim</td><td>Sim</td></tr><tr><td>Cluster</td><td>Sim (Redis Cluster)</td><td>Consistent Hashing</td></tr></tbody></table></div>
<h2>Implementação Prática com Redis</h2>
<p>Redis é mais robusto e oferece mais possibilidades. Vamos implementar um cache real para dados de usuário com Python usando <code>redis-py</code>:</p>
<pre><code class="language-python">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</code></pre>
<p>Redis permite operações mais sofisticadas. Implementar um leaderboard com Sorted Sets é trivial:</p>
<pre><code class="language-python"># 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}")</code></pre>
<h2>Estratégias de Cache e Padrões Comuns</h2>
<p>Existem três padrões principais: <strong>Cache-Aside</strong>, <strong>Write-Through</strong> e <strong>Write-Behind</strong>. 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.</p>
<p>Implementar invalidação eficiente é crítico. Usar cache tags, versioning ou event-driven invalidation previne stale data. Exemplo com invalidação por padrão:</p>
<pre><code class="language-python">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")</code></pre>
<h2>Otimização e Monitoramento em Produção</h2>
<p>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.</p>
<p>Implementar circuit breaker é essencial — se cache falhar, sua aplicação não deve quebrar. Use timeout curtos e fallbacks:</p>
<pre><code class="language-python">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))</code></pre>
<h2>Conclusão</h2>
<p>ElastiCache é ferramenta imprescindível para aplicações em escala. Três pontos-chave: <strong>(1)</strong> 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; <strong>(2)</strong> Implementar padrões corretos de invalidação (Cache-Aside com tags ou event-driven) evita dados obsoletos; <strong>(3)</strong> Resiliência é crítica — sempre tenha fallbacks para quando cache falha, monitore métricas continuamente e dimensione adequadamente para seu hit ratio esperado.</p>
<h2>Referências</h2>
<ul>
<li><a href="https://docs.aws.amazon.com/elasticache/" target="_blank" rel="noopener noreferrer">AWS ElastiCache Official Documentation</a></li>
<li><a href="https://redis.io/documentation" target="_blank" rel="noopener noreferrer">Redis Documentation</a></li>
<li><a href="https://github.com/redis/redis-py" target="_blank" rel="noopener noreferrer">redis-py Client Library</a></li>
<li><a href="https://docs.aws.amazon.com/elasticache/latest/red-ug/BestPractices.html" target="_blank" rel="noopener noreferrer">AWS ElastiCache Best Practices</a></li>
<li><a href="https://aws.amazon.com/blogs/database/caching-strategies-with-amazon-elasticache/" target="_blank" rel="noopener noreferrer">High Performance in-memory Caching Patterns</a></li>
</ul>