← 返回 SaaS 知识库

缓存策略

Redis / CDN / Stale-while-revalidate / Edge Cache — 从浏览器到数据库的完整缓存链路设计

📋 目录

  1. 为什么缓存是 SaaS 的生命线
  2. 缓存层级全景:从浏览器到数据库
  3. 核心缓存模式详解
  4. Redis 深度实战
  5. CDN 缓存策略
  6. HTTP 缓存头完全指南
  7. Stale-While-Revalidate 模式
  8. Edge Cache 与 Edge Computing
  9. 应用层缓存模式
  10. 缓存失效:计算机科学两大难题之一
  11. LLM/AI 场景的缓存策略
  12. 决策树:我该用什么缓存?
  13. 技术选型对比
  14. 常见反模式与坑
  15. 缓存架构检查清单

1. 为什么缓存是 SaaS 的生命线

缓存不是"锦上添花",而是 SaaS 产品从"能用"到"好用"的分水岭。一个没有缓存的系统,用户每次点击都打到数据库——当用户量从 100 涨到 10000,数据库直接崩溃。

100x
缓存可降低的数据库负载
<50ms
Redis 缓存命中延迟
5-50ms
CDN Edge 缓存延迟
70%
典型 SaaS 请求可缓存比例

🎯 缓存解决的核心问题

2. 缓存层级全景:从浏览器到数据库

一个完整的 SaaS 系统有 5 层缓存,每层的角色、失效策略、存储介质完全不同:

用户请求 → ① 浏览器缓存 (HTTP Cache / Service Worker) ↓ miss ② CDN 边缘缓存 (Cloudflare / Vercel Edge) ↓ miss ③ 应用层缓存 (Redis / Memcached / In-Memory) ↓ miss ④ 数据库查询缓存 (PostgreSQL cache / ORM L2 cache) ↓ miss ⑤ 源数据 (PostgreSQL / S3 / 外部 API)

浏览器缓存

Cache-Control / ETag / Service Worker

0ms 免费

最廉价、最快、但最难控制失效

CDN 边缘缓存

Cloudflare / Vercel / CloudFront

5-50ms 按流量计费

地理位置就近、静态资源和 API 响应都能缓存

应用层缓存

Redis / Memcached / 进程内 LRU

1-5ms 需要维护

最灵活、最常用、精细控制失效

数据库查询缓存

PostgreSQL shared_buffers / ORM L2

5-20ms 自动

DBA 领域、通常不需要手动管理

源数据

PostgreSQL / MySQL / S3 / API

10-500ms 最贵

最终真相、永远不能缓存错误数据

💡 黄金法则:越靠近用户的缓存,延迟越低、成本越低,但失效越难控制。好的缓存架构是在一致性性能之间找到正确的平衡点。

3. 核心缓存模式详解

3.1 Cache-Aside (旁路缓存)

最经典的缓存模式。应用程序负责从缓存读数据,miss 时从数据库加载并写入缓存。

读请求: GET cache:key → 命中? → 返回 ↓ miss SELECT FROM db → 写入 cache → 返回 写请求: UPDATE db → DELETE cache:key (或 SET cache:key = new_value)
# Python + Redis 实现 Cache-Aside
import redis
import json

r = redis.Redis(host='localhost', port=6379, decode_responses=True)

async def get_user(user_id: str) -> dict:
    # 1. 尝试从缓存读取
    cached = r.get(f"user:{user_id}")
    if cached:
        return json.loads(cached)
    
    # 2. 缓存 miss,查数据库
    user = await db.users.find_by_id(user_id)
    if user:
        # 3. 写入缓存,TTL 5 分钟
        r.setex(f"user:{user_id}", 300, json.dumps(user))
    return user

async def update_user(user_id: str, data: dict) -> dict:
    # 1. 先更新数据库
    user = await db.users.update(user_id, data)
    # 2. 删除缓存(而非更新),避免并发问题
    r.delete(f"user:{user_id}")
    return user

✅ 优势

  • 最简单、最安全
  • 缓存故障不阻塞(降级到数据库)
  • 适用绝大多数场景

❌ 劣势

  • 首次请求永远 miss(冷启动问题)
  • 短暂不一致窗口(DB 更新后、缓存删除前)
  • 高并发下可能"惊群"(缓存重建风暴)

3.2 Write-Through (写穿透)

写入时同时更新缓存和数据库,保证缓存始终最新。

# Write-Through 实现
async def save_user(user_id: str, data: dict) -> dict:
    user = await db.users.update(user_id, data)
    # 同步写入缓存(而非删除)
    r.setex(f"user:{user_id}", 300, json.dumps(user))
    return user

✅ 优势

  • 数据一致性最好
  • 读请求永远命中(无 miss)

