一、问题背景：一个被同步IO拖垮的接口

去年接手了一个订单查询服务，接口逻辑本身很简单：

  1. 校验用户token（调用户中心HTTP接口）
  2. 查询订单主表（MySQL）
  3. 查询订单商品明细（MySQL）
  4. 查询物流状态（调第三方物流HTTP接口）
  5. 组装返回

用Flask写的，代码大概长这样：

@app.route("/api/order/")
def get_order(order_id):
    token = request.headers.get("Authorization")
    user = requests.get(f"http://user-center/verify?token={token}", timeout=2).json()
    if not user.get("valid"):
        return {"code": 401}, 401

    order = db.query_one("SELECT * FROM orders WHERE id=%s", order_id)
    items = db.query("SELECT * FROM order_items WHERE order_id=%s", order_id)
    logistics = requests.get(f"http://logistics/api/{order['sn']}", timeout=3).json()

    return {"order": order, "items": items, "logistics": logistics}

单看逻辑没问题，但每个请求里串行做了 2次HTTP + 2次MySQL，累计耗时轻松超过300ms。Flask默认多线程模型（threaded=True）在4核8G机器上，Gunicorn开4个worker × 8线程，压测结果惨不忍睹：

wrk -t8 -c200 -d30s http://127.0.0.1:8000/api/order/12345

结果：

指标 数值
QPS 121
P50 620ms
P99 1840ms
错误率 3.2%（超时）

200并发下线程池被打满，请求排队，CPU大部分时间在等IO，利用率不到15%。这就是典型的IO密集型服务被同步模型限制的场景。

二、环境与版本

  • Python 3.11.6（3.11的asyncio有显著性能优化，比3.8快约20%）
  • aiohttp 3.9.1（HTTP客户端）
  • asyncpg 0.29.0（PostgreSQL/MySQL场景，MySQL用aiomysql 0.2.0）
  • FastAPI 0.109.0 + uvicorn 0.27.0（替代Flask）
  • wrk 4.2.0（压测）
  • 机器：4C8G，Ubuntu 22.04

说明一下为什么换FastAPI而不是继续用Flask + asyncio：Flask本质上不是async原生框架，虽然可以用asgiref包装，但性能损耗和生态兼容问题太多。FastAPI基于Starlette，原生async，且自带Pydantic校验，改造收益最大。

三、方案设计

核心思路是把所有阻塞IO换成异步IO，让单线程在等待期间去处理其他请求。

步骤 原方案 新方案
Web框架 Flask + Gunicorn FastAPI + uvicorn
HTTP客户端 requests aiohttp
MySQL驱动 PyMySQL aiomysql（连接池）
并发模型 多线程 单线程事件循环 + 协程
步骤1/4并发 串行 asyncio.gather 并发

关键点：步骤1（校验token）和步骤2、3、4其实没有强依赖，可以并发发起。校验失败再丢弃结果即可，这样总耗时从"求和"变成"取max"。

四、核心实现

4.1 依赖与配置

pip install fastapi==0.109.0 uvicorn==0.27.0 aiohttp==3.9.1 aiomysql==0.2.0

启动命令（注意worker数，IO密集型不建议开太多）：

uvicorn main:app --host 0.0.0.0 --port 8000 --workers 4 --loop uvloop --http httptools

uvloop和httptools是性能关键，实测比默认asyncio loop快约30%。

4.2 完整代码

# main.py
import asyncio
import aiohttp
import aiomysql
from fastapi import FastAPI, Header, HTTPException
from contextlib import asynccontextmanager

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

# MySQL连接池配置：min 5, max 20
MYSQL_CONF = dict(
    host="127.0.0.1", port=3306,
    user="app", password="***", db="order_db",
    minsize=5, maxsize=20,
    autocommit=True, charset="utf8mb4",
)

@asynccontextmanager
async def lifespan(app: FastAPI):
    global pool, session
    pool = await aiomysql.create_pool(**MYSQL_CONF)
    # 连接池复用，总超时5s，单次连接超时2s
    timeout = aiohttp.ClientTimeout(total=5, connect=2)
    session = aiohttp.ClientSession(timeout=timeout)
    yield
    session.close()
    await session.close()
    pool.close()
    await pool.wait_closed()

app = FastAPI(lifespan=lifespan)


