Python后端中间件专题03：详情页为什么越热门越慢——Redis cache-aside 从零接入

Python后端中间件专题03：详情页为什么越热门越慢——Redis cache-aside 从零接入

一张 P1 故障工单被几十个人反复刷新时，每次 GET 都读 PostgreSQL 并不是正确的“保守”：它会把可重用的读压力堆在事实库上。本课只改造详情读路径，用 cache-aside 完成 miss→DB→set 和 hit→直接返回。

本课只观察一个可重复的现象：同一详情连读两次，底层只回源一次。为此会把 cache-aside 放在 service 而不是 repository，保存带租户、版本和时间戳的 JSON 副本；这个证据证明加速，并不改变 PostgreSQL 的事实地位。

先带着第 02 篇的 HTTP/幂等基线，理解 JSON 序列化和装饰器式 service 组合。进入 project/ 后，本篇 selector 使用内存 cache fake；它不要求本机 Redis，也不证明 Redis TTL 或网络协议。

上一课练习答案

在加速之前，复核写入根本没有变。

答案 EX-02-01 [code]

以下代码是可运行的 TestClient + SQLite 实验。它不只比较 HTTP 返回值，还直接检查事务结果：

from fastapi.testclient import TestClient
from sqlalchemy import create_engine, func, select
from sqlalchemy.orm import sessionmaker
from sqlalchemy.pool import StaticPool

from ticketflow.api.app import create_app
from ticketflow.domain.tickets import TicketService
from ticketflow.storage.models import Base, OutboxEventModel, TicketModel
from ticketflow.storage.repository import TicketRepository

engine = create_engine(
"sqlite+pysqlite:///:memory:",
connect_args={"check_same_thread": False},
poolclass=StaticPool,
)
Base.metadata.create_all(engine)
sessions = sessionmaker(bind=engine, expire_on_commit=False)
client = TestClient(create_app(TicketService(TicketRepository(sessions))))
headers = {
"X-Tenant-Id": "acme",
"X-Actor-Id": "agent-7",
"Idempotency-Key": "request-001",
}
request_a = {"title": "VPN unavailable", "body": "Cannot connect", "priority": "high"}
request_b = {**request_a, "body": "A different intent"}

first = client.post("/tickets", headers=headers, json=request_a)
replay = client.post("/tickets", headers=headers, json=request_a)
conflict = client.post("/tickets", headers=headers, json=request_b)

assert [first.status_code, replay.status_code, conflict.status_code] == [201, 201, 409]
assert first.json()["id"] == replay.json()["id"]
with sessions() as session:
ticket_count = session.scalar(select(func.count()).select_from(TicketModel))
outbox_count = session.scalar(select(func.count()).select_from(OutboxEventModel))
assert (ticket_count, outbox_count) == (1, 1)
print(first.status_code, replay.status_code, conflict.status_code, ticket_count, outbox_count)

运行命令：python idempotency_probe.py

预期输出：

201 201 409 1 1

同键同体返回 201 是本项目的明确契约：它重放原始创建结果，没有创建第二个 Outbox 事件。同键异体则不可被伪装成重放，因此是 409。

答案 EX-02-02 [prose]

把缓存只加在 get 之后，三次 POST 结果必须保持 201、201、409，数据库行数仍是 1/1。读缓存不应参与写幂等裁决。“连读两次，repository 只读一次”只证明第二次从副本命中；当 Redis 中的 ID 不存在于 PostgreSQL 时，它仍然没有创建或修改业务事实的权力。

先观察：第二次读取为什么没有资格查库

检查点给底层 service 加了 get_count，连续读两次同一 (tenant_id, ticket_id)，断言返回的 Ticket 相同，但底层读取计数是 1。

运行命令：python -m pytest tests/unit/test_cached_ticket_service.py::test_cached_service_reads_database_once_and_rehydrates_ticket -q

本地 Python 3.11 实际输出：

. [100%]
1 passed

这个本地测试只证明装饰服务控制流。可信远端 selector 会清除唯一 key，首次 GET 后检查真实 Redis 的签名 envelope 与 PTTL，再停 PostgreSQL执行第二次 GET；只有第二次仍返回同一版本，才构成可观察的、无需查库的认证 Redis hit 证据。未运行仍为 PENDING。

由读数倒推 service 边界