❌ 劣势

  • 写操作延迟增加(双写)
  • 写多读少的场景浪费
  • 缓存可能存入永不被读取的数据

3.3 Write-Behind (异步写回)

先写缓存,异步批量写入数据库。适合写密集场景。

# Write-Behind:先写缓存,异步持久化
import asyncio

write_queue = asyncio.Queue()

async def save_user_async(user_id: str, data: dict) -> dict:
    # 1. 立即写入缓存(用户感知到即时更新)
    user = {**data, "id": user_id, "_dirty": True}
    r.setex(f"user:{user_id}", 300, json.dumps(user))
    
    # 2. 放入写队列,异步持久化
    await write_queue.put(("update", user_id, data))
    return user

async def write_behind_worker():
    """后台 worker:批量写入数据库"""
    while True:
        batch = []
        # 收集最多 100 条或等 1 秒
        try:
            item = await asyncio.wait_for(write_queue.get(), timeout=1.0)
            batch.append(item)
            while len(batch) < 100 and not write_queue.empty():
                batch.append(write_queue.get_nowait())
        except asyncio.TimeoutError:
            pass
        
        if batch:
            await db.users.bulk_update(batch)  # 批量写入
⚠️ Write-Behind 风险:如果缓存服务器在异步写入前崩溃,数据会丢失。只适用于可容忍少量丢失的场景(如浏览量计数器),不适用于金融交易。

3.4 Refresh-Ahead (预刷新)

在缓存过期前主动刷新,用户永远不感知 miss。

# Refresh-Ahead:过期前主动刷新
import time

async def get_with_refresh_ahead(key: str, ttl: int = 300, refresh_before: int = 60):
    """TTL 300 秒,过期前 60 秒主动刷新"""
    data = r.get(key)
    
    if data is None:
        # 完全 miss,同步加载
        result = await load_from_db(key)
        r.setex(key, ttl, json.dumps(result))
        return result
    
    # 检查剩余 TTL
    remaining = r.ttl(key)
    if remaining <= refresh_before:
        # 快过期了,返回旧数据 + 异步刷新
        result = json.loads(data)
        asyncio.create_task(refresh_cache(key, ttl))
        return result
    
    return json.loads(data)

async def refresh_cache(key: str, ttl: int):
    result = await load_from_db(key)
    r.setex(key, ttl, json.dumps(result))

模式选择速查

模式一致性读延迟写延迟适用场景
Cache-Aside最终一致miss 时高通用场景(默认选择)
Write-Through强一致极低读多写少、一致性优先
Write-Behind弱一致极低极低写密集、可容忍丢失
Refresh-Ahead最终一致极低热数据、零 miss 要求

4. Redis 深度实战

Redis 是 SaaS 缓存的事实标准。但会用 SET/GET 和用好 Redis 是两个完全不同的层次。

4.1 Redis 数据结构选择

结构场景内存效率示例
String简单 KV、计数器、JSON 对象⭐⭐缓存用户 profile JSON
Hash对象属性、部分读写⭐⭐⭐⭐用户信息分字段存储
List时间线、消息队列⭐⭐⭐最近活动列表
Sorted Set排行榜、延迟队列⭐⭐⭐热门内容排行
Set去重、标签、共同好友⭐⭐⭐用户标签集合
Bitmap布尔状态、签到⭐⭐⭐⭐⭐30 天签到状态 (4 字节)
HyperLogLog基数统计(UV)⭐⭐⭐⭐⭐页面 UV 统计 (12KB)

4.2 Hash vs String:一个关键选择

# ❌ 低效:用 String 存 JSON(每次要读整个对象)
r.set("user:123", json.dumps({"name": "张三", "email": "z@test.com", "role": "admin"}))
# 更新 email 需要读→改→写,非原子

# ✅ 高效:用 Hash 存对象字段
r.hset("user:123", mapping={
    "name": "张三",
    "email": "z@test.com", 
    "role": "admin"
})
# 只更新 email,原子操作
r.hset("user:123", "email", "new@test.com")
# 只读需要的字段
email = r.hget("user:123", "email")

# 内存节省:Hash 在 field < 512 时使用 ziplist,内存减少 50%+

4.3 Redis 缓存实战模式

🔒 防惊群:分布式锁 + Singleflight

# 高并发下防止缓存击穿(大量请求同时重建缓存)
import asyncio
from redis.lock import Lock

