引言
缓存是提升性能的银弹，但用不好会导致数据不一致、内存泄漏。本文从本地缓存到分布式缓存，讲解缓存模式、失效策略和一致性方案。
一、缓存模式
1.1 Cache-Aside（旁路缓存）
最常用：先查缓存，未命中查数据库，回写缓存。
python
复制
下载
def get_user(user_id): key = f"user:{user_id}" cached = redis.get(key) if cached: return json.loads(cached) user = db.query_user(user_id) if user: redis.setex(key, 3600, json.dumps(user)) return user1.2 Read-Through
缓存层负责加载，应用只与缓存交互。
1.3 Write-Through
写操作同时更新缓存和数据库。
1.4 Write-Behind
先写缓存，异步刷数据库。
二、本地缓存
2.1 functools.lru_cache
python
复制
下载
from functools import lru_cache @lru_cache(maxsize=128) def expensive_func(n): print(f"计算 {n}") return n ** 2 expensive_func(10) # 计算10 expensive_func(10) # 命中缓存带TTL的版本（Python 3.9+无原生支持）：
python
复制
下载
import time from functools import wraps def ttl_cache(seconds): def decorator(func): cache = {} @wraps(func) def wrapper(*args): now = time.time() if args in cache: result, expire = cache[args] if now < expire: return result result = func(*args) cache[args] = (result, now + seconds) return result return wrapper return decorator @ttl_cache(60) def get_config(key): return db.query_config(key)2.2 cachetools
bash
复制
下载
pip install cachetools
python
复制
下载
from cachetools import TTLCache, LRUCache, cached cache = TTLCache(maxsize=1000, ttl=300) @cached(cache) def get_user(user_id): return db.query_user(user_id)
2.3 多级缓存
python
复制
下载
class TwoLevelCache: def __init__(self, local, remote): self.local = local # TTLCache self.remote = remote # Redis def get(self, key): value = self.local.get(key) if value is not None: return value value = self.remote.get(key) if value is not None: self.local[key] = value return value def set(self, key, value, ttl=300): self.local[key] = value self.remote.setex(key, ttl, value)
三、Redis缓存
3.1 基础操作
python
复制
下载
import redis import json r = redis.Redis(host="localhost", port=6379, decode_responses=True) # 字符串 r.set("key", "value", ex=60) r.get("key") # JSON r.setex("user:1", 3600, json.dumps({"id": 1, "name": "Tom"})) # Hash r.hset("user:1", mapping={"id": 1, "name": "Tom"}) r.hgetall("user:1")3.2 缓存装饰器
python
复制
下载
import json import hashlib from functools import wraps def redis_cache(prefix, ttl=300): def decorator(func): @wraps(func) def wrapper(*args, **kwargs): key_data = f"{prefix}:{args}:{sorted(kwargs.items())}" key = hashlib.md5(key_data.encode()).hexdigest() cached = r.get(key) if cached: return json.loads(cached) result = func(*args, **kwargs) r.setex(key, ttl, json.dumps(result, default=str)) return result return wrapper return decorator @redis_cache("get_user", ttl=600) def get_user(user_id): return db.query_user(user_id)四、缓存三大问题
4.1 缓存穿透
查询不存在的数据，请求直达数据库。
python
复制
下载
NULL_VALUE = "__NULL__" def get_user(user_id): key = f"user:{user_id}" cached = r.get(key) if cached is not None: return None if cached == NULL_VALUE else json.loads(cached) user = db.query_user(user_id) if user is None: r.setex(key, 60, NULL_VALUE) # 缓存空值 return None r.setex(key, 3600, json.dumps(user)) return user布隆过滤器：
bash
复制
下载
pip install pybloom-live
python
复制
下载
from pybloom_live import BloomFilter bf = BloomFilter(capacity=1000000, error_rate=0.001) bf.add("user:1") if "user:1" in bf: # 可能存在，查缓存 pass else: # 一定不存在 return None4.2 缓存击穿
热点key过期瞬间，大量请求打数据库。
python
复制
下载
import threading _locks = {} def get_user_with_lock(user_id): key = f"user:{user_id}" cached = r.get(key) if cached: return json.loads(cached) lock = _locks.setdefault(key, threading.Lock()) with lock: # 双重检查 cached = r.get(key) if cached: return json.loads(cached) user = db.query_user(user_id) r.setex(key, 3600, json.dumps(user)) return user分布式锁版本用Redis：
python
复制
下载
lock_key = f"lock:{key}" if r.set(lock_key, "1", nx=True, ex=10): try: # 查数据库、回写缓存 pass finally: r.delete(lock_key)4.3 缓存雪崩
大量key同时过期。
python
复制
下载
import random def set_with_jitter(key, value, base_ttl=3600): ttl = base_ttl + random.randint(0, 600) r.setex(key, ttl, value)
其他措施：
多级缓存
限流降级
预热
五、缓存一致性
5.1 先更新数据库还是先删缓存
推荐：先更新数据库，再删除缓存（Cache-Aside）。
python
复制
下载
def update_user(user_id, data): db.update_user(user_id, data) r.delete(f"user:{user_id}")延迟双删：
python
复制
下载
def update_user(user_id, data): r.delete(f"user:{user_id}") db.update_user(user_id, data) time.sleep(0.5) # 延迟 r.delete(f"user:{user_id}")5.2 订阅binlog
用Canal订阅MySQL binlog，异步更新缓存，保证最终一致。
5.3 版本号
python
复制
下载
def get_user(user_id): version = r.get(f"user:version:{user_id}") or 0 key = f"user:{user_id}:{version}" # ...六、缓存淘汰策略
Redis支持：
noeviction：不淘汰，写入报错
allkeys-lru：所有key中LRU
allkeys-lfu：所有key中LFU
volatile-lru：过期key中LRU
volatile-ttl：优先淘汰TTL短的
bash
复制
下载
# redis.conf maxmemory 2gb maxmemory-policy allkeys-lru
七、实战：商品详情缓存
python
复制
下载
class ProductCache: def __init__(self, redis_client, db): self.redis = redis_client self.db = db self.local = TTLCache(maxsize=1000, ttl=60) def get(self, product_id): key = f"product:{product_id}" # L1：本地缓存 if key in self.local: return self.local[key] # L2：Redis cached = self.redis.get(key) if cached: data = json.loads(cached) self.local[key] = data return data # L3：数据库 product = self.db.query_product(product_id) if not product: return None data = product.to_dict() ttl = 3600 + random.randint(0, 300) self.redis.setex(key, ttl, json.dumps(data)) self.local[key] = data return data def invalidate(self, product_id): key = f"product:{product_id}" self.local.pop(key, None) self.redis.delete(key)八、总结
缓存核心：选对模式、防止三大问题、保证一致性。本地缓存快但有延迟，Redis统一但有网络开销，多级缓存兼顾。记住：缓存不是数据库，要能容忍丢失。
