缓存协作与 Redis 旁路 · Python 数据库实战

第 15 章 · 缓存协作与 Redis 旁路

本章目标：实现 Cache-Aside（旁路缓存） 读写流程；处理 Redis 与 MySQL 一致性 常见问题；理解 延迟双删 适用场景与局限；在 db-demo 用 redis-py + SQLAlchemy 完成商品 slug 缓存；失效策略与 django-web ch10 缓存章衔接；保证缓存 JSON 含 slug、is_published、price（分） 与 DB 一致。

学时建议：4～5 小时（含 1.5 小时 Redis 跟练）

前置：ch13～ch14；django-web ch10 缓存与 Celery；Redis 本地 Docker 可用。

15.1 为什么需要缓存

场景	直查 MySQL	Cache-Aside
商品 slug 详情	每次索引查询 + 磁盘	命中 Redis 亚毫秒
已发布列表	重复 range 扫描	短 TTL 列表缓存
热点 slug 秒杀	DB 连接打满	缓存 + 单飞（选修）

读请求 GET /products/{slug}
        │
        ▼
   cache.get(key)
        │
   ┌────┴────┐
   命中      未命中
   │          │
   │          ▼
   │     SELECT MySQL
   │          │
   │          ▼
   │     cache.set(key, TTL)
   │          │
   └────► JSON 响应（price 分）

shop-db / db-demo 不连接真实生产 Redis；地址示例 redis://127.0.0.1:6379/2。

15.2 Cache-Aside 标准流程

15.2.1 读路径

import json
import redis
from sqlalchemy import select
from db.models import Product
from db.session import get_write_session

r = redis.Redis.from_url("redis://127.0.0.1:6379/2", decode_responses=True)

PRODUCT_KEY = "product:slug:{slug}"  # product:slug:python-handbook
TTL_SEC = 300


def get_product_by_slug(slug: str) -> dict | None:
    key = PRODUCT_KEY.format(slug=slug)
    cached = r.get(key)
    if cached:
        return json.loads(cached)

    session = get_write_session()
    try:
        product = session.scalar(
            select(Product).where(Product.slug == slug, Product.is_published.is_(True))
        )
        if not product:
            return None
        data = {
            "slug": product.slug,
            "name": product.name,
            "price": product.price,  # 分，int
            "is_published": product.is_published,
        }
        r.setex(key, TTL_SEC, json.dumps(data))
        return data
    finally:
        session.close()

15.2.2 写路径（更新 DB 后删缓存）

def update_product_price(slug: str, price_cents: int) -> None:
    session = get_write_session()
    try:
        product = session.scalar(select(Product).where(Product.slug == slug))
        if not product:
            raise ValueError("not found")
        product.price = price_cents
        session.commit()
    finally:
        session.close()
    # 先更新库，再删缓存 — Cache-Aside 铁律
    r.delete(PRODUCT_KEY.format(slug=slug))
    r.delete("product_list:published:v1")  # 列表缓存一并失效

顺序	正确	错误
更新	先 DB commit，再 delete cache	先删缓存再 DB（并发脏读）
创建	commit 后 delete 列表 key	只删 slug 不删列表
字段	price 用分 int	float 元

15.3 与 django-web ch10 对照

能力	django-web ch10	本章 db-demo
后端	django-redis	redis-py
读	`cache.get` / `get_or_set`	手写 get/setex
写失效	`post_save` signal `cache.delete`	显式 delete
列表 key	`product_list:published:v1`	同名约定
原则	先更新库再删缓存	相同

django-web 示例（复习）：

# catalog/services.py — django-web ch10
def get_published_products():
    data = cache.get(PRODUCT_LIST_KEY)
    if data is not None:
        return data
    qs = list(Product.objects.filter(is_published=True).values(...)[:50])
    cache.set(PRODUCT_LIST_KEY, qs, timeout=120)
    return qs

衔接练习：读完 django-web ch10 §10.4～10.5 后，将等价逻辑用 SQLAlchemy + redis-py 在 db-demo 实现一遍。

15.4 Redis 与 DB 一致性问题

问题	描述	本章策略
缓存穿透	查不存在 slug，打穿 DB	空值缓存短 TTL
缓存击穿	热点 key 过期瞬间并发	互斥锁 / 单飞
缓存雪崩	大量 key 同时过期	TTL 加随机 jitter
双写不一致	删缓存失败	重试 + 监控；延迟双删
脏读	先删缓存后写库	禁止；严格先写库

15.4.1 空值缓存

def get_product_by_slug(slug: str) -> dict | None:
    key = PRODUCT_KEY.format(slug=slug)
    cached = r.get(key)
    if cached == "__NULL__":
        return None
    if cached:
        return json.loads(cached)
    # ... 查 DB ...
    if not product:
        r.setex(key, 60, "__NULL__")  # 短 TTL 防穿透
        return None

15.4.2 TTL jitter

import random
ttl = TTL_SEC + random.randint(0, 30)
r.setex(key, ttl, json.dumps(data))

15.5 延迟双删（Delayed Double Delete）

场景：并发读写导致旧数据回填缓存。

线程 A 更新 DB ──► 删缓存
线程 B 读未命中 ──► 读旧 DB？──► 写回旧缓存
线程 A 延迟再删 ──► 清掉 B 写的脏缓存

15.5.1 流程

更新数据库
删除缓存（第一次）
延迟 N ms（通常 300～500）
再删一次 同一 key

import threading

def invalidate_with_delayed_double_delete(slug: str, delay_ms: int = 500):
    key = PRODUCT_KEY.format(slug=slug)
    r.delete(key)

    def _delayed():
        import time
        time.sleep(delay_ms / 1000)
        r.delete(key)

    threading.Thread(target=_delayed, daemon=True).start()

15.5.2 局限与替代

方案	优点	缺点
延迟双删	实现简单	延迟难估；仍可能极短脏读
先写库再删缓存	主流 Cache-Aside	删失败需补偿
订阅 binlog 删缓存	强一致倾向	组件复杂（Canal 等）
读写都走 MQ 串行	一致性强	延迟高

教学结论：shop-db 采用 先写库 + 删缓存 + 列表 key 失效；延迟双删写进 docs/CACHE.md 作选修。

15.6 列表缓存与详情缓存

Key 模式	内容	TTL	失效时机
`product:slug:{slug}`	单商品 JSON	300s	该商品 update/delete
`product_list:published:v1`	id/slug/price 数组	120s	任意商品上下架

LIST_KEY = "product_list:published:v1"

def get_published_list(session) -> list[dict]:
    cached = r.get(LIST_KEY)
    if cached:
        return json.loads(cached)
    rows = session.execute(
        select(Product.slug, Product.name, Product.price, Product.is_published)
        .where(Product.is_published.is_(True))
        .order_by(Product.id.desc())
        .limit(50)
    ).all()
    data = [
        {"slug": s, "name": n, "price": p, "is_published": bool(pub)}
        for s, n, p, pub in rows
    ]
    r.setex(LIST_KEY, 120, json.dumps(data))
    return data

price 分：列表缓存中 price 仍为整数分，前端除以 100 展示。

15.7 本地 Redis 环境

db-demo/docker-compose.dev.yml：

services:
  redis:
    image: redis:7-alpine
    ports:
      - "6379:6379"
    volumes:
      - redis_data:/data

volumes:
  redis_data:

docker compose -f docker-compose.dev.yml up -d redis
redis-cli -n 2 ping   # PONG

依赖：

pip install redis sqlalchemy pymysql