# 方案 1:Redis 分布式锁
async def get_with_lock(key: str):
    cached = r.get(key)
    if cached:
        return json.loads(cached)
    
    # 只允许一个请求重建缓存
    lock_key = f"lock:{key}"
    lock = r.lock(lock_key, timeout=10)
    
    if lock.acquire(blocking=False):
        try:
            # 再次检查(double-check)
            cached = r.get(key)
            if cached:
                return json.loads(cached)
            
            result = await load_from_db(key)
            r.setex(key, 300, json.dumps(result))
            return result
        finally:
            lock.release()
    else:
        # 等待锁释放后重试
        await asyncio.sleep(0.1)
        return await get_with_lock(key)

# 方案 2:Python singleflight(更简单,进程内)
from asyncio import Lock as AsyncLock

_flight_locks: dict[str, AsyncLock] = {}
_flights: dict[str, asyncio.Event] = {}
_results: dict[str, any] = {}

async def singleflight_get(key: str):
    if key in _flights:
        await _flights[key].wait()
        return _results.get(key)
    
    _flights[key] = asyncio.Event()
    try:
        result = await load_from_db(key)
        r.setex(key, 300, json.dumps(result))
        _results[key] = result
        return result
    finally:
        _flights[key].set()
        del _flights[key]

🛡️ 防穿透:布隆过滤器 + 空值缓存

# 缓存穿透:查询不存在的数据,每次都打到数据库
# 解决方案 1:缓存空值
async def get_user_safe(user_id: str):
    cached = r.get(f"user:{user_id}")
    if cached is not None:
        if cached == "NULL":
            return None  # 缓存的空值
        return json.loads(cached)
    
    user = await db.users.find_by_id(user_id)
    if user:
        r.setex(f"user:{user_id}", 300, json.dumps(user))
    else:
        # 缓存空值,TTL 短一些(60 秒)
        r.setex(f"user:{user_id}", 60, "NULL")
    return user

# 解决方案 2:Redis 布隆过滤器(更适合大量不存在的 key)
# 先安装 redisbloom 模块
async def get_user_bloom(user_id: str):
    # 布隆过滤器快速判断是否可能存在
    if not r.bf.exists("users_bloom", user_id):
        return None  # 100% 不存在
    
    cached = r.get(f"user:{user_id}")
    if cached:
        return json.loads(cached)
    
    user = await db.users.find_by_id(user_id)
    if user:
        r.setex(f"user:{user_id}", 300, json.dumps(user))
    return user

🧊 防雪崩:随机 TTL + 永不过期

# 缓存雪崩:大量 key 同时过期,瞬间全部打到数据库
import random

# 方案 1:TTL 加随机偏移
base_ttl = 300  # 5 分钟
jitter = random.randint(-60, 60)  # ±1 分钟随机
r.setex(key, base_ttl + jitter, value)

# 方案 2:逻辑过期(永不过期 + 异步刷新)
async def get_with_logical_expiry(key: str):
    data = r.get(key)
    if data is None:
        return await load_and_cache(key)
    
    obj = json.loads(data)
    # 检查逻辑过期时间
    if obj["_expire_at"] < time.time():
        # 过期了:返回旧数据 + 触发异步刷新
        asyncio.create_task(refresh_cache(key))
    return obj["data"]

def set_with_logical_expiry(key: str, value, ttl: int = 300):
    r.set(key, json.dumps({
        "data": value,
        "_expire_at": time.time() + ttl
    }))  # 注意:没有设置 Redis TTL,永不过期

4.4 Redis 内存优化

📉 内存优化技巧

# Pipeline 批量操作示例
pipe = r.pipeline()
for user_id in user_ids:
    pipe.hgetall(f"user:{user_id}")
results = pipe.execute()  # 一次 RTT 完成 N 次查询

# 批量写入
pipe = r.pipeline()
for key, value in cache_items:
    pipe.setex(key, 300, json.dumps(value))
pipe.execute()

5. CDN 缓存策略

CDN 缓存让你的静态资源和 API 响应从距离用户最近的节点返回,延迟从 200ms 降到 20ms。

5.1 CDN 缓存什么?

资源类型是否缓存TTL 建议失效方式
静态资源 (JS/CSS/图片)✅ 强缓存1 年(内容哈希文件名)文件名含 hash,无需手动失效
HTML 入口⚠️ 协商缓存短 TTL 或 no-cache部署时自动更新
公开 API (列表/搜索)✅ 边缘缓存30s - 5minPurge API
私有 API (用户数据)❌ 不缓存
SSR 页面⚠️ 有条件缓存1-60s (ISR)按需 revalidate

5.2 Next.js ISR:CDN + SSR 的最佳平衡

// Next.js App Router: ISR (Incremental Static Regeneration)
// 页面缓存 60 秒,之后后台重新生成
export const revalidate = 60; // 秒

export default async function ProductPage({ params }) {
  const product = await db.products.findById(params.id);
  return <ProductDetail product={product} />;
}

