Caching Strategies: Redis for High-Performance Applications

1. Cache-Aside (Lazy Loading) - MOST COMMON
┌─────────────────────────────────────────────────────────────────┐
│ Read: Check cache → if miss → read DB → store in cache         │
│ Write: Write to DB → invalidate cache                          │
│                                                                 │
│ Pros: Simple, only cache what's requested                      │
│ Cons: Cache miss penalty, stale data possible                  │
│ Use: Most web applications                                     │
└─────────────────────────────────────────────────────────────────┘

2. Read-Through
┌─────────────────────────────────────────────────────────────────┐
│ Read: Library handles cache → miss → loads from DB             │
│ Write: Application writes to DB                                │
│                                                                 │
│ Pros: Application doesn't manage cache                         │
│ Cons: Cache library complexity                                 │
│ Use: When using caching library (like Redis cache in Django)   │
└─────────────────────────────────────────────────────────────────┘

3. Write-Through
┌─────────────────────────────────────────────────────────────────┐
│ Write: Write to cache → cache writes to DB                     │
│ Read: Read from cache                                          │
│                                                                 │
│ Pros: Cache always consistent                                  │
│ Cons: Write latency increased                                  │
│ Use: When consistency is critical                             │
└─────────────────────────────────────────────────────────────────┘

4. Write-Behind (Write-Back)
┌─────────────────────────────────────────────────────────────────┐
│ Write: Write to cache only → async write to DB                 │
│ Read: Read from cache                                          │
│                                                                 │
│ Pros: Very fast writes                                         │
│ Cons: Data loss risk if cache fails                            │
│ Use: High-write throughput, where some data loss acceptable    │
└─────────────────────────────────────────────────────────────────┘

5. Write-Around
┌─────────────────────────────────────────────────────────────────┐
│ Write: Write to DB only                                        │
│ Read: Check cache → if miss → read DB (no cache population)    │
│                                                                 │
│ Pros: Cache doesn't get polluted with rarely read data         │
│ Cons: Cache misses for recently written data                   │
│ Use: Write-heavy, read-some workloads                          │
└─────────────────────────────────────────────────────────────────┘

import redis
import json

redis_client = redis.Redis(host='localhost', port=6379, decode_responses=True)

# String - simple key-value (most common)
redis_client.set('user:123', json.dumps({'name': 'Alice', 'email': 'alice@example.com'}))
user = json.loads(redis_client.get('user:123'))

# With expiration (TTL)
redis_client.setex('session:abc123', 3600, 'user:123')  # Expires in 1 hour

# Hash - structured objects
redis_client.hset('user:456', mapping={'name': 'Bob', 'email': 'bob@example.com'})
redis_client.hget('user:456', 'name')
redis_client.hgetall('user:456')

# List - queues, recent items
redis_client.lpush('recent_views:user:123', 'product:1', 'product:2', 'product:3')
redis_client.ltrim('recent_views:user:123', 0, 9)  # Keep last 10
recent = redis_client.lrange('recent_views:user:123', 0, -1)

# Set - unique items, tags, relationships
redis_client.sadd('product:1:tags', 'electronics', 'sale', 'new')
redis_client.sismember('product:1:tags', 'sale')  # Check if has tag 'sale'

# Sorted Set - leaderboards, rankings
redis_client.zadd('leaderboard', {'user:123': 1500, 'user:456': 1420, 'user:789': 1600})
top_players = redis_client.zrevrange('leaderboard', 0, 9, withscores=True)  # Top 10

# HyperLogLog - approximate unique counts
redis_client.pfadd('unique_visitors:2025-03-04', 'ip1', 'ip2', 'ip3', 'ip1')
unique_count = redis_client.pfcount('unique_visitors:2025-03-04')  # ≈ 3

# Bitmap - boolean flags, daily active users
redis_client.setbit('active_users:2025-03-04', 123, 1)  # User 123 active
is_active = redis_client.getbit('active_users:2025-03-04', 123)

