一、问题背景：一个"看起来没问题"的API

去年接手了一个内部商品聚合查询服务，逻辑很简单：客户端传一个商品ID，服务端并发调用3个上游接口（库存、价格、评价），再查一次数据库拿基础信息，拼装后返回。

代码大概是这样的：

import requests
from flask import Flask, jsonify

app = Flask(__name__)

@app.route("/product/")
def get_product(pid):
    base = db.query("SELECT * FROM products WHERE id=%s", pid)
    stock = requests.get(f"http://stock-svc/stock/{pid}", timeout=2).json()
    price = requests.get(f"http://price-svc/price/{pid}", timeout=2).json()
    review = requests.get(f"http://review-svc/review/{pid}", timeout=2).json()
    return jsonify({**base, "stock": stock, "price": price, "review": review})

单看逻辑没毛病，但上线后监控数据很难看：32核机器，QPS卡在1200上不去，CPU利用率不到15%，P99延迟1.8秒。典型的"CPU闲着，请求堵着"——4个IO操作串行执行，每个平均200ms，一次请求光等待就800ms，Flask默认多线程模型下线程池很快被打满，后面的请求只能排队。

这不是加机器能解决的问题。核心矛盾是：IO密集型服务用了同步阻塞模型，线程数成了并发上限。

二、环境与版本

改造时的技术栈，版本号都列清楚，避免读者踩版本兼容的坑：

  • Python 3.11.6（3.11的asyncio有显著性能优化，比3.10快约10-20%）
  • FastAPI 0.104.1
  • uvicorn 0.24.0（uvloop 0.19.0，httptools 0.6.1）
  • aiohttp 3.9.1（客户端）
  • asyncpg 0.29.0
  • PostgreSQL 14.9

启动命令用的是：

uvicorn app.main:app --host 0.0.0.0 --port 8000 \
  --workers 8 --loop uvloop --http httptools \
  --limit-concurrency 2000 --backlog 4096

--workers 8 是按 32核 / 4 的经验值来的，每个worker一个独立事件循环，避免单循环成为瓶颈。

三、方案设计：三个关键决策

改造前想清楚了三件事：

1. 框架选FastAPI而不是aiohttp web。 后者性能略好，但FastAPI的依赖注入、Pydantic校验、OpenAPI文档对团队协作更友好，实测差距在5%以内，不值得为这点性能牺牲可维护性。

2. 上游调用并发化。 三个上游接口之间没有依赖，用 asyncio.gather 并发，理论耗时从 sum 变成 max。这是收益最大的一步。

3. 数据库连接池要够大但要克制。 asyncpg的pool不是越大越好，PG默认 max_connections=100，8个worker × 每池20连接 = 160，超了会直接报错。最终每worker配15，留了余量。

另外加了一层超时控制：每个上游接口单独 timeout，整体请求用 asyncio.wait_for 兜底，避免慢上游拖垮整个事件循环。

四、核心实现：改造后的代码

先看改造后的完整实现，再逐段解释。

import asyncio
import asyncpg
import aiohttp
from fastapi import FastAPI, HTTPException
from contextlib import asynccontextmanager

app = FastAPI()
session: aiohttp.ClientSession = None
pool: asyncpg.Pool = None

@asynccontextmanager
async def lifespan(app: FastAPI):
    global session, pool
    # 连接池：每worker 15个，总连接数 8*15=120 < PG max_connections
    pool = await asyncpg.create_pool(
        dsn="postgresql://user:pwd@db:5432/shop",
        min_size=5, max_size=15,
        command_timeout=1.0,
    )
    # aiohttp连接池：limit_per_host=50，避免打爆上游
    connector = aiohttp.TCPConnector(
        limit=200, limit_per_host=50,
        ttl_dns_cache=300, enable_cleanup_closed=True,
    )
    session = aiohttp.ClientSession(
        connector=connector,
        timeout=aiohttp.ClientTimeout(total=1.5, connect=0.5),
    )
    yield
    await session.close()
    await pool.close()

app = FastAPI(lifespan=lifespan)

async def fetch_json(url: str):
    async with session.get(url) as resp:
        resp.raise_for_status()
        return await resp.json()

async def get_product(pid: int):
    try:
        # 四个IO并发，整体耗时 = 最慢的那个
        base_task = pool.fetchrow(
            "SELECT id, name, category FROM products WHERE id=$1", pid)
        stock_task = fetch_json(f"http://stock-svc/stock/{pid}")
        price_task = fetch_json(f"http://price-svc/price/{pid}")
        review_task = fetch_json(f"http://review-svc/review/{pid}")

        base, stock, price, review = await asyncio.wait_for(
            asyncio.gather(base_task, stock_task, price_task, review_task),
            timeout=1.2,
        )
        if base is None:
            raise HTTPException(404, "product not found")
        return {**dict(base), "stock": stock, "price": price, "review": review}
    except asyncio.TimeoutError:
        raise HTTPException(504, "upstream timeout")
    except aiohttp.ClientError as e:
        raise HTTPException(502, f"upstream error: {type(e).__name__}")

