Caching Strategies: Patterns and Best Practices

Caching is one of the highest-leverage optimizations in system design. A well-designed cache can reduce latency by 10-100x and reduce database load by 90%+.

Why Cache?

Latency: Memory (ns-μs) vs Disk (ms) vs Network (ms)
Throughput: Reduce load on primary data store
Cost: Serve more requests with same infrastructure
Availability: Cache can serve stale data during outages

Cache Patterns

1. Cache-Aside (Lazy Loading)

Most common pattern. Application manages cache.

Read:
  1. Check cache
  2. If hit → return data
  3. If miss → query DB → store in cache → return data

Write:
  1. Update DB
  2. Invalidate/Update cache

# Cache-aside implementation
class CacheAside:
    def __init__(self, cache, db):
        self.cache = cache
        self.db = db
    
    def get(self, key: str):
        # Try cache first
        value = self.cache.get(key)
        if value is not None:
            return value
        
        # Cache miss - fetch from DB
        value = self.db.get(key)
        if value is not None:
            self.cache.set(key, value, ttl=300)
        return value
    
    def set(self, key: str, value: any):
        # Write to DB first
        self.db.set(key, value)
        # Then invalidate cache (or update)
        self.cache.delete(key)

Pros: Simple, cache only stores hot data, resilient to cache failures Cons: Cache miss penalty, potential for stale data on write

2. Read-Through

Cache sits between app and DB. Cache handles misses.

Read:
  1. App requests from cache
  2. Cache checks → if miss, loads from DB → returns to app

# Read-through cache (transparent to app)
class ReadThroughCache:
    def __init__(self, cache, db):
        self.cache = cache
        self.db = db
    
    def get(self, key: str):
        value = self.cache.get(key)
        if value is None:
            # Cache loads from DB automatically
            value = self.db.get(key)
            if value is not None:
                self.cache.set(key, value, ttl=300)
        return value

Pros: Application logic simpler, cache manages population Cons: More complex cache implementation, first request always slow

3. Write-Through

Writes go to cache and DB synchronously.

Write:
  1. App writes to cache
  2. Cache writes to DB (synchronously)
  3. Return success

class WriteThroughCache:
    def set(self, key: str, value: any):
        # Write to both cache and DB
        self.cache.set(key, value, ttl=300)
        self.db.set(key, value)  # Synchronous

Pros: Cache always consistent with DB, no stale reads Cons: Higher write latency, DB must be available

4. Write-Back (Write-Behind)

Writes go to cache first, DB updated asynchronously.

Write:
  1. App writes to cache
  2. Return success immediately
  2. Cache asynchronously writes to DB (batched)

import asyncio
from collections import deque

class WriteBackCache:
    def __init__(self, cache, db, flush_interval=5):
        self.cache = cache
        self.db = db
        self.write_queue = deque()
        self.flush_interval = flush_interval
        asyncio.create_task(self._flush_loop())
    
    def set(self, key: str, value: any):
        self.cache.set(key, value, ttl=300)
        self.write_queue.append((key, value))
    
    async def _flush_loop(self):
        while True:
            await asyncio.sleep(self.flush_interval)
            await self._flush()
    
    async def _flush(self):
        if not self.write_queue:
            return
        # Batch write to DB
        batch = []
        while self.write_queue:
            batch.append(self.write_queue.popleft())
        await self.db.batch_set(batch)

Pros: Lowest write latency, batches DB writes Cons: Data loss risk on cache failure, complex implementation

5. Refresh-Ahead

Proactively refresh cache before expiration.

class RefreshAheadCache:
    def __init__(self, cache, db, refresh_threshold=0.8):
        self.cache = cache
        self.db = db
        self.refresh_threshold = refresh_threshold  # 80% of TTL
    
    async def get(self, key: str):
        value, ttl = await self.cache.get_with_ttl(key)
        
        # Proactively refresh if near expiration
        if value is not None and ttl < self.cache.default_ttl * self.refresh_threshold:
            asyncio.create_task(self._refresh(key))
        
        if value is None:
            value = await self.db.get(key)
            if value:
                await self.cache.set(key, value)
        return value
    
    async def _refresh(self, key: str):
        value = await self.db.get(key)
        if value:
            await self.cache.set(key, value)