# Geospatial - location-based data
redis_client.geoadd('locations', 13.361389, 38.115556, 'Palermo')
distances = redis_client.geodist('locations', 'Palermo', 'Catania', unit='km')

# Pub/Sub - real-time messaging
pubsub = redis_client.pubsub()
pubsub.subscribe('notifications')
# In another process:
redis_client.publish('notifications', json.dumps({'event': 'order_created', 'order_id': 123}))

"There are only two hard things in Computer Science: cache invalidation and naming things." - Phil Karlton

Cache Invalidation Strategies:

1. Time-To-Live (TTL) - Simplest
   ┌─────────────────────────────────────────────────────────────────┐
   │ redis_client.setex('key', 300, value)  # Expires in 5 minutes  │
   │                                                                 │
   │ Pros: Simple, automatic                                        │
   │ Cons: Stale data until TTL expires                            │
   │ Use: When eventual consistency acceptable                      │
   └─────────────────────────────────────────────────────────────────┘

2. Write Invalidation (Cache-Aside)
   ┌─────────────────────────────────────────────────────────────────┐
   │ def update_user(user_id, data):                                │
   │     db.execute("UPDATE users SET ...", data)                   │
   │     redis_client.delete(f"user:{user_id}")  # Invalidate cache │
   │                                                                 │
   │ Pros: Guaranteed fresh on next read                            │
   │ Cons: Extra delete operation                                   │
   └─────────────────────────────────────────────────────────────────┘

3. Write Update
   ┌─────────────────────────────────────────────────────────────────┐
   │ def update_user(user_id, data):                                │
   │     db.execute("UPDATE users SET ...", data)                   │
   │     redis_client.set(f"user:{user_id}", json.dumps(data))      │
   │                                                                 │
   │ Pros: Cache always fresh                                       │
   │ Cons: Two writes (DB + cache)                                  │
   └─────────────────────────────────────────────────────────────────┘

4. Versioned Keys
   ┌─────────────────────────────────────────────────────────────────┐
   │ version = db.query("SELECT version FROM users WHERE id = %s", user_id)│
   │ key = f"user:{user_id}:v{version}"                             │
   │                                                                 │
   │ Pros: Easy to detect staleness                                 │
   │ Cons: More complex                                             │
   └─────────────────────────────────────────────────────────────────┘

5. Pattern Invalidation
   ┌─────────────────────────────────────────────────────────────────┐
   │ # Invalidate all user caches for a user                        │
   │ keys = redis_client.keys(f"user:*:{user_id}")                  │
   │ redis_client.delete(*keys)                                     │
   │                                                                 │
   │ # Or use SCAN for production (don't use KEYS in prod!)        │
   │ for key in redis_client.scan_iter(match=f"user:*:{user_id}"):  │
   │     redis_client.delete(key)                                   │
   └─────────────────────────────────────────────────────────────────┘

Common Invalidation Scenarios:

# User profile update
@cache.invalidate(pattern=f"user:{user_id}")
def update_profile(user_id, data):
    return db.update_user(user_id, data)

# Product price change invalidates all product caches
@cache.invalidate(pattern=f"product:*")
def update_price(product_id, new_price):
    return db.update_product(product_id, {'price': new_price})

# Bulk invalidation on deployment
redis_client.flushall()  # Clear everything on deploy

# Selective flush by environment
redis_client.flushdb()   # Clear current database only

import threading
import time
import random

