asyncio重构Flask接口：QPS从120提升到2100的优化记录

# before_app.py
import requests
import pymysql
from flask import Flask, jsonify, request

app = Flask(__name__)

DB = pymysql.connect(host="10.0.0.10", user="app", password="xxx", database="user")

@app.route("/api/v1/user/profile")
def profile():
    uid = request.args.get("user_id")
    if not uid:
        return jsonify({"code": 400, "msg": "missing user_id"}), 400

    # 1. MySQL 串行
    with DB.cursor() as cur:
        cur.execute("SELECT nickname, level FROM user WHERE id=%s", (uid,))
        row = cur.fetchone()
        if not row:
            return jsonify({"code": 404, "msg": "user not found"}), 404
        base = {"nickname": row[0], "level": row[1]}

    # 2. 三个HTTP串行
    tags = requests.get(f"http://user-center/tags?uid={uid}", timeout=1).json()
    risk = requests.get(f"http://risk/level?uid={uid}", timeout=1).json()
    orders = requests.get(f"http://order/recent?uid={uid}", timeout=1).json()

    return jsonify({
        "code": 0,
        "data": {**base, "tags": tags, "risk": risk, "orders": orders}
    })

if __name__ == "__main__":
    app.run(host="0.0.0.0", port=8000)

# after_app.py
import asyncio
import aiohttp
import aiomysql
from fastapi import FastAPI, Query, HTTPException
from contextlib import asynccontextmanager

app = FastAPI()

HTTP_TIMEOUT = 0.8          # 下游HTTP超时
DB_TIMEOUT = 0.3            # DB超时
GATHER_TIMEOUT = 1.2        # 整体兜底超时

session: aiohttp.ClientSession = None
pool: aiomysql.Pool = None


@asynccontextmanager
async def lifespan(app: FastAPI):
    global session, pool
    # 全局session，连接池参数按下游数量调优
    connector = aiohttp.TCPConnector(limit=200, limit_per_host=50, ttl_dns_cache=300)
    session = aiohttp.ClientSession(connector=connector, timeout=aiohttp.ClientTimeout(total=HTTP_TIMEOUT))
    pool = await aiomysql.create_pool(
        host="10.0.0.10", user="app", password="xxx", db="user",
        minsize=5, maxsize=20, pool_recycle=1800, autocommit=True
    )
    yield
    await session.close()
    pool.close()
    await pool.wait_closed()


app.router.lifespan_context = 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_base(uid: str):
    async with pool.acquire() as conn:
        async with conn.cursor() as cur:
            await cur.execute("SELECT nickname, level FROM user WHERE id=%s", (uid,))
            row = await cur.fetchone()
            return {"nickname": row[0], "level": row[1]} if row else None


async def safe_fetch(url: str, default):
    """单个下游失败不影响整体，返回默认值"""
    try:
        return await asyncio.wait_for(fetch_json(url), timeout=HTTP_TIMEOUT)
    except Exception:
        return default


@app.get("/api/v1/user/profile")
async def profile(user_id: str = Query(...)):
    # 风控必须成功，单独拿
    try:
        risk = await asyncio.wait_for(
            fetch_json(f"http://risk/level?uid={user_id}"), timeout=HTTP_TIMEOUT
        )
    except Exception:
        raise HTTPException(status_code=503, detail="risk service unavailable")

    # 其余三个并发
    base_task = asyncio.wait_for(get_base(user_id), timeout=DB_TIMEOUT)
    tags_task = safe_fetch(f"http://user-center/tags?uid={user_id}", {})
    orders_task = safe_fetch(f"http://order/recent?uid={user_id}", {"count": 0})

    try:
        base, tags, orders = await asyncio.wait_for(
            asyncio.gather(base_task, tags_task, orders_task),
            timeout=GATHER_TIMEOUT
        )
    except asyncio.TimeoutError:
        raise HTTPException(status_code=504, detail="upstream timeout")

    if not base:
        raise HTTPException(status_code=404, detail="user not found")

    return {"code": 0, "data": {**base, "tags": tags, "risk": risk, "orders": orders}}