Caching Strategies for High Performance Systems - AI Agent Tools

← Back to Guides Caching Strategies for High Performance Systems Caching is the most effective way to improve application performance. A well-designed caching strategy can reduce latency by 100x and database load by 90%. This guide covers production caching patterns. Cache-Aside (Lazy Loading) The application checks the cache first. On a miss, it reads from the database, writes to the cache, and returns the result. import redis import json from functools import wraps class CacheAside: def __init__(self, redis_client, ttl=300): self.redis = redis_client self.ttl = ttl def get_or_set(self, key, loader_func): cached = self.redis.get(key) if cached: return json.loads(cached) value = loader_func() self.redis.setex(key, self.ttl, json.dumps(value)) return value def invalidate(self, key): self.redis.delete(key) # Usage cache = CacheAside(redis_client, ttl=600) user = cache.get_or_set( f"user:{{user_id}}", lambda: db.get_user(user_id) ) Read-Through Cache The cache provider transparently loads from the database on a cache miss. The application only talks to the cache. class ReadThroughCache: def __init__(self, cache, db_loader): self.cache = cache self.db_loader = db_loader def get(self, key): value = self.cache.get(key) if value is None: value = self.db_loader(key) self.cache.set(key, value) return value Write-Through Cache Every write goes to both the cache and the database synchronously. The cache is always consistent with the database. class WriteThroughCache: def __init__(self, cache, db): self.cache = cache self.db = db def write(self, key, value): self.db.write(key, value) self.cache.set(key, value) Write-Behind (Write-Back) Cache Writes go to the cache only. The cache asynchronously writes to the database after a delay. Much faster writes but risks data loss on failure. import threading import queue class WriteBehindCache: def __init__(self, cache, db, flush_interval=5): self.cache = cache self.db = db self.dirty = queue.Queue() self.flush_interval = flush_interval self._start_flusher() def write(self, key, value): self.cache.set(key, value) self.dirty.put((key, value)) def _start_flusher(self): def flusher(): while True: key, value = self.dirty.get() self.db.write(key, value) threading.Thread(target=flusher, daemon=True).start() Cache Invalidation The hardest problem in caching is knowing when to invalidate. Strategies: TTL expiration : Set a time-to-live. Simple but stale data between TTL and next read. Event-based invalidation : Invalidate when the underlying data changes. Most accurate but requires coordination. Versioned keys : Include a version number in the cache key. When data changes, bump the version. Lazy invalidation : Accept some staleness. Good for data that changes infrequently. # Event-based invalidation @dataclass class CacheInvalidator: cache: redis.Redis def on_user_updated(self, event): self.cache.delete(f"user:{{event.user_id}}") # Also delete collection caches self.cache.delete("users:all") # Delete search result caches for key in self.cache.scan_iter("users:search:*"): self.cache.delete(key) Cache Stampede Prevention When a popular cache entry expires, many requests simultaneously miss the cache and hit the database. Prevent this with request coalescing: import asyncio import time class StampedeGuard: def __init__(self, cache, ttl=300): self.cache = cache self.ttl = ttl self._locks = {} async def get_or_set(self, key, loader): cached = self.cache.get(key) if cached is not None: return cached # If another request is already loading, wait for it if key in self._locks: return await self._locks[key] # We're the first — load it future = asyncio.get_event_loop().create_future() self._locks[key] = future try: value = await loader() self.cache.setex(key, self.ttl, value) future.set_result(value) return value except Exception as e: future.set_exception(e) raise finally: del self._locks[key] # Alternative: probabilistic early expiration def jittered_ttl(base_ttl, current_age): # Expire early with some probability to spread out cache misses exp = base_ttl * (0.8 + 0.2 * random.random()) return int(exp) Multi-Level Caching Production systems use multiple cache layers with different characteristics: L1: In-process cache (LRU, ~1ms) — Fastest, limited by process memory L2: Redis/Memcached (~2-5ms) — Shared, persistent across processes L3: CDN edge cache (~10-50ms) — Closest to users, for static content Database (~5-50ms) — Source of truth from functools import lru_cache import redis class MultiLevelCache: def __init__(self, redis_client): self.redis = redis_client self._l1 = {} def get(self, key, loader): # L1: in-process if key in self._l1: return self._l1[key] # L2: Redis cached = self.redis.get(key) if cached: self._l1[key] = cached return cached # Load from source value = loader() self.redis.setex(key, 300, value) self._l1[key] = value return value Related Products Browse all products →