要让“第二次不查库”仍不越过事实边界，我们用 CachedTicketService 装饰原有 TicketService，而不是把 Redis 逻辑插进 repository。repository 继续只处理数据库事务，因此缓存宕机不会改变 SQL 提交语义；API 面向的 service 接口也保持不变，可以单独检验命中、回源和故障。

缓存 key 必须同时包含租户与工单 ID。如果 key 只是 ticket:{id}，即使 UUID 冲突概率很低，代码也丧失了“租户隔离在每层都显式”的可审计性。真正的 Redis adapter 会 URL-escape key 片段，避免 tenant:a + ticket 和 tenant + a:ticket 一类组合碰撞；payload 与短期负标记均用部署密钥做 HMAC。这样正常 hit 不必额外查库，Redis 故障仍可回源，而未知 JSON 只能当作 miss，不能当作业务事实。密钥经 TICKETFLOW_CACHE_SIGNING_KEY 在 API 与 Worker 间一致配置；轮换令旧缓存自然退化为 miss。

副本还要包含 version、created_at 和 updated_at。version 有助于后续发现陈旧副本，时间戳帮助解释它何时生成；两者都不自动提供强一致。loader 始终调用原 service，只把严格 JSON 数据交给缓存，命中后 _deserialize 重建领域 Ticket。因此不能把 ORM 实例塞进 Redis、不能丢掉 tenant key、也不能只看两次响应相同就声称命中——本实验的读取计数正是在排除最后一种误读。

本课练习

为下一次写入准备两个问题。

练习 EX-03-01 [code]

写两个缓存 payload：一个是 version=1 的旧标题，一个是 version=2 的新标题。实现 choose_newer(cached, database)，断言新版本被选中；同时打印 updated_at 用于排障，并在注释中解释为什么光有版本与时间戳仍不是强一致。

练习 EX-03-02 [prose]

画出 PATCH 已提交、删缓存、与一个更早启动的 GET 回填旧值之间的顺序，说明为什么“先更新 DB，再删缓存”是必要基线，却仍然留有一个并发窗口。

第 04 篇会回答 EX-03-01/02，并把更新路径改成“数据库提交成功后删除缓存”。不要期待它承诺强一致：我们会刻意重演旧 GET 在删除之后回填 v1 的残余竞态。

完整核心模块：缓存装饰服务

from __future__ import annotations

from datetime import datetime
from typing import Protocol

from .tickets import Ticket, TicketService

class TicketCache(Protocol):
def get_or_load_ticket(self, tenant_id: str, ticket_id: str, loader): ...

def _serialize(ticket: Ticket) –> dict[str, object]:
"""Create strict JSON data; the cache never receives Python-only values."""

return {
"id": ticket.id,
"tenant_id": ticket.tenant_id,
"actor_id": ticket.actor_id,
"title": ticket.title,
"body": ticket.body,
"priority": ticket.priority,
"status": ticket.status,
"version": ticket.version,
"created_at": ticket.created_at.isoformat(),
"updated_at": ticket.updated_at.isoformat(),
}

def _deserialize(payload: dict[str, object]) –> Ticket:
return Ticket(
id=str(payload["id"]),
tenant_id=str(payload["tenant_id"]),
actor_id=str(payload["actor_id"]),
title=str(payload["title"]),
body=str(payload["body"]),
priority=str(payload["priority"]),
status=str(payload["status"]),
version=int(payload["version"]),
created_at=datetime.fromisoformat(str(payload["created_at"])),
updated_at=datetime.fromisoformat(str(payload["updated_at"])),
)

class CachedTicketService:
"""Decorate detail reads with cache-aside while writes keep DB authority."""

def __init__(self, service: TicketService, cache: TicketCache) –> None:
self._service = service
self._cache = cache

def create(self, *args): # type: ignore[no-untyped-def]
return self._service.create(*args)

def get(self, tenant_id: str, ticket_id: str) –> Ticket | None:
def load() –> dict[str, object] | None:
ticket = self._service.get(tenant_id, ticket_id)
return _serialize(ticket) if ticket is not None else None

payload = self._cache.get_or_load_ticket(tenant_id, ticket_id, load)
return _deserialize(payload) if payload is not None else None

def update(
self,
tenant_id: str,
ticket_id: str,
title: str | None = None,
body: str | None = None,
priority: str | None = None,
) –> Ticket | None:
return self._service.update(
tenant_id, ticket_id, title=title, body=body, priority=priority
)

one,
) -> Ticket | None:
return self._service.update(
tenant_id, ticket_id, title=title, body=body, priority=priority
)