// 按需 revalidate(如后台编辑商品后)
import { revalidatePath } from 'next/cache';

async function updateProduct(formData) {
  await db.products.update(formData);
  revalidatePath(`/products/${formData.id}`); // 立即刷新
}

5.3 Cloudflare Workers:边缘缓存 API

// Cloudflare Worker: 边缘缓存 API 响应
export default {
  async fetch(request, env) {
    const cache = caches.default;
    const url = new URL(request.url);
    
    // 只缓存 GET 请求的公开 API
    if (request.method !== 'GET' || url.pathname.startsWith('/api/private')) {
      return fetch(request);
    }
    
    const cacheKey = new Request(url.toString(), { method: 'GET' });
    const cached = await cache.match(cacheKey);
    
    if (cached) {
      // 添加 Age 头,让客户端知道这是缓存
      const response = new Response(cached.body, cached);
      response.headers.set('X-Cache', 'HIT');
      return response;
    }
    
    // Miss: 请求源站
    const response = await fetch(request);
    
    if (response.status === 200) {
      // 缓存响应 60 秒
      const headers = new Headers(response.headers);
      headers.set('Cache-Control', 'public, max-age=60, s-maxage=60');
      const cachedResponse = new Response(response.body, { headers });
      
      // 异步写入缓存(不阻塞响应)
      await cache.put(cacheKey, cachedResponse.clone());
      return cachedResponse;
    }
    
    return response;
  }
};

6. HTTP 缓存头完全指南

HTTP 缓存头是浏览器和 CDN 缓存的基础。理解它们是设计缓存策略的前提。

6.1 核心头部

Cache-Control 指令详解

指令作用适用场景
max-age=3600缓存 3600 秒内有效不常变化的资源
s-maxage=60CDN 缓存 60 秒(覆盖 max-age)CDN 缓存短于浏览器
no-cache可以缓存,但每次使用前必须验证需要最新但仍想缓存
no-store完全不缓存敏感数据
private只浏览器缓存,CDN 不缓存用户私有数据
public允许 CDN 缓存公开资源
stale-while-revalidate=60过期后 60 秒内返回旧值+后台刷新SWR 模式
stale-if-error=300源站错误时,5 分钟内可返回旧缓存容灾降级
immutable过期前不会重新验证内容哈希 URL

6.2 常见场景的 Cache-Control 配置

# 1. 静态资源(JS/CSS/字体/图片)— 文件名含 hash
Cache-Control: public, max-age=31536000, immutable
# 一年缓存 + immutable = 浏览器连验证请求都不发

# 2. HTML 入口文件 — 需要及时更新
Cache-Control: no-cache
# 每次都验证,但 ETag/Last-Modified 未变时 304 不传 body

# 3. 公开 API(商品列表等)
Cache-Control: public, s-maxage=60, stale-while-revalidate=300
# CDN 缓存 60 秒,过期后 5 分钟内返回旧值+后台刷新

# 4. 用户私有 API
Cache-Control: private, no-cache
# 只浏览器缓存,每次验证

# 5. 完全不缓存(支付/实时数据)
Cache-Control: no-store, no-cache, must-revalidate

# 6. SSR 页面
Cache-Control: public, s-maxage=10, stale-while-revalidate=59
# 10 秒 CDN 缓存 + 59 秒 SWR

6.3 ETag vs Last-Modified

特性ETagLast-Modified
精度内容级别(哈希)秒级(1 秒内多次修改丢失)
计算成本需要哈希计算文件系统元数据
分布式一致性❌ 不同服务器 ETag 可能不同✅ 时间相对一致
推荐场景CDN 单源 / CDN 自己生成多源站分布式

7. Stale-While-Revalidate 模式

SWR 是缓存策略中"用户体验最好"的模式——先给旧数据(快),后台刷新新数据(准)。

7.1 HTTP 层 SWR

# HTTP 响应头
Cache-Control: max-age=60, stale-while-revalidate=300
# 含义:60 秒内直接用缓存;60-360 秒内返回旧缓存 + 后台验证

# 时间线:
# 0-60s:  直接返回缓存 (fresh)
# 60-360s: 返回旧缓存 + 后台重新验证 (stale)
# >360s: 缓存完全过期,等待新数据

7.2 React SWR 库

// SWR 库:React 数据获取 + 缓存
import useSWR from 'swr';

const fetcher = (url) => fetch(url).then(r => r.json());

