Python Web 性能：FastAPI 异步机制与 Uvicorn worker 调优

# WSGI (PEP 3333) - 同步，阻塞模型
def app(environ, start_response):
    # environ 是字典，所有操作同步
    status = '200 OK'
    headers = [('Content-Type', 'text/plain')]
    start_response(status, headers)
    return [b"Hello"]

# ASGI (ASGI 3.0+) - 异步，双向通道
async def app(scope, receive, send):
    # scope: 连接元信息字典
    # receive: 异步消息接收器
    # send: 异步消息发送器
    await send({
        "type": "http.response.start",
        "status": 200,
        "headers": [[b"content-type", b"text/plain"]],
    })
    await send({"type": "http.response.body", "body": b"Hello"})

# HTTP 请求生命周期事件序列
# 1. http.request.scope  - 建立连接，传递连接信息
# 2. http.request       - 接收请求体（可多次）
# 3. http.response.start - 开始响应
# 4. http.response.body  - 传输响应体（可多次）

# WebSocket 状态机
# websocket.connect -> websocket.accept -> websocket.receive -> websocket.send -> websocket.disconnect

# Lifespan 状态机
# lifespan.startup -> 运行中 -> lifespan.shutdown

# 简化版请求处理流程
async def handle_request(scope, receive, send):
    # 1. 解析 ASGI scope 为请求对象
    request = Request(scope, receive)

    # 2. 路由匹配
    match, remaining_path = routing.match(request.url_path)
    request.path_params = match.kwargs

    # 3. 依赖注入链解析
    dependency_cache = {}
    solved = []
    for dep in route.dependencies:
        solved.append(await solve_dependency(dep, request, dependency_cache))

    # 4. 执行路由函数
    response = await endpoint(*solved, **request.path_params)

    # 5. 响应序列化与发送
    await response(request, send)

from fastapi import Depends, HTTPException

# 同步依赖 - 每个请求阻塞 worker
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

# 异步依赖 - 非阻塞，适合高并发
async def get_async_db():
    async with AsyncSession(engine) as session:
        yield session

# 使用方式
@app.get("/items/{item_id}")
async def get_item(
    item_id: int,
    db = Depends(get_async_db)  # 异步依赖
):
    return await db.get(Item, item_id)

# FastAPI 依赖注入核心逻辑（简化）
class Depends:
    def __init__(self, dependency, use_cache: bool = True):
        self.dependency = dependency
        self.use_cache = use_cache

async def solve_dependencies(request, depends):
    cache_key = id(request) if self.use_cache else None
    results = []

    for dep in depends:
        # 检查缓存
        if cache_key and cache_key in dep_cache:
            results.append(dep_cache[cache_key])
        else:
            # 调用依赖函数
            result = dep.dependency()
            if asyncio.iscoroutine(result):
                result = await result
            results.append(result)
            if cache_key:
                dep_cache[cache_key] = result
    return results

# 陷阱：串行求解（总耗时 = t1 + t2 + t3）
@app.get("/search")
async def search(
    user=.Depends(get_current_user),      # t1 = 10ms
    cache=.Depends(get_cache),                # t2 = 5ms
    db=.Depends(get_db),                      # t3 = 15ms
):
    pass

# 优化：使用 dataclasses朕统一起求解
from dataclasses import dataclass

@dataclass
class RequestContext:
    user: User
    cache: Cache
    db: Session

async def solve_context(request) -> RequestContext:
    # 并发求解所有依赖
    user, cache, db = await asyncio.gather(
        get_current_user(request),
        get_cache(request),
        get_db(request)
    )
    return RequestContext(user, cache, db)

# Request 级别（默认）- 每次请求重新求解
async def get_db():
    return SessionLocal()

# App 级别（单例）- 整个应用生命周期内只求解一次
@app.get("/stats")
async def stats(
    cache=.Depends(get_cache, use_cache=True)
):
    # get_cache 只在首次调用时执行
    pass

# function 级别 - 每次调用都重新求解（禁用缓存）
@app.get("/random")
async def random(
    rng=.Depends(get_random_generator, use_cache=False)
):
    return rng.random()

# gunicorn.conf.py
import multiprocessing

bind = "0.0.0.0:8000"
workers = multiprocessing.cpu_count() * 2 + 1  # 推荐公式
worker_class = "uvicorn.workers.UvicornWorker"
keepalive = 65
timeout = 30
graceful_timeout = 30
max_requests = 1000
max_requests_jitter = 50

# 日志配置accesslog = "-"
errorlog = "-"
loglevel = "info"

# 使用 gunicorn 的 pre_fork / post_fork 钩子监控
import gc
import resource

def pre_fork(server, worker):
    # worker 启动前记录内存
    worker.memory_before = resource.getrusage(resource.RUSAGE_SELF).ru_maxrss

def worker_abort(worker):
    # worker 被强制终止时记录
    import logging
    logging.error(f"Worker {worker.pid} abort, memory: {worker.memory_before}KB")

# 或使用 uwsgidebug 插件

from pydantic import BaseModel, ConfigDict

class Item(BaseModel):
    model_config = ConfigDict(
        str_strip_whitespace=True,
        coerce_numbers_to_str=False,
        # strict=True 会禁用 coercion，提升性能
    )

    id: int
    name: str
    price: float

# Benchmark: 10000 条数据验证
# model_validate vs parse_obj: ~1.8x 性能提升
items = [Item(id=i, name=f"item_{i}", price=i*0.99) for i in range(10000)]
validated = [Item.model_validate(d) for d dict_data]  # v2 API