当前位置：首页 > news >正文

后端缓存好？缓存实用的方案实例直接用就是前端缓存好还

news 2025/10/12 11:45:44

后端缓存好？缓存实用的方案实例直接用就是前端缓存好还

文章目录

怎么选？
典型分层方案
什么时候偏前端？
什么时候偏后端？
关键实践清单
结论
缓存方案实例
- 一、目标与分工
- 二、架构与关键路径
- 三、Nginx 网关缓存（可直接用）
- 四、FastAPI 应用层缓存（可复制运行）
- - 1) 依赖与初始化
  - 2) 缓存键与装饰器（含防击穿锁、随机 TTL 抗雪崩、标签失效）
  - 3) 示例接口
  - 4) 主动失效（按标签/按键）
  - 5) 负缓存/穿透防护（可选）
- 五、Redis 策略与配置
- 六、Docker Compose（开箱即用）
- 七、前端/浏览器层建议（可选做）
- 八、测试用例（马上验证）
- 九、监控与告警
- 十、常见坑处理
- 十一、选型建议

都重要。前端缓存负责“离用户最近的静态与短期数据”，后端缓存负责“跨用户复用与复杂查询的结果”。最佳实践通常是分层缓存（CDN/浏览器 → 网关/服务端 → 数据库），各司其职。

怎么选？

维度	前端缓存（浏览器/Service Worker/CDN 边缘）	后端缓存（反向代理/应用层/Redis/数据库缓存）
适用数据	公共静态资源、低敏感公开数据、列表页骨架等	计算密集/查询昂贵结果、个性化但可控的数据片段
粒度	资源级（HTML/CSS/JS/图片），也可做接口响应片段	业务对象/查询结果/页面片段
命中范围	强（同一资源所有用户受益，尤其 CDN）	强（同一查询或对象多用户受益）
一致性	难做强一致；适合最终一致 + 短 TTL	更易控制一致性与失效策略
安全/隐私	需谨慎（避免把私有数据缓存到共享层）	可基于用户维度安全隔离（如按 user_id 分桶）
失效控制	依赖 HTTP 缓存头、URL 版本号、SW 逻辑	由应用主动失效，粒度更细（键精确删除）
成本/复杂度	较低（合理设置 Cache-Control/ETag）	中等（键设计、淘汰策略、回源风暴治理）

典型分层方案

CDN/浏览器层（首选减载）：
边缘/网关层（如 Nginx/反向代理）：
- 对公开 GET 接口设置短 TTL（5–120s）吸收突刺流量。
- 对热门页面做页面片段缓存（ESI/SSI）或整页短缓存。
应用/数据层（如 Redis）：

什么时候偏前端？

静态资源、公共接口、SEO 友好的落地页、对“首屏 TTFB/FCP”极敏感的场景。
全球流量、带宽贵：CDN 命中能显著降本提速。

关键实践清单

HTTP 缓存头：Cache-Control（含 s-maxage、stale-while-revalidate）、ETag/If-None-Match、Last-Modified。
缓存键设计：包含影响结果的所有维度（语言/地区/版本/用户/权限/查询条件）。
TTL 策略：公共数据长 TTL + 版本哈希；动态数据短 TTL + 主动失效。
一致性：对强一致读，绕过缓存或采用短 TTL + 回源校验；对最终一致，接受微小延迟换取性能。
风暴治理：单飞（singleflight）/互斥重建、限速、降级兜底。
监控与命中率：命中率、回源量、P95/P99、重建耗时、错误率。

结论

静态/公共 → 前端（CDN/浏览器）长缓存 + 版本号。
动态/昂贵/需控制失效 → 后端（Redis/代理）短缓存 + 主动失效。
两者组合：前端“兜头部延迟”，后端“护数据库压力”。如果只能选一个，先上后端缓存更稳；有条件再加前端层吃满提速与降本。

缓存方案实例

Nginx + FastAPI + Redis 的可落地分层缓存方案（前端/CDN、网关、应用、数据层）。内容含：架构、Nginx 配置、FastAPI 代码（含装饰器/锁/失效）、Redis 策略、Docker Compose、测试与监控清单。

一、目标与分工

CDN/浏览器层：静态资源长缓存 + 公开接口短缓存（SWR）。
Nginx（边缘/网关）：公开 GET 接口短 TTL、吸收突刺、记录命中率。
应用层（FastAPI）：昂贵查询对象/结果缓存、精细失效（按键/按标签）、防击穿/雪崩。
Redis：高性能 KV 缓存、互斥锁（防风暴）、可选布隆/短期负缓存。

二、架构与关键路径

Browser/PWA ──> CDN(可选) ──> Nginx(反向代理+proxy_cache)└──> FastAPI(App缓存装饰器/主动失效/ETag/SWR)└──> Redis(对象/查询结果缓存、锁、标签集)└──> DB/外部服务

三、Nginx 网关缓存（可直接用）

作用：公开 GET 接口短缓存（例如 30~120s），并输出命中状态；静态资源用指纹+长缓存。

