Caching Patterns
Store frequently accessed data in fast-access storage to reduce latency and backend load
Caching stores copies of frequently accessed data in fast-access storage layers (memory, SSD) to reduce latency, decrease database load, and improve application performance. Different caching strategies offer different trade-offs between consistency, performance, and complexity.
Use caching when you have read-heavy workloads, expensive computations, frequently accessed data, or need to reduce database load and improve response times.
General-purpose caching for web applications, user profiles, product catalogs, and session data
# Cache-Aside Pattern Implementation
def get_user(user_id):
# Try to get from cache first
cache_key = f"user:{user_id}"
user = cache.get(cache_key)
if user is not None:
return user # Cache hit
# Cache miss - fetch from database
user = database.query("SELECT * FROM users WHERE id = ?", user_id)
if user:
# Populate cache for future requests
cache.set(cache_key, user, ttl=3600) # 1 hour TTL
return user
def update_user(user_id, data):
# Update database
database.update("UPDATE users SET ... WHERE id = ?", user_id, data)
# Invalidate cache to ensure consistency
cache_key = f"user:{user_id}"
cache.delete(cache_key)
Financial systems, inventory management, booking systems where consistency is critical
# Write-Through Cache Pattern
def save_user(user_id, data):
cache_key = f"user:{user_id}"
# Write to database first
database.update("UPDATE users SET ... WHERE id = ?", user_id, data)
# Then update cache (write-through)
cache.set(cache_key, data, ttl=3600)
return data
def get_user(user_id):
cache_key = f"user:{user_id}"
user = cache.get(cache_key)
if user is None:
# Cache miss - load from database
user = database.query("SELECT * FROM users WHERE id = ?", user_id)
if user:
cache.set(cache_key, user, ttl=3600)
return user
Analytics systems, logging, metrics collection, social media likes/views counters
# Write-Behind Cache Pattern
import asyncio
from collections import defaultdict
class WriteBehindCache:
def __init__(self):
self.cache = {}
self.write_queue = defaultdict(dict)
self.flush_interval = 5 # seconds
async def set(self, key, value):
# Write to cache immediately
self.cache[key] = value
# Queue for async database write
self.write_queue[key] = value
return value
async def flush_worker(self):
"""Background worker to flush writes to database"""
while True:
await asyncio.sleep(self.flush_interval)
if self.write_queue:
# Batch write to database
batch = dict(self.write_queue)
self.write_queue.clear()
try:
await database.batch_update(batch)
except Exception as e:
# On failure, re-queue writes
self.write_queue.update(batch)
logger.error(f"Failed to flush cache: {e}")
# Usage
cache = WriteBehindCache()
asyncio.create_task(cache.flush_worker())
async def increment_view_count(post_id):
key = f"post:{post_id}:views"
current = cache.cache.get(key, 0)
await cache.set(key, current + 1)
Uses Memcached for caching user data, reducing database queries by 90%
Trillions of cache requests per dayCaches timelines and tweets using Redis, serving millions of reads per second
500+ million tweets per dayMulti-layer caching with Redis and Memcached for pins and boards
Billions of pins, 400+ million usersHorizontal - Add more cache nodes with sharding
Medium - Cache invalidation and consistency challenges
Low to Medium - Memory costs, but reduces database costs