function UserProfile({ userId }) {
  const { data, error, isLoading, isValidating } = useSWR(
    `/api/users/${userId}`,
    fetcher,
    {
      refreshInterval: 30000,    // 30 秒自动刷新
      revalidateOnFocus: true,   // 窗口聚焦时刷新
      revalidateOnReconnect: true, // 网络恢复时刷新
      dedupingInterval: 2000,    // 2 秒内相同 key 去重
      suspense: true,            // Suspense 模式
    }
  );

  if (error) return <div>加载失败</div>;
  if (!data) return <div>加载中...</div>;

  return (
    <div>
      {isValidating && <span className="text-sm text-gray-400">更新中...</span>}
      <h2>{data.name}</h2>
      <p>{data.email}</p>
    </div>
  );
}

7.3 TanStack Query(更强大的 SWR 替代)

// TanStack Query v5:企业级数据缓存
import { useQuery, useMutation, useQueryClient } from '@tanstack/react-query';

function useUser(userId: string) {
  return useQuery({
    queryKey: ['user', userId],
    queryFn: () => fetch(`/api/users/${userId}`).then(r => r.json()),
    staleTime: 5 * 60 * 1000,   // 5 分钟内不重新获取
    gcTime: 30 * 60 * 1000,      // 30 分钟后垃圾回收(原 cacheTime)
    refetchOnWindowFocus: true,
  });
}

function useUpdateUser() {
  const queryClient = useQueryClient();
  
  return useMutation({
    mutationFn: (data) => fetch('/api/users', { 
      method: 'PUT', 
      body: JSON.stringify(data) 
    }),
    onSuccess: (_data, variables) => {
      // 乐观更新:立即更新缓存
      queryClient.setQueryData(['user', variables.id], variables);
      // 或者直接失效,重新获取
      queryClient.invalidateQueries({ queryKey: ['user', variables.id] });
    },
  });
}

🎯 TanStack Query vs SWR 选择

8. Edge Cache 与 Edge Computing

Edge Cache 把缓存推到离用户最近的节点。Edge Computing 更进一步——在边缘节点直接执行计算逻辑。

8.1 Edge Cache 平台对比

平台边缘缓存边缘计算KV 存储定价
Cloudflare✅ Cache API✅ Workers✅ KV / D1 / R2免费 10 万请求/天
Vercel✅ Edge Cache✅ Edge Functions✅ Edge ConfigHobby 免费
CloudFront✅ CDN 缓存✅ Lambda@Edge⚠️ 需 DynamoDB按流量计费
Fastly✅ Varnish-based✅ Compute@Edge⚠️ 有限企业定价
Deno Deploy⚠️ 需手动✅ Deno Runtime✅ KV免费 100 万请求

8.2 Edge 缓存实战:Cloudflare KV

// Cloudflare Worker + KV:边缘缓存 + 持久化
export default {
  async fetch(request, env) {
    const url = new URL(request.url);
    const cacheKey = `api:${url.pathname}`;
    
    // 1. 先查 KV(持久化边缘存储)
    const kvValue = await env.MY_KV.get(cacheKey, { type: 'json' });
    if (kvValue) {
      return new Response(JSON.stringify(kvValue), {
        headers: { 
          'Content-Type': 'application/json',
          'X-Cache': 'KV-HIT',
          'Cache-Control': 'public, s-maxage=60' 
        },
      });
    }
    
    // 2. Miss: 请求源站
    const response = await fetch(request);
    const data = await response.json();
    
    // 3. 写入 KV,TTL 5 分钟
    await env.MY_KV.put(cacheKey, JSON.stringify(data), { 
      expirationTtl: 300 
    });
    
    return new Response(JSON.stringify(data), {
      headers: { 
        'Content-Type': 'application/json',
        'X-Cache': 'MISS',
        'Cache-Control': 'public, s-maxage=60' 
      },
    });
  },
};

8.3 Vercel Edge 缓存

// Next.js Route Handler: Edge Runtime + 缓存
export const runtime = 'edge';
export const revalidate = 60; // ISR: 60 秒

export async function GET(request) {
  const { searchParams } = new URL(request.url);
  const category = searchParams.get('category');
  
  // 这个响应会被 Vercel Edge Cache 缓存 60 秒
  const products = await fetch(`https://api.example.com/products?cat=${category}`, {
    next: { revalidate: 60 }  // fetch 级别的 ISR
  }).then(r => r.json());
  
  return Response.json(products);
}

9. 应用层缓存模式

9.1 进程内缓存 (In-Memory LRU)

最快的缓存——零网络延迟。适合不频繁变化且不需要跨进程共享的数据。

# Python: functools.lru_cache (最简单)
from functools import lru_cache

@lru_cache(maxsize=1024)
def get_config(key: str) -> str:
    return db.configs.get(key)

# Node.js: 自实现 LRU
class LRUCache {
  constructor(maxSize = 1000) {
    this.maxSize = maxSize;
    this.cache = new Map(); // Map 保持插入顺序
  }
  
