Redis / CDN / Stale-while-revalidate / Edge Cache — 从浏览器到数据库的完整缓存链路设计
缓存不是"锦上添花",而是 SaaS 产品从"能用"到"好用"的分水岭。一个没有缓存的系统,用户每次点击都打到数据库——当用户量从 100 涨到 10000,数据库直接崩溃。
一个完整的 SaaS 系统有 5 层缓存,每层的角色、失效策略、存储介质完全不同:
Cache-Control / ETag / Service Worker
0ms 免费
最廉价、最快、但最难控制失效
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 最贵
最终真相、永远不能缓存错误数据
最经典的缓存模式。应用程序负责从缓存读数据,miss 时从数据库加载并写入缓存。
# 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
写入时同时更新缓存和数据库,保证缓存始终最新。
# 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
先写缓存,异步批量写入数据库。适合写密集场景。
# 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) # 批量写入
在缓存过期前主动刷新,用户永远不感知 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 要求 |
Redis 是 SaaS 缓存的事实标准。但会用 SET/GET 和用好 Redis 是两个完全不同的层次。
| 结构 | 场景 | 内存效率 | 示例 |
|---|---|---|---|
String | 简单 KV、计数器、JSON 对象 | ⭐⭐ | 缓存用户 profile JSON |
Hash | 对象属性、部分读写 | ⭐⭐⭐⭐ | 用户信息分字段存储 |
List | 时间线、消息队列 | ⭐⭐⭐ | 最近活动列表 |
Sorted Set | 排行榜、延迟队列 | ⭐⭐⭐ | 热门内容排行 |
Set | 去重、标签、共同好友 | ⭐⭐⭐ | 用户标签集合 |
Bitmap | 布尔状态、签到 | ⭐⭐⭐⭐⭐ | 30 天签到状态 (4 字节) |
HyperLogLog | 基数统计(UV) | ⭐⭐⭐⭐⭐ | 页面 UV 统计 (12KB) |
# ❌ 低效:用 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%+
# 高并发下防止缓存击穿(大量请求同时重建缓存)
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
# 缓存雪崩:大量 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,永不过期
u:123 vs user:profile:123,10 亿 key 差 1GB+allkeys-lru(默认推荐)vs volatile-lrulazyfree-lazy-eviction yes,大 key 异步删除不阻塞# 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()
CDN 缓存让你的静态资源和 API 响应从距离用户最近的节点返回,延迟从 200ms 降到 20ms。
| 资源类型 | 是否缓存 | TTL 建议 | 失效方式 |
|---|---|---|---|
| 静态资源 (JS/CSS/图片) | ✅ 强缓存 | 1 年(内容哈希文件名) | 文件名含 hash,无需手动失效 |
| HTML 入口 | ⚠️ 协商缓存 | 短 TTL 或 no-cache | 部署时自动更新 |
| 公开 API (列表/搜索) | ✅ 边缘缓存 | 30s - 5min | Purge API |
| 私有 API (用户数据) | ❌ 不缓存 | — | — |
| SSR 页面 | ⚠️ 有条件缓存 | 1-60s (ISR) | 按需 revalidate |
// 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}`); // 立即刷新
}
// 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;
}
};
HTTP 缓存头是浏览器和 CDN 缓存的基础。理解它们是设计缓存策略的前提。
| 指令 | 作用 | 适用场景 |
|---|---|---|
max-age=3600 | 缓存 3600 秒内有效 | 不常变化的资源 |
s-maxage=60 | CDN 缓存 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 |
# 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
| 特性 | ETag | Last-Modified |
|---|---|---|
| 精度 | 内容级别(哈希) | 秒级(1 秒内多次修改丢失) |
| 计算成本 | 需要哈希计算 | 文件系统元数据 |
| 分布式一致性 | ❌ 不同服务器 ETag 可能不同 | ✅ 时间相对一致 |
| 推荐场景 | CDN 单源 / CDN 自己生成 | 多源站分布式 |
SWR 是缓存策略中"用户体验最好"的模式——先给旧数据(快),后台刷新新数据(准)。
# HTTP 响应头
Cache-Control: max-age=60, stale-while-revalidate=300
# 含义:60 秒内直接用缓存;60-360 秒内返回旧缓存 + 后台验证
# 时间线:
# 0-60s: 直接返回缓存 (fresh)
# 60-360s: 返回旧缓存 + 后台重新验证 (stale)
# >360s: 缓存完全过期,等待新数据
// 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>
);
}
// 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] });
},
});
}
Edge Cache 把缓存推到离用户最近的节点。Edge Computing 更进一步——在边缘节点直接执行计算逻辑。
| 平台 | 边缘缓存 | 边缘计算 | KV 存储 | 定价 |
|---|---|---|---|---|
| Cloudflare | ✅ Cache API | ✅ Workers | ✅ KV / D1 / R2 | 免费 10 万请求/天 |
| Vercel | ✅ Edge Cache | ✅ Edge Functions | ✅ Edge Config | Hobby 免费 |
| CloudFront | ✅ CDN 缓存 | ✅ Lambda@Edge | ⚠️ 需 DynamoDB | 按流量计费 |
| Fastly | ✅ Varnish-based | ✅ Compute@Edge | ⚠️ 有限 | 企业定价 |
| Deno Deploy | ⚠️ 需手动 | ✅ Deno Runtime | ✅ KV | 免费 100 万请求 |
// 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'
},
});
},
};
// 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);
}
最快的缓存——零网络延迟。适合不频繁变化且不需要跨进程共享的数据。
# 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 });
}
}
# 缓存耗时的计算结果
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');
# 进程内 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)
Phil Karlton 的名言:"There are only two hard things in Computer Science: cache invalidation and naming things."
| 策略 | 一致性 | 复杂度 | 适用场景 |
|---|---|---|---|
| TTL 过期 | 最终一致(有窗口) | ⭐ 最简单 | 所有场景的基础 |
| 主动删除 | 强一致(几乎) | ⭐⭐ | 写后立即读 |
| 事件驱动失效 | 强一致 | ⭐⭐⭐ | 多服务共享缓存 |
| 版本号/标签 | 强一致 | ⭐⭐⭐ | 批量失效 |
# 数据库变更 → 消息队列 → 缓存失效
# 使用 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}')
# 给缓存 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")
AI 场景的缓存有独特挑战:响应是非确定性的、成本极高、延迟敏感。这是 Agent 研究中的关键发现。
# 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
# 语义缓存:相似问题复用缓存(不仅限于精确匹配)
# 使用向量相似度判断"语义相同"的查询
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)
| 引擎 | 类型 | 数据结构 | 持久化 | 集群 | 最佳场景 |
|---|---|---|---|---|---|
| Redis | 内存 KV | 7 种 | ✅ RDB/AOF | ✅ Cluster | 通用缓存、排行榜、消息队列 |
| Memcached | 内存 KV | String | ❌ | ✅ 一致性哈希 | 纯缓存、简单 KV、多线程 |
| Dragonfly | 内存 KV | Redis 兼容 | ✅ | ✅ | Redis 替代、25x 内存效率 |
| KeyDB | 内存 KV | Redis 兼容 | ✅ | ✅ Active-Active | 多主 Redis 替代 |
| Cloudflare KV | 边缘 KV | String + Metadata | ✅ | ✅ 全球分布 | 边缘缓存、配置中心 |
| Vercel KV | 边缘 KV | Redis 兼容 | ✅ | ✅ | Next.js 项目、Edge Runtime |
| 库 | 大小 | 核心特性 | 框架 | 适合 |
|---|---|---|---|---|
| TanStack Query | ~40KB | 乐观更新、无限滚动、Mutation、DevTools | React/Vue/Svelte/Solid | 企业级应用 |
| SWR | ~13KB | SWR 模式、聚焦刷新、轻量 | React | 中小项目 |
| Apollo Client | ~80KB | GraphQL 缓存、规范化、本地状态 | React/Vue | GraphQL 项目 |
| RTK Query | ~15KB | Redux 集成、缓存、轮询 | React (Redux) | Redux 用户 |
| zustand | ~2KB | 状态管理(非数据缓存) | React | 轻量状态管理 |
不是所有数据都该缓存。频繁写入的数据(如库存)缓存反而增加不一致风险。原则:读多写少才缓存,读写比 < 3:1 的不考虑缓存。
不设 TTL 的缓存是定时炸弹——内存泄漏、数据陈旧。即使逻辑过期,也必须设 Redis TTL 作为安全网。SETEX > SET。
超过 100KB 的 value 在 Redis 中是 big key,会导致阻塞。解决方案:压缩(gzip)、分片、或者只缓存 ID 列表。
很多开发者习惯"更新数据库 → 删除缓存",但如果缓存命中率很高,删除后下次请求必然 miss。写穿透(更新缓存而非删除)在高命中场景更好。
先删缓存再更新数据库?高并发下旧数据可能回填缓存。解决方案:
A 缓存依赖 B 缓存的结果,B 又依赖 C…… 一层 miss 全链 miss,比不缓存还慢。原则:缓存层不依赖其他缓存,每层直接查数据源。
SaaS 全栈知识库 · 缓存策略 · 2026