Redis is fast—sub-millisecond reads. Databases are slow—10-100ms. Caching bridges the gap, reducing database load by 90%.

Why Caching Matters

Database query: 50ms Redis cache hit: 1ms Cache hit rate: 95%

Average response time = (0.95 × 1ms) + (0.05 × 50ms) = 3.5ms
Without cache: 50ms

That's 14x faster with 95% cache hit rate.

Pattern 1: Cache-Aside (Lazy Loading)

Most common. Code checks cache; if miss, query DB and populate cache.

def get_user(user_id):
    # Check cache first
    cached = redis.get(f"user:{user_id}")
    if cached:
        return json.loads(cached)
    
    # Cache miss: query database
    user = db.query("SELECT * FROM users WHERE id = %s", user_id)
    
    # Populate cache (expires in 1 hour)
    redis.setex(f"user:{user_id}", 3600, json.dumps(user))
    
    return user

Advantages

• Simple to implement
• Flexible TTL (time-to-live)
• Handles missing data gracefully

Disadvantages

• Cache misses are slow (hit DB)
• Stale data possible (if DB updates)
• Thundering herd: Many requests on miss → many DB queries

Pattern 2: Write-Through

Cache is updated immediately on writes. Reads only hit cache (usually).

def update_user(user_id, data):
    # Update database
    db.execute("UPDATE users SET ... WHERE id = %s", user_id, data)
    
    # Update cache immediately
    redis.setex(f"user:{user_id}", 3600, json.dumps(data))
    
    return data

def get_user(user_id):
    # Always in cache (should be)
    cached = redis.get(f"user:{user_id}")
    if cached:
        return json.loads(cached)
    
    # Fallback (shouldn't happen if writes are consistent)
    user = db.query("SELECT * FROM users WHERE id = %s", user_id)
    redis.setex(f"user:{user_id}", 3600, json.dumps(user))
    return user

Advantages

• Faster reads (always cache hits)
• Data consistency (cache updated on write)

Disadvantages

• Complex logic (must update both DB and cache)
• Write latency (two operations)
• Cache grows unbounded (need eviction policy)

Pattern 3: Write-Behind (Write-Back)

Asynchronously write to cache, then background job updates DB.

# Write to cache immediately (fast write)
def update_user(user_id, data):
    redis.setex(f"user:{user_id}", 3600, json.dumps(data))
    
    # Queue background job
    job_queue.add("sync_user", user_id, data)
    return data

# Background worker
def sync_user_to_db(user_id, data):
    time.sleep(5)  # Batch updates
    db.execute("UPDATE users SET ... WHERE id = %s", user_id, data)

Advantages

• Fastest writes (only Redis, ~1ms)
• Reduces DB load significantly

Disadvantages

• Risk of data loss (Redis crash before DB sync)
• Complexity (background jobs, error handling)
• Eventual consistency (not immediate)

Use only for non-critical data (session state, counters).

Cache Invalidation Strategies

Invalidation is hard. There are only two hard things in Computer Science: cache invalidation and naming things.

Strategy 1: TTL-Based (Expiration)

# Set expiration on cache entry
redis.setex(f"user:{user_id}", 3600, json.dumps(user))  # 1 hour TTL

Simple but can serve stale data.

Strategy 2: Event-Based (Immediate)

On writes, invalidate cache:

def update_user(user_id, data):
    db.execute("UPDATE users SET ... WHERE id = %s", user_id, data)
    
    # Invalidate cache immediately
    redis.delete(f"user:{user_id}")

Next read triggers refresh.

Strategy 3: Tag-Based (Cascade)

Invalidate related keys:

# On write, invalidate user AND user's posts
def update_user(user_id, data):
    db.execute("UPDATE users SET ... WHERE id = %s", user_id, data)
    
    redis.delete(f"user:{user_id}")
    
    # Also delete user's posts (related data)
    redis.delete(f"posts:user:{user_id}")
    
    # Or use tags (if Redis 7.0+)
    redis.delete(redis.smembers(f"user:{user_id}:tags"))

Strategy 4: Versioning

Change key on update:

