Caching is the most impactful performance optimization available. A well-designed caching strategy can reduce latency by 10-100x and reduce load on origin systems by 90% or more. This guide covers caching at every layer - from CDN to application to database.
The Caching Pyramid
┌─────────────────────────────────────┐
│ CDN / Edge Cache │ ← Milliseconds, global
├─────────────────────────────────────┤
│ Application Cache │ ← Milliseconds
├─────────────────────────────────────┤
│ Database Query Cache │ ← Milliseconds
├─────────────────────────────────────┤
│ Primary Database │ ← Milliseconds to seconds
└─────────────────────────────────────┘
Each layer caches different data with different characteristics.
Redis Deep Dive
Redis Data Structures
import redis
import json
r = redis.Redis(host='localhost', port=6379, db=0)
# Strings - Simple values
r.set('user:123:name', 'John Doe')
r.set('user:123:email', '[email protected]')
r.setex('session:abc123', 3600, json.dumps({'user_id': 123}))
# Hashes - Objects
r.hset('user:123', mapping={
'name': 'John Doe',
'email': '[email protected]',
'created_at': '2024-01-15'
})
user = r.hgetall('user:123')
# Lists - Ordered collections
r.lpush('recent_searches:user123', 'shoes')
r.ltrim('recent_searches:user123', 0, 9) # Keep only 10
# Sets - Unique collections
r.sadd('product:views:2024-01-15', 'product1', 'product2', 'product3')
unique_viewers = r.scard('product:views:2024-01-15')
# Sorted Sets - Leaderboards
r.zadd('leaderboard:score', {'player1': 100, 'player2': 95, 'player3': 90})
top_players = r.zrevrange('leaderboard:score', 0, 9, withscores=True)
Caching Patterns
Cache-Aside (Lazy Loading):
# Standard cache-aside pattern
def get_user(user_id):
# Try cache first
cached = redis.get(f'user:{user_id}')
if cached:
return json.loads(cached)
# Cache miss - fetch from database
user = db.query('SELECT * FROM users WHERE id = ?', user_id)
if user:
# Store in cache with TTL
redis.setex(f'user:{user_id}', 3600, json.dumps(user))
return user
def update_user(user_id, data):
# Update database first
db.execute('UPDATE users SET ... WHERE id = ?', user_id, data)
# Invalidate cache
redis.delete(f'user:{user_id}')
Write-Through:
# Write-through - cache updated on every write
def create_order(order_data):
# Write to database
order_id = db.insert('orders', order_data)
# Write to cache immediately
cache_key = f'order:{order_id}'
redis.setex(cache_key, 3600, json.dumps({**order_data, 'id': order_id}))
return order_id
def update_order(order_id, data):
db.update('orders', data, 'id = ?', order_id)
# Update cache
cache_key = f'order:{order_id}'
cached = redis.get(cache_key)
if cached:
order = json.loads(cached)
order.update(data)
redis.setex(cache_key, 3600, json.dumps(order))
Write-Behind:
# Write-behind - async database writes
import asyncio
from collections import deque
write_queue = deque()
async def update_user_async(user_id, data):
# Update cache immediately
redis.setex(f'user:{user_id}', 3600, json.dumps(data))
# Queue for async database write
write_queue.append({
'table': 'users',
'id': user_id,
'data': data
})
async def flush_writes():
while True:
if write_queue:
batch = []
while write_queue and len(batch) < 100:
batch.append(write_queue.popleft())
# Bulk write to database
db.batch_update('users', batch)
await asyncio.sleep(1)
Redis Cluster Patterns
# Redis Cluster connection
from redis.cluster import RedisCluster
nodes = [
{'host': 'redis-1', 'port': 6379},
{'host': 'redis-2', 'port': 6379},
{'host': 'redis-3', 'port': 6379},
]
rc = RedisCluster(startup_nodes=nodes, decode_responses=True)
# Automatic key distribution
rc.set('user:1:name', 'Alice')
rc.set('user:2:name', 'Bob')
# Keys automatically sharded across nodes
Redis Pub/Sub for Cache Invalidation
# Publisher - notify other services of changes
def invalidate_user_cache(user_id):
# Local cache delete
redis.delete(f'user:{user_id}')
# Notify other instances
redis.publish('cache:invalidate', json.dumps({
'key': f'user:{user_id}',
'pattern': f'user:{user_id}:*'
}))
# Subscriber - listen for invalidations
def listen_for_invalidation():
pubsub = redis.pubsub()
pubsub.subscribe('cache:invalidate')
for message in pubsub.listen():
if message['type'] == 'message':
data = json.loads(message['data'])
if '*' in data['pattern']:
# Handle pattern
for key in redis.keys(data['pattern']):
redis.delete(key)
else:
redis.delete(data['key'])
Application-Level Caching
In-Memory Caching
from functools import lru_cache
from threading import Lock
import time
# Simple in-memory cache with TTL
class Cache:
def __init__(self, ttl=300):
self._cache = {}
self._timestamps = {}
self._ttl = ttl
self._lock = Lock()
def get(self, key):
with self._lock:
if key in self._cache:
if time.time() - self._timestamps[key] < self._ttl:
return self._cache[key]
del self._cache[key]
del self._timestamps[key]
return None
def set(self, key, value):
with self._lock:
self._cache[key] = value
self._timestamps[key] = time.time()
def delete(self, key):
with self._lock:
self._cache.pop(key, None)
self._timestamps.pop(key, None)
def clear(self):
with self._lock:
self._cache.clear()
self._timestamps.clear()
# Usage with LRU cache
@lru_cache(maxsize=1000)
def get_product_category(product_id):
# Expensive database query
return db.query('SELECT category FROM products WHERE id = ?', product_id)
# Manual cache management
from cachetools import TTLCache, cached
product_cache = TTLCache(maxsize=10000, ttl=300)
@cached(cache=product_cache)
def get_product(product_id):
return db.query('SELECT * FROM products WHERE id = ?', product_id)
Distributed Caching Patterns
# Multi-layer cache with local + Redis
class MultiLayerCache:
def __init__(self, local_ttl=60, remote_ttl=3600):
self.local = {} # Local dict
self.redis = redis.Redis()
self.local_ttl = local_ttl
self.remote_ttl = remote_ttl
def get(self, key):
# Check local cache first
if key in self.local:
value, timestamp = self.local[key]
if time.time() - timestamp < self.local_ttl:
return value
del self.local[key]
# Check Redis
value = self.redis.get(key)
if value:
# Populate local cache
self.local[key] = (value, time.time())
return json.loads(value)
return None
def set(self, key, value):
# Write to both layers
self.redis.setex(key, self.remote_ttl, json.dumps(value))
self.local[key] = (value, time.time())
CDN Caching
Cache-Control Headers
# Dynamic response with proper cache headers
def get_product(request, product_id):
product = db.get_product(product_id)
# Per-user responses shouldn't be cached
if request.user:
return jsonify(product) # No cache headers = no CDN caching
# Public, cacheable content
response = jsonify(product)
response.headers['Cache-Control'] = 'public, max-age=300'
response.headers['Vary'] = 'Accept-Encoding'
return response
def get_static_assets(request, filename):
# Long cache for static assets (with versioned URLs)
response = send_file(f'static/{filename}')
response.headers['Cache-Control'] = 'public, max-age=31536000, immutable'
return response
def get_user_dashboard(request):
# Private - only browser cache
response = jsonify(request.user.dashboard)
response.headers['Cache-Control'] = 'private, max-age=0, no-cache'
return response
CDN Cache Invalidation
# CloudFront invalidation via Lambda
import boto3
def invalidate_cdn(paths):
cloudfront = boto3.client('cloudfront')
response = cloudfront.create_invalidation(
DistributionId='E1234567890ABC',
InvalidationBatch={
'CallerReference': f'invalidation-{int(time.time())}',
'Paths': {
'Quantity': len(paths),
'Items': paths
}
}
)
return response['Invalidation']['Id']
# Usage
invalidate_cdn([
'/api/products/*',
'/static/images/*'
])
// Cloudflare cache purge via API
async function purgeCache(urls) {
const response = await fetch(
'https://api.cloudflare.com/client/v4/zones/:zone_id/purge_cache',
{
method: 'POST',
headers: {
'Authorization': `Bearer ${API_TOKEN}`,
'Content-Type': 'application/json'
},
body: JSON.stringify({
files: urls,
tags: ['product-update'],
prefixs: ['/api/']
})
}
);
return response.json();
}
Cache Invalidation Strategies
Time-Based Invalidation
# Fixed TTL
CACHE_TTL = {
'user_profile': 3600, # 1 hour
'product_list': 300, # 5 minutes
'config': 86400, # 24 hours
'analytics': 60, # 1 minute
}
def get_cached(key, ttl_key, fetcher):
cached = redis.get(key)
if cached:
return json.loads(cached)
value = fetcher()
redis.setex(key, CACHE_TTL[ttl_key], json.dumps(value))
return value
Event-Based Invalidation
# Invalidate on data changes
def on_product_updated(product_id):
# Invalidate specific cache
redis.delete(f'product:{product_id}')
# Invalidate list caches
for pattern in ['products:list:*', 'products:category:*']:
for key in redis.keys(pattern):
redis.delete(key)
# Notify other services
redis.publish('cache:invalidate', {
'type': 'product',
'id': product_id
})
Stale-While-Revalidate
# Serve stale content while refreshing in background
def get_product_with_revalidation(product_id):
cache_key = f'product:{product_id}'
# Try to get from cache
cached = redis.get(cache_key)
if cached:
product = json.loads(cached)
# Check if stale
is_stale = redis.ttl(cache_key) < 60
if is_stale:
# Schedule background refresh (don't block)
asyncio.create_task(refresh_product_cache(product_id))
return product
# Cache miss - fetch synchronously
return refresh_product_cache(product_id)
async def refresh_product_cache(product_id):
product = db.get_product(product_id)
# Update cache
redis.setex(f'product:{product_id}', 3600, json.dumps(product))
return product
Cache Monitoring
Key Metrics
# Prometheus Redis exporter configuration
- name: redis
rules:
- alert: CacheHitRateLow
expr: |
(redis_keyspace_hits_total /
(redis_keyspace_hits_total + redis_keyspace_misses_total)) < 0.8
for: 5m
labels:
severity: warning
annotations:
summary: "Cache hit rate below 80%"
- alert: RedisMemoryHigh
expr: redis_memory_used_bytes / redis_memory_max_bytes > 0.9
for: 5m
labels:
severity: warning
annotations:
summary: "Redis memory usage above 90%"
- alert: RedisEvictionsHigh
expr: rate(redis_evicted_keys_total[5m]) > 10
for: 5m
labels:
severity: warning
annotations:
summary: "High number of key evictions"
Cache Analytics
# Cache hit/miss tracking
class CacheMetrics:
def __init__(self):
self.hits = 0
self.misses = 0
self.errors = 0
def record_hit(self):
self.hits += 1
def record_miss(self):
self.misses += 1
@property
def hit_rate(self):
total = self.hits + self.misses
return self.hits / total if total > 0 else 0
metrics = CacheMetrics()
def get_cached(key):
try:
value = redis.get(key)
if value:
metrics.record_hit()
return json.loads(value)
metrics.record_miss()
return None
except Exception:
metrics.errors += 1
return None
Memcached vs Redis
| Feature | Redis | Memcached |
|---|---|---|
| Data structures | Strings, Hashes, Lists, Sets, Sorted Sets | Strings only |
| Persistence | RDB + AOF | No |
| Clustering | Redis Cluster | Memcached SASL |
| Pub/Sub | Yes | No |
| Lua scripts | Yes | No |
| Memory efficiency | Moderate | Higher |
| Performance | Very fast | Extremely fast |
| Use case | Rich data, persistence | Simple caching |
# Memcached is simpler - no complex data types
import pymemcache
mc = pymemcache.Client(('localhost', 11211))
# Simple string caching only
mc.set('key', 'value')
value = mc.get('key')
# No TTL with default
mc.set('key', 'value', expire=3600)
Best Practices
Cache Anti-Patterns
Don’t cache too much data:
# Bad: Caching entire database rows
user = cache.get(f'user:{user_id}') # Contains password hash!
# Good: Cache only what's needed
user_summary = cache.get(f'user:{user_id}:summary')
Don’t use indefinite TTLs:
# Bad: No expiration
redis.set('config', json.dumps(config))
# Good: Reasonable TTL with refresh
redis.setex('config', 3600, json.dumps(config))
Handle cache misses gracefully:
# Bad: Cache stampede
for i in range(100): # 100 simultaneous requests
user = get_user(123) # All hit DB simultaneously
# Good: Request coalescing
lock = redis.lock('user:123:lock', timeout=5)
if lock.acquire():
try:
user = get_user(123)
finally:
lock.release()
else:
# Wait for other request to populate cache
time.sleep(0.1)
user = get_user(123)
Conclusion
Effective caching requires strategy, not just technology:
- Use CDN for all static content
- Implement cache-aside for read-heavy workloads
- Use write-through for critical data
- Implement cache invalidation on data changes
- Monitor hit rates and adjust TTLs
- Plan for cache failures gracefully
Start with CDN, add Redis for session and query caching, then layer in application caches as needed.
External Resources
Related Articles
- Frontend Performance Optimization - Client-side caching
- CDN Configuration - Edge caching
- Database DevOps - Database caching
Comments