async def verify_token(token: str) -> dict:
    async with session.get(
        "http://user-center/verify",
        params={"token": token},
    ) as resp:
        return await resp.json()


async def query_order(order_id: int) -> tuple:
    async with pool.acquire() as conn:
        async with conn.cursor(aiomysql.DictCursor) as cur:
            await cur.execute(
                "SELECT * FROM orders WHERE id=%s", (order_id,)
            )
            order = await cur.fetchone()
            await cur.execute(
                "SELECT * FROM order_items WHERE order_id=%s", (order_id,)
            )
            items = await cur.fetchall()
    return order, items


async def query_logistics(sn: str) -> dict:
    async with session.get(f"http://logistics/api/{sn}") as resp:
        return await resp.json()


@app.get("/api/order/{order_id}")
async def get_order(order_id: int, authorization: str = Header(None)):
    if not authorization:
        raise HTTPException(status_code=401)

    # 关键：token校验和其他查询并发执行
    verify_task = asyncio.create_task(verify_token(authorization))
    order_task = asyncio.create_task(query_order(order_id))

    user = await verify_task
    if not user.get("valid"):
        order_task.cancel()
        raise HTTPException(status_code=401, detail="invalid token")

    order, items = await order_task
    if not order:
        raise HTTPException(status_code=404)

    # 物流查询依赖订单sn，只能等order完成后再发起
    logistics = await query_logistics(order["sn"])

    return {"order": order, "items": items, "logistics": logistics}

4.3 进一步优化：物流查询也可以提前

上面代码里物流查询依赖order.sn，但如果我们能从order_id直接推算出sn（实际业务中订单号规则固定），就能三步全并发：

sn = generate_sn(order_id)  # 本地计算，无IO
user, (order, items), logistics = await asyncio.gather(
    verify_token(authorization),
    query_order(order_id),
    query_logistics(sn),
    return_exceptions=True,
)

return_exceptions=True很关键，避免某个子任务异常导致整个请求500。

五、踩坑与优化

坑1：连接池大小设错反而变慢。
一开始把maxsize设成100，结果MySQL端too many connections，而且频繁创建销毁连接开销大。最终minsize=5, maxsize=20，配合uvicorn 4 workers，正好吃满MySQL默认151连接上限又不浪费。

坑2：aiohttp的ClientSession必须复用。
最初每个请求里async with aiohttp.ClientSession() as s，压测QPS只有400。改成全局单例后直接翻倍。ClientSession内部维护连接池，每次新建等于放弃复用。

坑3：忘记await导致协程警告。
pool.close()如果写成同步调用会报coroutine was never awaited，必须await pool.wait_closed()。

坑4：CPU密集操作阻塞事件循环。
订单金额计算里有段JSON序列化+签名逻辑，同步执行会阻塞loop。用await asyncio.to_thread(sign, data)丢到线程池解决。

坑5：uvicorn worker数量。
IO密集型不是越多越好。测下来4核机器--workers 4最佳，开8个反而因为上下文切换QPS下降约12%。

六、效果数据

同样命令压测：

wrk -t8 -c200 -d30s http://127.0.0.1:8000/api/order/12345
指标 改造前 改造后 提升
QPS 121 2317 19.1x
P50 620ms 18ms 34x
P99 1840ms 47ms 39x
错误率 3.2% 0% -
CPU利用率 15% 68% -
单机可承载日调用量 ~1000万 ~2亿 -

线上灰度一周后，服务器从8台降到3台，成本降了62%，P99稳定在60ms以内。唯一需要注意的是慢查询会拖累整个事件循环——某次DBA加索引失败导致一条SQL跑了3秒，整个服务的P99瞬间飙到3s。所以异步服务里SQL超时和熔断必须做，我们用asyncio.wait_for给每个DB查询包了500ms超时。

七、总结

asyncio不是银弹，但对IO密集型Web API来说，收益极其明显。关键三点：

  1. 无依赖的IO调用必须并发，asyncio.gather是最简单的武器；
  2. 连接池、Session一定要全局复用，否则异步的优势会被创建开销吃掉；
  3. 任何同步阻塞都是毒药，DB慢查询、CPU密集、第三方同步库，都要隔离处理。

如果你的接口QPS卡在几百上不去，CPU又没跑满，先别急着加机器，看看是不是同步IO在拖后腿。