def update_user(user_id, data):
    # Get current version
    version = redis.get(f"user:{user_id}:version") or 1
    new_version = int(version) + 1
    
    # Write new version
    redis.setex(f"user:{user_id}:v{new_version}", 3600, json.dumps(data))
    
    # Update version pointer
    redis.setex(f"user:{user_id}:version", 3600, new_version)
    
    # Update DB
    db.execute("UPDATE users SET ... WHERE id = %s", user_id, data)

def get_user(user_id):
    version = redis.get(f"user:{user_id}:version")
    if version:
        cached = redis.get(f"user:{user_id}:v{version}")
        if cached:
            return json.loads(cached)
    
    # Fallback
    user = db.query("SELECT * FROM users WHERE id = %s", user_id)
    redis.setex(f"user:{user_id}", 3600, json.dumps(user))
    return user

Avoids delete operations (atomic key updates).

Thundering Herd Solution

When cache expires, many requests hit DB simultaneously:

def get_with_lock(key, db_query, ttl=3600):
    # Check cache
    cached = redis.get(key)
    if cached:
        return json.loads(cached)
    
    # Acquire lock (only one wins)
    lock_key = f"{key}:lock"
    if redis.set(lock_key, "1", nx=True, ex=10):
        # Lock acquired: fetch from DB
        try:
            data = db_query()
            redis.setex(key, ttl, json.dumps(data))
        finally:
            redis.delete(lock_key)  # Release lock
        return data
    else:
        # Lock held: wait and retry cache
        time.sleep(0.1)
        return get_with_lock(key, db_query, ttl)

Only one request hits DB; others wait.

Caching Data Structures

Different data → different strategies:

User Profile (infrequent changes)

redis.hset(f"user:{user_id}", mapping={
    "name": "John",
    "email": "john@example.com",
    "role": "admin"
})
redis.expire(f"user:{user_id}", 3600)

# Fetch specific field
name = redis.hget(f"user:{user_id}", "name")

# Increment counter
redis.hincrby(f"user:{user_id}", "visit_count", 1)

Leaderboard (frequently updated)

# Add scores (sorted set)
redis.zadd("leaderboard", {"user1": 100, "user2": 85, "user3": 90})

# Get top 10
top = redis.zrevrange("leaderboard", 0, 9, withscores=True)

# Rank of specific user
rank = redis.zrevrank("leaderboard", "user1")

Session (high-frequency, short-lived)

# Store entire session
redis.setex(f"session:{session_id}", 1800, json.dumps({
    "user_id": 42,
    "ip": "192.168.1.1",
    "created_at": time.time()
}))

# Check existence
if redis.exists(f"session:{session_id}"):
    print("Session valid")

Rate Limiting

# Allow 10 requests per minute
def rate_limit(user_id):
    key = f"rate:{user_id}"
    count = redis.incr(key)
    if count == 1:
        redis.expire(key, 60)
    return count <= 10

Monitoring Cache Health

# Cache hit ratio
def cache_stats(redis):
    info = redis.info("stats")
    hits = info["keyspace_hits"]
    misses = info["keyspace_misses"]
    total = hits + misses
    
    hit_rate = (hits / total) * 100 if total > 0 else 0
    print(f"Hit rate: {hit_rate:.1f}%")
    
    # Target: >95% for good performance
    return hit_rate

Best Practices

✅ Do	❌ Don't
Cache read-heavy data	Cache write-heavy data (cache thrashing)
Set appropriate TTL	Cache forever (stale data)
Monitor hit rates	Assume cache is working
Use versioning/tags	Delete keys without tracking
Handle cache misses	Fail when cache unavailable
Compress large values	Store raw large objects
Use pipelining	One operation per request

Checklist

• Choose pattern (cache-aside, write-through)
• Set TTL for all keys
• Plan invalidation strategy
• Handle thundering herd (locking)
• Monitor cache hit rates (target >95%)
• Compress large values
• Use appropriate data structures (hash, sorted set)
• Plan fallback for cache failures
• Load test before production
• Document cache keys (ttl, invalidation)

Conclusion

Redis caching is a multiplier: 14x faster responses, 90% less database load. Start with cache-aside. Graduate to write-through for consistency. Monitor hit rates religiously.

Most performance problems are solved by caching. Make it your first optimization.