// Cache-Aside 模式
def get_user(user_id):
# 1. 查缓存
user = cache.get(f'user:{user_id}')
if user:
return user # 命中，直接返回
# 2. 查数据库
user = db.query(f'SELECT * FROM users WHERE id = {user_id}')
# 3. 写入缓存
cache.set(f'user:{user_id}', user, ttl=600)
return user
def update_user(user_id, data):
# 1. 更新数据库
db.update('users', data)
# 2. 删除缓存（不是更新！）
cache.delete(f'user:{user_id}')// Read-Through 模式（代码更简洁）
def get_user(user_id):
# 缓存库自动处理数据库查询
user = cache.get_or_load(user_id, lambda: db.get_user(user_id))
return user
// Write-Through 模式
def update_user(user_id, data):
# 缓存库自动同步到数据库
cache.set(user_id, data) # 自动写入数据库// Write-Behind 模式
def update_counter(post_id):
# 1. 立即更新缓存（极快）
cache.incr(f'views:{post_id}')
# 立即返回，不等待数据库
# 2. 后台异步批量写入数据库
async def flush_to_db():
while True:
await asyncio.sleep(5) # 每5秒批量写入
batch = cache.get_many('views:*')
db.batch_update(batch)
asyncio.create_task(flush_to_db())| 模式 | 复杂度 | 性能 | 一致性 | 适用场景 |
|---|---|---|---|---|
| Cache-Aside | 中 | 高 | 中 | 大多数场景 |
| Read-Through | 低 | 中 | 高 | 简单场景 |
| Write-Behind | 高 | 极高 | 低 | 写多、可丢失 |