  get(key) {
    if (!this.cache.has(key)) return undefined;
    // 移到末尾(最近使用)
    const value = this.cache.get(key);
    this.cache.delete(key);
    this.cache.set(key, value);
    return value;
  }
  
  set(key, value, ttlMs) {
    this.cache.delete(key); // 删除旧的(如果存在)
    if (this.cache.size >= this.maxSize) {
      // 删除最旧的(第一个)
      const firstKey = this.cache.keys().next().value;
      this.cache.delete(firstKey);
    }
    this.cache.set(key, { value, expireAt: Date.now() + ttlMs });
  }
}

9.2 计算结果缓存 (Memoization)

# 缓存耗时的计算结果
import hashlib
import redis

r = redis.Redis()

def cached_compute(data: str, ttl: int = 3600):
    """缓存任意计算结果"""
    key = f"compute:{hashlib.md5(data.encode()).hexdigest()}"
    cached = r.get(key)
    if cached:
        return json.loads(cached)
    
    # 昂贵的计算
    result = expensive_computation(data)
    r.setex(key, ttl, json.dumps(result))
    return result

# Next.js: unstable_cache (React Server Components)
import { unstable_cache } from 'next/cache';

const getCachedProducts = unstable_cache(
  async (category) => {
    return db.products.findByCategory(category);
  },
  ['products'],
  { revalidate: 60, tags: ['products'] }
);

// 按需失效
import { revalidateTag } from 'next/cache';
revalidateTag('products');

9.3 多级缓存 (L1 + L2)

# 进程内 L1 + Redis L2 + 数据库
import time

class MultiLevelCache:
    def __init__(self, l1_max=500, l1_ttl=60, l2_ttl=300):
        self.l1 = {}  # 进程内字典
        self.l1_max = l1_max
        self.l1_ttl = l1_ttl
        self.l2_ttl = l2_ttl
        self.r = redis.Redis(decode_responses=True)
    
    async def get(self, key: str):
        # L1: 进程内缓存(最快)
        if key in self.l1:
            entry = self.l1[key]
            if entry['expire_at'] > time.time():
                return entry['value']
            del self.l1[key]
        
        # L2: Redis(次快)
        cached = self.r.get(key)
        if cached:
            value = json.loads(cached)
            # 回填 L1
            self._set_l1(key, value)
            return value
        
        # L3: 数据库(最慢)
        value = await self._load_from_db(key)
        if value is not None:
            self.r.setex(key, self.l2_ttl, json.dumps(value))
            self._set_l1(key, value)
        return value
    
    def _set_l1(self, key, value):
        if len(self.l1) >= self.l1_max:
            # 简单淘汰:删除最旧的
            oldest = min(self.l1.items(), key=lambda x: x[1]['expire_at'])
            del self.l1[oldest[0]]
        self.l1[key] = {'value': value, 'expire_at': time.time() + self.l1_ttl}
    
    def invalidate(self, key: str):
        self.l1.pop(key, None)
        self.r.delete(key)

10. 缓存失效:计算机科学两大难题之一

Phil Karlton 的名言:"There are only two hard things in Computer Science: cache invalidation and naming things."

10.1 失效策略对比

策略一致性复杂度适用场景
TTL 过期最终一致(有窗口)⭐ 最简单所有场景的基础
主动删除强一致(几乎)⭐⭐写后立即读
事件驱动失效强一致⭐⭐⭐多服务共享缓存
版本号/标签强一致⭐⭐⭐批量失效

10.2 事件驱动失效(推荐)

# 数据库变更 → 消息队列 → 缓存失效
# 使用 PostgreSQL LISTEN/NOTIFY 或 Redis Pub/Sub

# PostgreSQL 方案
import asyncio
import asyncpg

async def listen_for_changes():
    conn = await asyncpg.connect(DATABASE_URL)
    await conn.add_listener('cache_invalidation', on_invalidation)
    
async def on_invalidation(connection, pid, channel, payload):
    """收到数据库变更通知,删除对应缓存"""
    key = payload  # 如 "user:123"
    r.delete(key)
    print(f"Cache invalidated: {key}")

# 数据库触发器
# CREATE OR REPLACE FUNCTION notify_cache_invalidation()
# RETURNS trigger AS $$
# BEGIN
#   PERFORM pg_notify('cache_invalidation', 'user:' || NEW.id);
#   RETURN NEW;
# END;
# $$ LANGUAGE plpgsql;

# Redis Pub/Sub 方案(跨服务)
import redis

pubsub = r.pubsub()
pubsub.subscribe('cache:invalidate')

def invalidation_listener():
    for message in pubsub.listen():
        if message['type'] == 'message':
            key = message['data']
            r.delete(key)