class CacheStampedeProtection:
    """Prevent multiple simultaneous cache rebuilds"""

    def __init__(self, redis_client):
        self.redis = redis_client
        self.locks = {}
        self.lock_lock = threading.Lock()

    def get_or_compute(self, key, compute_func, ttl=300):
        """Get from cache or compute with mutex protection"""
        # Try to get from cache
        cached = self.redis.get(key)
        if cached is not None:
            return cached

        # Acquire lock for this key
        with self.get_lock(key):
            # Double-check cache (another thread might have computed it)
            cached = self.redis.get(key)
            if cached is not None:
                return cached

            # Compute value
            value = compute_func()

            # Store in cache with jitter (to prevent simultaneous expiry)
            jitter = random.randint(0, 60)
            self.redis.setex(key, ttl + jitter, value)
            return value

    def get_lock(self, key):
        with self.lock_lock:
            if key not in self.locks:
                self.locks[key] = threading.Lock()
            return self.locks[key]

# Redis-based distributed lock (for multi-server)
import redis_lock

def get_with_distributed_lock(redis_client, key, compute_func, ttl=300):
    """Get from cache or compute with Redis distributed lock"""
    cached = redis_client.get(key)
    if cached:
        return cached

    lock = redis_lock.Lock(redis_client, f"lock:{key}", expire=30)

    if lock.acquire(blocking_timeout=5):
        try:
            # Double-check after acquiring lock
            cached = redis_client.get(key)
            if cached:
                return cached
            value = compute_func()
            redis_client.setex(key, ttl, value)
            return value
        finally:
            lock.release()
    else:
        # Wait for the value to be computed by another process
        time.sleep(1)
        return redis_client.get(key)

# Cache warmup (prevent cold start stampede)
def warmup_cache():
    """Pre-populate cache on application start"""
    popular_products = db.query("SELECT id FROM products ORDER BY views DESC LIMIT 100")
    for product in popular_products:
        compute_product_details(product.id)  # Populates cache

Redis Eviction Policies (when memory limit reached):

┌─────────────────────────────────────────────────────────────────┐
│ Policy          │ Behavior                                      │
├─────────────────┼───────────────────────────────────────────────┤
│ noeviction      │ Return error on write (default)              │
├─────────────────┼───────────────────────────────────────────────┤
│ allkeys-lru     │ Remove least recently used from all keys     │
│                 │ Best for general caching                     │
├─────────────────┼───────────────────────────────────────────────┤
│ volatile-lru    │ Remove LRU from keys with TTL only           │
├─────────────────┼───────────────────────────────────────────────┤
│ allkeys-random  │ Remove random key (fast but inefficient)     │
├─────────────────┼───────────────────────────────────────────────┤
│ volatile-random │ Remove random key with TTL                   │
├─────────────────┼───────────────────────────────────────────────┤
│ volatile-ttl    │ Remove key with shortest TTL first           │
└─────────────────┴───────────────────────────────────────────────┘

Configuring eviction:
# redis.conf
maxmemory 4gb
maxmemory-policy allkeys-lru

# At runtime
CONFIG SET maxmemory 4gb
CONFIG SET maxmemory-policy allkeys-lru

Setting TTL strategies:

def set_with_smart_ttl(key, value, popularity_score=None):
    """Set TTL based on popularity"""
    if popularity_score is None:
        ttl = 300  # 5 minutes for normal data
    elif popularity_score > 1000:
        ttl = 60   # 1 minute for very popular (fresher data)
    elif popularity_score > 100:
        ttl = 300  # 5 minutes for popular
    else:
        ttl = 3600 # 1 hour for unpopular (cache longer)

    redis_client.setex(key, ttl, value)

def set_with_randomized_ttl(key, value, base_ttl=300):
    """Add jitter to prevent simultaneous expiry stampedes"""
    jitter = random.randint(0, 60)
    redis_client.setex(key, base_ttl + jitter, value)

import hashlib
import json
import time
from functools import wraps
from typing import Any, Callable
import redis
import redis_lock