# /etc/nginx/conf.d/cache.conf
proxy_cache_path /var/cache/nginx levels=1:2 keys_zone=apicache:100mmax_size=5g inactive=10m use_temp_path=off;
map $request_method $bypass_non_get {default 1;GET     0;
}
# 对登录态/私有请求绕过缓存（有 Authorization/ Cookie）
map $http_authorization $bypass_auth { default 1; "" 0; }
map $http_cookie        $bypass_cookie { default 1; "" 0; }
server {listen 80;server_name _;# 静态资源（带文件指纹）location ~* \.(?:css|js|png|jpg|jpeg|gif|svg|woff2?)$ {root /var/www/html;add_header Cache-Control "public, max-age=31536000, immutable";try_files $uri =404;}# 公开接口（示例：/api/public/**）location ^~ /api/public/ {proxy_pass         http://app:8000;proxy_set_header   Host $host;proxy_set_header   X-Forwarded-For $remote_addr;# 缓存键：考虑查询串proxy_cache_key "$scheme$host$request_uri";# 仅 GET 且无鉴权/无 Cookie 才缓存set $bypass 0;if ($bypass_non_get) { set $bypass 1; }if ($bypass_auth)    { set $bypass 1; }if ($bypass_cookie)  { set $bypass 1; }proxy_no_cache       $bypass;proxy_cache_bypass   $bypass;proxy_cache          apicache;proxy_cache_valid    200  60s;      # 命中 60sproxy_cache_valid    301 302 10m;proxy_cache_valid    any  30s;proxy_ignore_headers Set-Cookie;add_header X-Cache-Status $upstream_cache_status;add_header Cache-Control "public, s-maxage=60, stale-while-revalidate=120";proxy_headers_hash_max_size 512;proxy_headers_hash_bucket_size 128;# 回源慢时使用陈旧缓存proxy_cache_use_stale error timeout http_500 http_502 http_503 http_504 updating;}# 私有/敏感接口（默认不缓存）location /api/ {proxy_pass       http://app:8000;proxy_set_header Host $host;add_header Cache-Control "no-store";}
}

监控命中率：$upstream_cache_status 会显示 HIT/MISS/BYPASS/EXPIRED/STALE。

四、FastAPI 应用层缓存（可复制运行）

1) 依赖与初始化

pip install fastapi uvicorn redis[async] orjson python-multipart

# app/main.py
import asyncio, hashlib, json, time
from typing import Any, Callable, Dict, Optional
from fastapi import FastAPI, Request, Depends, HTTPException
from fastapi.responses import ORJSONResponse
from redis import asyncio as aioredis
app = FastAPI(default_response_class=ORJSONResponse)
redis = aioredis.from_url("redis://redis:6379/0", encoding="utf-8", decode_responses=True)
APP_CACHE_TAG_PREFIX = "tag:"
APP_CACHE_LOCK_PREFIX = "lock:"
APP_CACHE_KEY_PREFIX = "resp:"

2) 缓存键与装饰器（含防击穿锁、随机 TTL 抗雪崩、标签失效）

def _stable_key(parts: Dict[str, Any]) -> str:
raw = json.dumps(parts, sort_keys=True, separators=(",", ":"))
h = hashlib.sha256(raw.encode()).hexdigest()[:32]
return f"{APP_CACHE_KEY_PREFIX}{h}"
async def _with_mutex_lock(key: str, ttl: int = 10) -> bool:
# SETNX + EX，拿到锁返回 True，拿不到 False
return await redis.set(f"{APP_CACHE_LOCK_PREFIX}{key}", "1", ex=ttl, nx=True)
async def _unlock(key: str):
await redis.delete(f"{APP_CACHE_LOCK_PREFIX}{key}")
def cacheable(ttl: int = 60, tags: Optional[list[str]] = None, vary_user: bool = False):
"""
- ttl: 秒；会自动加入随机抖动 ±20% 防雪崩
- tags: 业务标签（如 ["product:123","list:home"]），用于批量失效
- vary_user: 是否按用户维度缓存（私有但可控）
"""
def decorator(func: Callable):
async def wrapper(request: Request, *args, **kwargs):
user_id = request.headers.get("X-User-Id") if vary_user else None
key_parts = {
"path": request.url.path,
"query": dict(request.query_params),
"user": user_id,
"ver": request.headers.get("X-App-Version", "v1"),  # 版本/地域/语言等可加入
}
key = _stable_key(key_parts)
# 读缓存
cached = await redis.get(key)
if cached:
payload = json.loads(cached)
return ORJSONResponse(payload, headers={
"Cache-Control": f"public, max-age=30, stale-while-revalidate=120",
"ETag": payload.get("_etag",""),
})
# 防击穿：仅一个并发去重建
if await _with_mutex_lock(key):
try:
data: Dict[str, Any] = await func(request, *args, **kwargs)
# ETag（弱校验可用 hash）
etag = hashlib.md5(orjson.dumps(data)).hexdigest()
data["_etag"] = etag
# TTL 抖动
jitter = max(1, int(ttl * 0.2))
real_ttl = ttl + (int(time.time()) % (2*jitter) - jitter)
await redis.set(key, json.dumps(data), ex=max(1, real_ttl))
# 标签索引（tag -> set(keys)）
if tags:
for tag in tags:
await redis.sadd(f"{APP_CACHE_TAG_PREFIX}{tag}", key)
return ORJSONResponse(data, headers={
"Cache-Control": f"public, max-age=30, stale-while-revalidate=120",
"ETag": etag,
})
finally:
await _unlock(key)
else:
# 其他并发短等，或返回兜底
for _ in range(20):
await asyncio.sleep(0.05)
cached2 = await redis.get(key)
if cached2:
payload = json.loads(cached2)
return ORJSONResponse(payload, headers={
"Cache-Control": f"public, max-age=30, stale-while-revalidate=120",
"ETag": payload.get("_etag",""),
})
# 兜底直查
data = await func(request, *args, **kwargs)
return ORJSONResponse(data, headers={"Cache-Control": "no-store"})
return wrapper
return decorator

