接口上线后最先被业务压垮的往往不是业务逻辑本身，而是三类横切能力：重复读打穿数据库恶意刷接口拖垮限流阈值日志散落导致线上问题无法复盘。缓存负责把热点读请求挡在Redis前。限流负责按IP或使用者Token控制调用频率。结构化日志与请求链路ID负责在故障发生后用同一条request_id把整条调用链串起来。

一、整体架构，缓存、限流、日志如何叠在请求链上

一次请求进入FastAPI后大致顺序是：

• RequestContextMiddleware最外层。尽早生成X-Request-ID写入contextvars后续所有日志自动带上。
• SlowAPIMiddleware在进入路由前做限流计数，超限直接429。
• 路由内部的读接口通过装饰器查Redis，未命中再回源。
• 无论是业务HTTPException还是未捕获的RuntimeError都会进入统一处理器或中间件兜底，写入带request_id的错误日志。

工程目录按职责拆分：

demo/
├── logs/app.log # 按天切割的结构化日志
├── app/
│ ├── core/config.py # Redis URL、限流默认值、日志目录
│ ├── core/logging_setup.py # JSON格式化、TimedRotatingFileHandler
│ ├── core/lifespan.py # 启动时init Redis
│ ├── cache/redis_client.py # redis.asyncio/fakeredis
│ ├── cache/decorators.py # 结果缓存装饰器
│ ├── rate_limit/limiter.py # slowapi Limiter、IP/Token key_func
│ ├── middleware/request_context.py
│ ├── exceptions/handlers.py
│ └── api/routes.py
├── run.py
└── scripts/smoke_test.py

二、异步Redis集成

为什么用redis.asyncio而不是同步redis？

旧版独立包Aioredis已合并进官方/redis/asyncio. 在异步路由里若使用同步.get。会阻塞事件循环，与同步SQLAlchemy相同风险。异步客户端每次IO都 alert await ,多请求才能真正并发。

初始化与关闭

async def init_redis -> Any:
if settings.use_fake_redis:
from fakeredis import aioredis as fake_aioredis
_redis = fake_aioredis.FakeRedis
await _redis.ping
return _redis
from redis import asyncio as aioredis
_redis = aioredis.from_url(
settings.redis_url,encoding="utf-8",decode_responses=True,socket_connect_timeout=5,socket_timeout=5,)
await _redis.ping
return _redis

"decode_responses=True": 让 .get/set 直接返回 。后续 .json.loads. .ping: 启动失败即告警,不等到运行时才抛500.

缓存读写封装

def cache_key -> str:
return ":".join for p in parts]])
async def cache_get -> Any | None:
raw = await get_redis.get
if raw is None:
return None
return json.loads
async def cache_set -> None:
ttl = ttl if ttl is not None else settings.cache_default_ttl
await get_redis.set,ex=ttl)

"cache_key": 强制加业务前缀，避免多项目共用一个DB时键冲突——分布式缓存最易忽略细节。统一JSON序列化+秒级TTL防止热点键永久占内存。所有操作都是 await，不堵事件循环.

三、接口结果缓存

显式键装饰器

def product_cache_key -> str:
return cache_key)
@cached_by_key
async def get_product_cached -> dict:
return await _load_product_from_db
async def wrapper:
key = key_builder
hit = await cache_get
if hit is not None:
hit = {**hit。"source": "cache"}
return hit
result = await func
await cache_set
return result 

"product_cache_key": 键形如 "demo:cache:product:" + id ,可读可手动删除，比将整个参数哈希更利于运维。

@router.get @limiter.limit @cached_by_key

• P1 请求 : source=db 耗时 ~200ms+
• P2 请求 : source=cache 毫秒级返回.
• "DELETE /api/v1/products/{id}/cache" 用主动失效策略.

sequenceDiagram participant C as 客户端 participant API as 路由 participant Cache as Redis participant DB as 模拟 DB C->>API: GET /products/ API->Cache: GET key Cache-->API: miss API->DB: await sleep 查库 DB-->API: product API->Cache:set key EX API-->C : source=db C->>API: GET /products/ API->Cache: GET key Cache-->API : hit API-->C : source=cache

分布式缓存的含义

"多 FastAPI worker / 多台机器共享同一 Redis". 单机内存dict仅当前进程生效，多 worker 会各自击穿 DB。把存储换成 Redis URL 后 任意实例写入键 他人立即可读。限流同理，在真实 Redis 模式下 use same URL 作 storage。多实例共享计数，

四、slowapi限流：IP 与 使用者 Token

Limiter 初始化

limiter = Limiter(
key_func=get_remote_address,default_limits=,# 默认 /minute
storage_uri=_storage_uri,)
def _storage_uri -> str:
if settings.use_fake_redis:
return "memory://"
return settings.redis_url
app.state.limiter = limiterapp.add_middleware