# 发布失效事件
async def update_user(user_id, data):
    await db.users.update(user_id, data)
    r.publish('cache:invalidate', f'user:{user_id}')

10.3 标签化批量失效

# 给缓存 key 打标签,支持批量失效
class TaggedCache:
    def __init__(self, redis_client):
        self.r = redis_client
    
    def set(self, key: str, value, tags: list[str], ttl: int = 300):
        """设置缓存 + 标签"""
        pipe = self.r.pipeline()
        pipe.setex(key, ttl, json.dumps(value))
        
        # 记录 key 属于哪些标签
        for tag in tags:
            pipe.sadd(f"tag:{tag}", key)
            pipe.expire(f"tag:{tag}", ttl + 60)  # 标签比数据多活一会儿
        
        pipe.execute()
    
    def invalidate_tag(self, tag: str):
        """失效某个标签下的所有缓存"""
        keys = self.r.smembers(f"tag:{tag}")
        if keys:
            pipe = self.r.pipeline()
            for key in keys:
                pipe.delete(key)
            pipe.delete(f"tag:{tag}")
            pipe.execute()

# 使用示例
cache = TaggedCache(r)
cache.set("product:1", product_data, tags=["products", "category:electronics"])
cache.set("product:2", product_data_2, tags=["products", "category:books"])

# 某个分类下的商品全部更新了 → 批量失效
cache.invalidate_tag("category:electronics")

# 所有商品都更新了
cache.invalidate_tag("products")

11. LLM/AI 场景的缓存策略

AI 场景的缓存有独特挑战:响应是非确定性的、成本极高、延迟敏感。这是 Agent 研究中的关键发现。

🔗 与 Agent 研究的关联

11.1 Prompt 缓存(Provider 级别)

# OpenAI Prompt Caching:相同前缀的 prompt 自动缓存
# - 缓存输入 token 的 50% 费用
# - 缓存命中延迟降低 80%
# - 最少 1024 token 才触发

# 最佳实践:把不变的部分放在 prompt 前面
response = openai.chat.completions.create(
    model="gpt-4o",
    messages=[
        # 系统指令(每次相同 → 被缓存)
        {"role": "system", "content": SYSTEM_PROMPT},
        # 工具描述(每次相同 → 被缓存)  
        {"role": "system", "content": TOOLS_DESCRIPTION},
        # 对话历史(会变化,放在后面)
        *conversation_history,
        # 用户当前输入
        {"role": "user", "content": user_input},
    ],
    # Anthropic: 显式启用缓存标记
    # headers={"anthropic-beta": "prompt-caching-2024-07-31"}
)

# OpenClaw 的 Cache Boundary 设计:
# 
# 这个标记把 prompt 分为"可缓存区"和"动态区"
# 可缓存区(标记上方):系统指令、工具定义、用户画像
# 动态区(标记下方):对话历史、当前输入
# Provider 会自动缓存可缓存区的 token

11.2 语义缓存 (Semantic Cache)

# 语义缓存:相似问题复用缓存(不仅限于精确匹配)
# 使用向量相似度判断"语义相同"的查询

from redisvl.index import SearchIndex
from redisvl.query import VectorQuery

class SemanticCache:
    def __init__(self, threshold=0.92, ttl=3600):
        self.threshold = threshold  # 相似度阈值
        self.ttl = ttl
        # Redis + 向量搜索
        self.index = SearchIndex.from_dict({
            "name": "semantic_cache",
            "prefix": "scache:",
            "fields": [
                {"name": "prompt", "type": "text"},
                {"name": "response", "type": "text"},
                {"name": "embedding", "type": "vector",
                 "attrs": {"dims": 1536, "algorithm": "flat", "distance_metric": "cosine"}},
            ]
        })
    
    async def get(self, prompt: str, embedding: list[float]):
        """查找语义相似的缓存"""
        query = VectorQuery(
            vector=embedding,
            vector_field_name="embedding",
            return_fields=["prompt", "response", "__vector_score"],
            num_results=1,
        )
        results = self.index.query(query)
        
        if results and float(results[0]['__vector_score']) >= self.threshold:
            return results[0]['response']  # 缓存命中
        return None
    
    async def set(self, prompt: str, response: str, embedding: list[float]):
        """缓存 prompt-response 对"""
        self.index.load([{
            "prompt": prompt,
            "response": response,
            "embedding": embedding,
        }], ttl=self.ttl)

11.3 AI 缓存层级

用户输入 → ① 精确匹配缓存 (hash → Redis) ↓ miss ② 语义缓存 (embedding → Vector DB, ≥0.92 相似度) ↓ miss ③ Prompt Cache (Provider 级别, 前缀匹配) ↓ miss ④ 完整 LLM 调用 ($0.01-0.10 / 1K token)
💡 AI 缓存的 ROI:一个日活 10 万用户的 AI 产品,语义缓存命中率 30% 就能节省每月 $5,000+ 的 API 费用。精确匹配缓存命中率通常在 10-15%,语义缓存可以提升到 25-40%。