3) 示例接口

@app.get("/api/public/products")
@cacheable(ttl=60, tags=["products:list"])
async def list_products(request: Request):
# TODO: 实际查询DB/外部服务（这里返回假数据）
return {"items": [{"id": 1, "name": "A"}, {"id": 2, "name":"B"}]}
@app.get("/api/public/product/{pid}")
@cacheable(ttl=120, tags=lambda req,pid: [f"product:{pid}"])  # 也可用固定列表
async def get_product(request: Request, pid: int):
return {"id": pid, "name": f"product-{pid}", "price": 99}

可选：把 tags 支持成 Union[List[str], Callable]，上例里我放了一个思路，你也可以直接写固定列表：tags=[f"product:{pid}"]（运行时构造）。

4) 主动失效（按标签/按键）

async def invalidate_by_tags(tags: list[str]):
for tag in tags:
tkey = f"{APP_CACHE_TAG_PREFIX}{tag}"
members = await redis.smembers(tkey)
if members:
await redis.delete(*members)
await redis.delete(tkey)
@app.post("/admin/product/{pid}/update")
async def update_product(pid: int):
# 1) 执行DB更新...
# 2) 失效相关缓存
await invalidate_by_tags([f"product:{pid}", "products:list"])
return {"ok": True}

5) 负缓存/穿透防护（可选）

查询不存在对象时，缓存一个**短 TTL（10~30s）**的“空标记”（如 {"_none":1}）；
再次请求直接返回 404 或空，避免反复打 DB。

五、Redis 策略与配置

maxmemory & 策略：根据机器内存设置 maxmemory，使用 allkeys-lru 或 volatile-lru。
键设计：包含影响结果的所有维度（语言/地区/版本/用户/权限/分页/筛选）。
TTL 分散：应用层已做 ±20% 抖动，避免雪崩。
热点互斥：上面装饰器里的 SETNX 锁即可。
标签索引：tag:<biz> → Set(key1,key2,...)，方便批量失效。

redis.conf 关键项（示例）：

maxmemory 1gb
maxmemory-policy allkeys-lru

六、Docker Compose（开箱即用）

# docker-compose.yml
version: "3.8"
services:
redis:
image: redis:7-alpine
command: ["redis-server","--appendonly","yes","--maxmemory","1gb","--maxmemory-policy","allkeys-lru"]
ports: ["6379:6379"]
volumes: ["./data/redis:/data"]
app:
image: python:3.11-slim
working_dir: /app
volumes: ["./app:/app"]
command: bash -lc "pip install fastapi uvicorn[standard] redis[async] orjson && uvicorn main:app --host 0.0.0.0 --port 8000"
depends_on: [redis]
ports: ["8000:8000"]
nginx:
image: nginx:1.27-alpine
volumes:
- ./nginx/cache.conf:/etc/nginx/conf.d/default.conf:ro
- ./static:/var/www/html:ro
- nginx_cache:/var/cache/nginx
depends_on: [app]
ports: ["80:80"]
volumes:
nginx_cache:

七、前端/浏览器层建议（可选做）

静态资源文件指纹（app.0a1b2c.js）+ immutable。
PWA/Service Worker：对公开接口做 stale-while-revalidate（注意不要缓存私有数据）。
HTML 不建议长缓存；可用短 TTL + ETag。

八、测试用例（马上验证）

# 1) 静态资源头部
curl -I http://localhost/app.0a1b2c.js
# 2) 公开接口（观察 X-Cache-Status）
curl -i "http://localhost/api/public/products"
curl -i "http://localhost/api/public/products"
# 3) 私有请求绕过缓存
curl -i -H "Authorization: Bearer xxx" "http://localhost/api/public/products"
# 4) 更新后主动失效
curl -i -X POST "http://localhost/admin/product/2/update"
curl -i "http://localhost/api/public/product/2"

九、监控与告警

Nginx：开启日志字段（已添加 X-Cache-Status），统计命中率/回源率/P95。
应用：埋点 cache.hit/miss/rebuild/lock_wait；记录重建耗时。
Redis：监控 used_memory, keyspace_hits/misses, evicted_keys, expired_keys。
告警：命中率异常下降、锁等待过长、回源突增、Redis 内存接近上限等。