几个点值得单独说：

lifespan 里建池，别在模块顶层建。 我最初把 ClientSession 写成了模块级变量，结果多worker下每个进程都建，而且uvicorn热重载时会泄漏连接。用lifespan能保证进程退出时干净关闭。

asyncio.gather 默认的异常行为是"第一个异常就抛出，但其他task不取消"。 这意味着如果stock接口挂了，price和review的请求还在跑，浪费资源。严谨点应该用 return_exceptions=True 再手动处理，或者用 TaskGroup（3.11+）。我为了代码简洁用了gather+wait_for兜底，实际生产建议用TaskGroup。

超时是分层的： aiohttp层 total=1.5s 管单请求，asyncio.wait_for 的1.2s管整体，数据库 command_timeout=1.0。三层超时确保任何一层卡住都不会无限等待。

ttl_dns_cache=300 是个容易被忽略的优化。 默认aiohttp每次新建连接都要解析DNS，内网服务也走DNS，加上这个缓存后P50延迟降了约15ms。

五、踩坑与优化：官方文档不会告诉你的

改造过程不是一次成功的，记录几个真实踩过的坑：

坑1：事件循环被同步代码阻塞。 改造初期我留了一处 json.loads 读本地配置文件在请求路径里，虽然只要2ms，但高并发下事件循环被反复阻塞，QPS上不去。asyncio里任何同步IO或CPU密集操作都是毒药，要么扔进 run_in_executor，要么启动时预加载。后来所有配置都在lifespan里读好。

坑2：连接池耗尽导致的"假死"。 压测到5000 QPS时出现大量超时，日志里是 TimeoutError: Pool is full。原因是上游price服务偶尔变慢，连接被占住，新请求排队等连接。解决办法是给 session.get 单独设 timeout 并配合 connector.limit_per_host，让等待连接的时间可控。另外把 max_size 从15调到20，同时把PG的 max_connections 从100调到150。

坑3：uvloop和某些C扩展不兼容。 项目里有个老库用了 gevent 的monkey patch，和uvloop冲突导致段错误。最终方案是移除该依赖。如果你的项目里有非asyncio-aware的C扩展，慎用uvloop。

坑4：--workers 不是越多越好。 一开始设了16个worker，结果上下文切换开销反而让QPS降了8%。32核机器上8-12个worker是甜点区，因为每个worker内部已经是异步高并发了，worker数超过物理核数/2收益递减。

优化后的最终参数：8 worker、uvloop、每worker 20个DB连接、aiohttp limit=200/limit_per_host=50、三层超时（1.5s/1.2s/1.0s）。

六、效果数据

用wrk压测，同一台32核64G机器，请求体相同（随机pid，命中率约70%），压测时长5分钟取稳定值：

指标 改造前（Flask+requests） 改造后（FastAPI+asyncio） 提升
QPS 1,200 8,700 7.25x
P50 延迟 620 ms 85 ms 7.3x
P99 延迟 1,800 ms 210 ms 8.6x
CPU 利用率 12% 68% -
单机内存 4.2 GB 2.8 GB -33%
服务器数量 6 台 3 台（含冗余） -50%

QPS提升7倍多，主要来自两处：串行IO变并发（理论4x）和线程模型换成事件循环（消除线程池排队）。P99从1.8s降到210ms，用户体验是质变。

成本上，原来6台机器缩到3台，云账单每月省了约40%。这也是我坚持做这次改造的直接动力。

七、总结

这次改造最大的体会是：asyncio不是银弹，它解决的是"IO等待"问题，不解决"代码写得烂"问题。 如果上游接口本身慢、或者有同步阻塞代码，换成asyncio照样卡。

几条可以直接抄走的经验：

  1. 所有外部调用必须设超时，且分层设置（连接/单请求/整体）。
  2. 连接池大小要算总账，worker数 × 池大小 < 下游max_connections。
  3. ClientSession 和 Pool 在lifespan里创建和关闭，别用模块级变量。
  4. 请求路径上绝对不能有同步IO，配置文件启动时读好。
  5. worker数从"核数/4"开始调，压测找甜点，别拍脑袋。

最后提醒一句：异步改造的收益和IO占比强相关。如果你的服务是CPU密集型，asyncio帮不上忙，该上多进程上多进程。判断标准很简单——看监控里CPU利用率和线程等待时间，CPU闲而请求堵，那就是asyncio的用武之地。