Cache Invalidation Strategies

Time-Based (TTL)

# Simple TTL
cache.set(key, value, ttl=300)  # 5 minutes

Event-Based (Explicit Invalidation)

# On write, invalidate related keys
def update_user(user_id, data):
    db.update_user(user_id, data)
    cache.delete(f"user:{user_id}")
    cache.delete(f"user:{user_id}:profile")
    cache.delete(f"user:{user_id}:posts")

Key-Based Invalidation (Versioning)

# Include version in key
cache_key = f"user:{user_id}:v{user.version}"

# On update, increment version
user.version += 1
db.save(user)
# Old cache key becomes unreachable

Cache Tags (Redis 7+)

# Tag related keys
SET user:1:profile "..." EX 300
SADD tag:user:1 user:1:profile

# Invalidate all user:1 keys
SMEMBERS tag:user:1 → DEL each key

Common Pitfalls

1. Thundering Herd

Multiple requests hit expired cache simultaneously.

# Solution: Use distributed lock
async def get_with_lock(key: str):
    value = await cache.get(key)
    if value:
        return value
    
    # Try to acquire lock
    lock = await redis.set(f"lock:{key}", "1", nx=True, ex=10)
    if lock:
        try:
            value = await db.get(key)
            await cache.set(key, value)
            return value
        finally:
            await redis.delete(f"lock:{key}")
    else:
        # Wait for other request to populate
        await asyncio.sleep(0.1)
        return await cache.get(key)

2. Cache Stampede (Dogpile Effect)

Same as thundering herd - use probabilistic early expiration.

# Add jitter to TTL
import random
ttl = base_ttl + random.randint(0, base_ttl // 4)
cache.set(key, value, ttl=ttl)

3. Hot Keys

Single key gets disproportionate traffic.

# Solution: Key sharding + local cache
def get_sharded(key: str):
    # Split hot key across multiple cache entries
    shard = hash(key) % 10
    return cache.get(f"{key}:shard:{shard}")

4. Cache Pollution

Cache filled with rarely-accessed data.

# Solution: LRU eviction + frequency-based admission
# Redis: maxmemory-policy allkeys-lfu
# Or use TinyLFU / W-TinyLFU (Ristretto, Caffeine)

Multi-Level Caching

Request → CDN (edge) → API Gateway → Redis (distributed) → Local (in-memory) → DB
              ↓              ↓               ↓                ↓
           Static         Auth/Rate       Session/         Hot data
           Assets         Limit           User Data

// Multi-level cache in Go
type MultiLevelCache struct {
    local  *ristretto.Cache  // In-process, ~100MB
    shared *redis.Client     // Distributed
}

func (m *MultiLevelCache) Get(key string) (string, bool) {
    // L1: Local cache (fastest)
    if val, ok := m.local.Get(key); ok {
        return val.(string), true
    }
    
    // L2: Shared cache
    val, err := m.shared.Get(ctx, key).Result()
    if err == nil {
        m.local.Set(key, val, 1)
        return val, true
    }
    
    return "", false
}

Cache Sizing

Cache Size	Hit Rate (Typical)	Use Case
1% of data	50-70%	Very hot subset
5% of data	80-90%	Most applications
10% of data	90-95%	Read-heavy workloads
20%+ of data	95%+	Specialized

Monitoring Metrics

# Key cache metrics
cache_hit_rate{instance="..."}  # Target: >90%
cache_miss_rate{instance="..."}
cache_eviction_rate{instance="..."}
cache_memory_usage_bytes{instance="..."}
cache_latency_p99{instance="..."}  # Target: <1ms