• "default_limits" 用于未单独声明限制的接口.
• "storage_uri": 单机 demo 用 memory://,多 worker 用 Redis => 分布式限流.
• "key_func": 默认按客户端 IP 取键.

MUST mount on app:

• @limiter.limit 装饰 的路由 必须声明 request 参数。否则 slowapi 无法读取 IP/Header.

# ip_limited 按 IP 限制示例

python def ipkey -> str : host=request.client.host if request.client else 'unknown' forwarded=request.headers.get if forwarded : host=forwarded.split.strip return f'ip:{host}'

@router.get @limiter.limit async def ip_limited -> dict : ...

/minute 表示每分钟最多 5 次.生产环境需解析 X-Forwarded-For，否则所有人看似 Nginx IP，会误伤站点 . 前缀 ip:,为和 Token 限制隔离，以免碰撞。 第六次出现 429，并带 request_id 和 Retry‑After 建议客户端重试。

# token_limited 按使用者 Token 限制示例 python def get_user_token_or_ip -> str : auth=request.headers.get if auth.lower.startswith : token=auth.strip if token:return f'token:{token}' return f'ip:{get_client_ip}' @router.get @limiter.limit async def user_limited) : ...

• 登录使用者用 Bearer Token 做维度；不同办公网出口 IP 下不同使用者互不影响；未登录回退到 IP，
• 校验 Token 合法性 再计数 超限，否则攻击者可无效 Token 滑过桶。

---

五、结构化日志与请求链路 ID

contextvars传递 request_id

python request_id_ctx : ContextVar = ContextVar client_ip_ctx : ContextVar = ContextVar

• 中间件入口 set 并出口 reset;同一请求任意深度 logger.info 都能读取当前 request_id，无需层层传参。

JSON 格式化 & 按天切割

python class JsonFormatter: def format: payload={ 'ts': datetime.now.isoformat,'level': record.levelname,'logger': record.name,'msg': record.getMessage,'requestid': requestidctx.get。'clientip': clientipctx.get} ... return json.dumps

filehandler = TimedRotatingFileHandler,when='midnight',interval=1,backupCount=14,encoding='utf-8') filehandler.suffix='%Y-%m-%d'

• 每行 JSON，可被 Filebeat/Loki/ELK 采集。
• TimedRotatingFileHandler 每天零点切割保留14天；文件名包含日期后缀，
• 控制台 & 文件共用 Formatter，本地开发与服务器一致。

中间件写入访问日志

python class RequestContextMiddleware: async def dispatch: request_id=request.headers.get or str) start=time.perf_counter response=await call_next elapsed_ms=-start)*1000. response.headers=request_id response.headers=f'{elapsed_ms:.2f}ms' logger.info('request completed'。extra={'path':request.url.path,'method':request.method,'status_code':response.status_code,'duration_ms':round}) return response

• 客户端可传自己的 X‑Request‑ID，服务端生成 UUID 并回传。
• duration_ms 写入字段便于慢请求直接过滤，无需 grep 文本。

---

六统一异常日志捕获

分类处理

python def registerexceptionhandlers ->None :

 @app.exception_handler
async def rate_limit_handler:
logger.warning
return JSONResponse(status_code=429,content={'detail':'请求过于频繁，请稍后再试','request_id':request_id_ctx.get} )
@app.exception_handler
async def http_exception_handler:
return JSONResponse(status_code=exc.status_code,content={'detail':exc.detail,'request_id':request_id_ctx.get} )
@app.exception_handler
async def unhandled_exception_handler:
logger.exception
return JSONResponse(status_code=500。content={'detail':'服务器内部错误','request_id':request_id_ctx.get} )

• 限流校验错误 => warning.
• HTTPException => info 或 error.
• 未知异常 => exception + stack trace.

与 BaseHTTPMiddleware 的兼容

Starlette 的 BaseHTTPMiddleware 有时会将路由抛出的异常重新抛到外层，所以 demo 在 RequestContextMiddleware 内对未处理异常做兜底：记录 “request failed” 日志并返回统一500 JSON。不过，

---

七 请求全链路串联示意

---

高并发接口的三大支柱：

1. 热点读取 把多次查询放进 Redis。并给前缀 + TTL，避免数据库击穿。
2. 分布式限流 用 slowapi + 同一 Redis 存储计数；按 IP 或 Bearer Token 防止刷单。
3. 结构化全链路监控 每条 log 带 X‑Request‑ID 且以 JSON 写文件 → 日志网站快速定位整个调用链。

以上方案已在 demo 项目中实现。上线后只需拿到响应头中的 X‑Request‑ID → 在日志网站过滤 → 查看耗时 / 缓存命中 / 异常堆栈。即可快速定位线上问题，相比传统 grep 散乱文本高效得多。

---