class CacheLayer:
    """Production-ready caching layer"""

    def __init__(self, redis_url='redis://localhost:6379/0'):
        self.redis = redis.from_url(redis_url, decode_responses=True)
        self.default_ttl = 300

    def cache(self, ttl: int = None, key_prefix: str = None):
        """Decorator for caching function results"""
        def decorator(func: Callable):
            @wraps(func)
            def wrapper(*args, **kwargs):
                # Generate cache key
                key_parts = [key_prefix or func.__name__]
                key_parts.extend(str(arg) for arg in args)
                key_parts.extend(f"{k}:{v}" for k, v in kwargs.items())
                cache_key = hashlib.md5("_".join(key_parts).encode()).hexdigest()

                # Try cache
                cached = self.redis.get(cache_key)
                if cached:
                    return json.loads(cached)

                # Compute and cache
                result = func(*args, **kwargs)
                cache_ttl = ttl or self.default_ttl
                self.redis.setex(cache_key, cache_ttl, json.dumps(result))
                return result
            return wrapper
        return decorator

    def cache_with_stampede_protection(self, ttl: int = None):
        """Decorator with stampede protection"""
        def decorator(func: Callable):
            @wraps(func)
            def wrapper(*args, **kwargs):
                # Generate cache key
                key_parts = [func.__name__]
                key_parts.extend(str(arg) for arg in args)
                key_parts.extend(f"{k}:{v}" for k, v in kwargs.items())
                cache_key = hashlib.md5("_".join(key_parts).encode()).hexdigest()

                # Try cache
                cached = self.redis.get(cache_key)
                if cached:
                    return json.loads(cached)

                # Distributed lock to prevent stampede
                lock = redis_lock.Lock(self.redis, f"lock:{cache_key}", expire=30)

                if lock.acquire(blocking_timeout=5):
                    try:
                        # Double-check after lock
                        cached = self.redis.get(cache_key)
                        if cached:
                            return json.loads(cached)
                        result = func(*args, **kwargs)
                        cache_ttl = ttl or self.default_ttl
                        self.redis.setex(cache_key, cache_ttl, json.dumps(result))
                        return result
                    finally:
                        lock.release()
                else:
                    # Another process is computing, wait for it
                    time.sleep(1)
                    return json.loads(self.redis.get(cache_key))
            return wrapper
        return decorator

    def invalidate_pattern(self, pattern: str):
        """Invalidate all keys matching pattern"""
        for key in self.redis.scan_iter(match=pattern):
            self.redis.delete(key)

    def get_or_set(self, key: str, compute_func: Callable, ttl: int = None):
        """Get from cache or compute and set"""
        cached = self.redis.get(key)
        if cached is not None:
            return json.loads(cached)

        value = compute_func()
        cache_ttl = ttl or self.default_ttl
        self.redis.setex(key, cache_ttl, json.dumps(value))
        return value

    def set_many(self, mapping: dict, ttl: int = None):
        """Set multiple keys with pipeline for efficiency"""
        pipe = self.redis.pipeline()
        for key, value in mapping.items():
            if ttl:
                pipe.setex(key, ttl, json.dumps(value))
            else:
                pipe.set(key, json.dumps(value))
        pipe.execute()

    def get_many(self, keys: list):
        """Get multiple keys with pipeline"""
        pipe = self.redis.pipeline()
        for key in keys:
            pipe.get(key)
        results = pipe.execute()
        return [json.loads(r) if r else None for r in results]


# Usage examples
cache = CacheLayer()

# Basic caching
@cache.cache(ttl=600, key_prefix='user_profile')
def get_user_profile(user_id: int):
    return db.query("SELECT * FROM users WHERE id = %s", user_id)

# Caching with stampede protection
@cache.cache_with_stampede_protection(ttl=300)
def get_expensive_computation(param: str):
    # Expensive operation
    return result

# Manual cache control
cache.invalidate_pattern("user_profile:*")
cache.set_many({
    "product:1": {"name": "Laptop", "price": 999},
    "product:2": {"name": "Mouse", "price": 29},
}, ttl=3600)

products = cache.get_many(["product:1", "product:2"])