Caching with Redis

Probabilistic early expiration: Refresh the cache a few seconds before TTL expires, but only on a fraction of requests. Alternatively, use a lease lock: the first request that gets a cache miss acquires a lock, fetches from DB, and updates the cache. Other requests wait or serve slightly stale data.

Cache: User profiles, product catalogs, configuration, reference data, computed aggregates, and frequently-read query results.

Don't cache: Rapidly changing data (stock prices, real-time analytics), large blobs (videos, images), or data where consistency is critical and the DB can handle the load.

Use Redis Pub/Sub or a cache versioning prefix. When data changes, publish an invalidation message to a Redis channel. All app instances subscribe to the channel and clear their local or remote caches. Alternatively, change a version prefix (v1 → v2) in your cache keys to silently invalidate old entries without explicit messaging.

import json
import redis

r = redis.Redis(host="localhost", port=6379, decode_responses=True)

def write_through_user(user_id: int, data: dict, db_update_fn):
    """Write to cache and database atomically."""
    cache_key = f"user:{user_id}"

    # Update database first
    db_update_fn(user_id, data)

    # Then update cache
    r.setex(cache_key, 600, json.dumps(data))

# Usage
def db_update_user(user_id, data):
    # Execute SQL UPDATE
    pass

write_through_user(42, {"id": 42, "name": "Alice", "email": "alice@new.com"}, db_update_user)

async function writeThroughProduct(productId, data, dbUpdateFn) {
  const cacheKey = `product:${productId}`;

  // Update database
  await dbUpdateFn(productId, data);

  // Update cache
  await redis.setex(cacheKey, 300, JSON.stringify(data));
}

import json
import redis
import threading

r = redis.Redis(host="localhost", port=6379, decode_responses=True)

def write_behind_update(entity_type: str, entity_id: int, data: dict):
    """Write to cache immediately, queue for async DB write."""
    cache_key = f"{entity_type}:{entity_id}"

    # Write to cache
    r.setex(cache_key, 300, json.dumps(data))

    # Queue for async persistence
    r.lpush("pending_writes", json.dumps({
        "type": entity_type,
        "id": entity_id,
        "data": data,
        "timestamp": int(time.time())
    }))

# Background worker that flushes pending writes
def flush_pending_writes(db_write_fn, batch_size=100):
    while True:
        items = r.rpop("pending_writes", batch_size)
        if not items:
            threading.Event().wait(1)
            continue

        for item in items:
            entry = json.loads(item)
            try:
                db_write_fn(entry["type"], entry["id"], entry["data"])
            except Exception as e:
                # Re-queue failed writes
                r.lpush("pending_writes", json.dumps(entry))
                print(f"Write failed: {e}")

import json
import redis
import time

r = redis.Redis(host="localhost", port=6379, decode_responses=True)

def get_with_stampede_protection(cache_key: str, fetcher, ttl: int = 300):
    """Prevent cache stampede using a Redis lock."""
    cached = r.get(cache_key)
    if cached:
        return json.loads(cached)

    lock_key = f"lock:{cache_key}"
    # Try to acquire lock with NX + expiry
    acquired = r.set(lock_key, "1", ex=10, nx=True)

    if acquired:
        try:
            result = fetcher()
            r.setex(cache_key, ttl, json.dumps(result))
            return result
        finally:
            r.delete(lock_key)
    else:
        # Wait briefly and retry
        time.sleep(0.1)
        return get_with_stampede_protection(cache_key, fetcher, ttl)

# Probabilistic early expiration (reduces stampede without locks)
def get_with_early_refresh(cache_key: str, fetcher, ttl: int = 300, early_refresh_pct: float = 0.1):
    cached = r.get(cache_key)
    if cached:
        ttl_remaining = r.ttl(cache_key)
        # Refresh early with small probability
        if ttl_remaining < ttl * early_refresh_pct and random.random() < 0.1:
            try:
                result = fetcher()
                r.setex(cache_key, ttl, json.dumps(result))
                return result
            except Exception:
                pass  # Serve stale on refresh failure
        return json.loads(cached)

    # Cache miss
    result = fetcher()
    r.setex(cache_key, ttl, json.dumps(result))
    return result

const Redis = require("ioredis");
const pub = new Redis();
const sub = new Redis();

// Subscribe to invalidation channel
sub.subscribe("cache:invalidate");
sub.on("message", (channel, message) => {
  const { key } = JSON.parse(message);
  redis.del(key);
  console.log(`Invalidated: ${key}`);
});

// Publish invalidation on data update
async function updateUser(userId, data) {
  await db.query("UPDATE users SET ... WHERE id = ?", [userId]);
  const cacheKey = `user:${userId}`;
  await redis.del(cacheKey);
  await pub.publish("cache:invalidate", JSON.stringify({ key: cacheKey }));
}

import redis
import json

r = redis.Redis(host="localhost", port=6379, decode_responses=True)
pubsub = r.pubsub()

# Subscriber
pubsub.subscribe("cache:invalidate")
for message in pubsub.listen():
    if message["type"] == "message":
        data = json.loads(message["data"])
        r.delete(data["key"])

# Publisher
def invalidate_cache(key: str):
    r.delete(key)
    r.publish("cache:invalidate", json.dumps({"key": key}))

import redis

r = redis.Redis(host="localhost", port=6379)

# Atomic rate limiter using Lua
RATE_LIMIT_SCRIPT = """
local current = redis.call('INCR', KEYS[1])
if current == 1 then
    redis.call('EXPIRE', KEYS[1], ARGV[1])
end
return current
"""

rate_limiter = r.register_script(RATE_LIMIT_SCRIPT)

def is_rate_limited(key: str, max_requests: int, window: int) -> bool:
    current = rate_limiter(keys=[key], args=[window])
    return current > max_requests

# Atomic compare-and-swap for cache
CAS_SCRIPT = """
local cached = redis.call('GET', KEYS[1])
if cached == ARGV[1] then
    redis.call('SET', KEYS[1], ARGV[2], 'EX', ARGV[3])
    return 1
end
return 0
"""

cas = r.register_script(CAS_SCRIPT)

from redis.sentinel import Sentinel

sentinel = Sentinel([
    ("sentinel1", 26379),
    ("sentinel2", 26379),
    ("sentinel3", 26379),
])

# Get master connection
master = sentinel.master_for("mymaster", socket_timeout=0.5)

# Get replica connection for reads
replica = sentinel.slave_for("mymaster", socket_timeout=0.5)

# Writes go to master
master.set("key", "value")

# Reads can go to replica
value = replica.get("key")

const Redis = require("ioredis");

// Sentinel-aware client
const redis = new Redis({
  sentinels: [
    { host: "sentinel1", port: 26379 },
    { host: "sentinel2", port: 26379 },
  ],
  name: "mymaster",
  role: "master",
});