12. 决策树:我该用什么缓存?

🌳 缓存选型决策树

数据是静态资源(JS/CSS/图片)?
→ CDN + immutable Cache-Control + 文件名哈希
数据是公开 API 响应(非用户私有)?
→ CDN s-maxage + SWR + Redis 应用缓存
数据是用户私有数据?
→ Redis 应用缓存 (private, no-cache) + TanStack Query 前端
数据是 LLM 请求/响应?
→ Prompt Cache (Provider) + 语义缓存 (Vector DB) + Redis 精确缓存
数据是实时/不可缓存的?
→ no-store + WebSocket/SSE 实时推送
读多写少?
→ Write-Through 或 Refresh-Ahead(保证缓存始终热)
写多读少?
→ 不缓存,或 Write-Behind(如计数器)
需要跨服务共享缓存?
→ Redis 集群 + Pub/Sub 失效通知
只需要进程内缓存?
→ LRU (单进程) 或 Caffeine (Java) / lru-cache (Node)

13. 技术选型对比

13.1 缓存引擎对比

引擎类型数据结构持久化集群最佳场景
Redis内存 KV7 种✅ RDB/AOF✅ Cluster通用缓存、排行榜、消息队列
Memcached内存 KVString✅ 一致性哈希纯缓存、简单 KV、多线程
Dragonfly内存 KVRedis 兼容Redis 替代、25x 内存效率
KeyDB内存 KVRedis 兼容✅ Active-Active多主 Redis 替代
Cloudflare KV边缘 KVString + Metadata✅ 全球分布边缘缓存、配置中心
Vercel KV边缘 KVRedis 兼容Next.js 项目、Edge Runtime

13.2 Redis vs Memcached

Redis 优势

  • 7 种数据结构(Hash, Set, Sorted Set 等)
  • 持久化(RDB + AOF)
  • Pub/Sub、Lua 脚本、Stream
  • 集群模式、哨兵模式
  • Bitmap、HyperLogLog、Geo

Memcached 优势

  • 多线程(Redis 单线程)
  • 更简单的内存管理
  • 更低的 CPU 开销
  • LRU 更成熟
  • 适合纯缓存场景
💡 2026 年推荐:除非你已有 Memcached 基础设施,否则新项目一律用 Redis。Redis 的多数据结构和生态优势远超 Memcached 的多线程优势。如果 Redis 单线程成为瓶颈,考虑 Dragonfly(Redis 协议兼容 + 多线程)。

13.3 前端数据缓存库对比

大小核心特性框架适合
TanStack Query~40KB乐观更新、无限滚动、Mutation、DevToolsReact/Vue/Svelte/Solid企业级应用
SWR~13KBSWR 模式、聚焦刷新、轻量React中小项目
Apollo Client~80KBGraphQL 缓存、规范化、本地状态React/VueGraphQL 项目
RTK Query~15KBRedux 集成、缓存、轮询React (Redux)Redux 用户
zustand~2KB状态管理(非数据缓存)React轻量状态管理

14. 常见反模式与坑

❌ 反模式 1:缓存一切

不是所有数据都该缓存。频繁写入的数据(如库存)缓存反而增加不一致风险。原则:读多写少才缓存,读写比 < 3:1 的不考虑缓存。

❌ 反模式 2:永不过期

不设 TTL 的缓存是定时炸弹——内存泄漏、数据陈旧。即使逻辑过期,也必须设 Redis TTL 作为安全网。SETEX > SET

❌ 反模式 3:缓存大对象

超过 100KB 的 value 在 Redis 中是 big key,会导致阻塞。解决方案:压缩(gzip)、分片、或者只缓存 ID 列表。

❌ 反模式 4:只删不设

很多开发者习惯"更新数据库 → 删除缓存",但如果缓存命中率很高,删除后下次请求必然 miss。写穿透(更新缓存而非删除)在高命中场景更好。

❌ 反模式 5:缓存和数据库双写不一致

先删缓存再更新数据库?高并发下旧数据可能回填缓存。解决方案:

❌ 反模式 6:缓存依赖链

A 缓存依赖 B 缓存的结果,B 又依赖 C…… 一层 miss 全链 miss,比不缓存还慢。原则:缓存层不依赖其他缓存,每层直接查数据源。

15. 缓存架构检查清单

✅ 设计阶段

✅ 实现阶段

✅ 运维阶段

🔗 相关模块

SaaS 全栈知识库 · 缓存策略 · 2026