asyncio重构Flask接口，QPS从120飙到1800的完整记录

# before.py - Flask同步版本
import requests
from flask import Flask, jsonify

app = Flask(__name__)

def fetch_user(uid):
    r = requests.get(f"http://svc-a/user/{uid}", timeout=2)
    return r.json()

def fetch_orders(uid):
    r = requests.get(f"http://svc-b/orders/{uid}", timeout=2)
    return r.json()

def fetch_recommend(uid):
    r = requests.get(f"http://svc-c/recommend/{uid}", timeout=2)
    return r.json()

@app.route("/api/agg/")
def aggregate(uid):
    # 三个请求串行，每个平均300ms，总耗时约900ms
    user = fetch_user(uid)
    orders = fetch_orders(uid)
    reco = fetch_recommend(uid)
    return jsonify({"user": user, "orders": orders, "reco": reco})

# after.py - asyncio + httpx 并发版本
import asyncio
import httpx
from flask import Flask, jsonify

app = Flask(__name__)
# 全局信号量，限制同时最多20个协程在跑IO
_semaphore = asyncio.Semaphore(20)
_client = None

def get_client():
    global _client
    if _client is None or _client.is_closed:
        # httpx.AsyncClient内部有连接池，必须全局复用
        _client = httpx.AsyncClient(
            limits=httpx.Limits(max_connections=50, max_keepalive_connections=20),
            timeout=httpx.Timeout(2.0, connect=0.5),
            http2=True  # 内部服务支持HTTP2，省握手时间
        )
    return _client

async def fetch_with_limit(client, url):
    async with _semaphore:
        try:
            # wait_for替代3.10的timeout，超时抛asyncio.TimeoutError
            resp = await asyncio.wait_for(client.get(url), timeout=2.0)
            return resp.json()
        except (httpx.TimeoutException, asyncio.TimeoutError, httpx.HTTPError) as e:
            # 失败快速返回降级数据，不让聚合接口整体失败
            return {"error": str(e), "data": None}

async def fetch_all(uid):
    client = get_client()
    urls = [
        f"http://svc-a/user/{uid}",
        f"http://svc-b/orders/{uid}",
        f"http://svc-c/recommend/{uid}",
    ]
    # gather并发，return_exceptions=True防止一个挂了全挂
    results = await asyncio.gather(
        *(fetch_with_limit(client, url) for url in urls),
        return_exceptions=True
    )
    return results

@app.route("/api/agg/")
def aggregate(uid):
    # 每个请求独立事件循环，注意：这里不能复用全局loop
    try:
        user, orders, reco = asyncio.run(fetch_all(uid))
    except Exception as e:
        return jsonify({"error": "aggregate failed"}), 500
    return jsonify({"user": user, "orders": orders, "reco": reco})

if __name__ == "__main__":
    # 生产用uvicorn跑，不用app.run
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000, workers=4)

# 错误示范：同步DB查询在协程里
async def bad_sql():
    cursor = pymysql.connect(...)  # 阻塞事件循环！
    cursor.execute("SELECT ...")

# 正确做法：丢到线程池
async def good_sql():
    result = await asyncio.to_thread(sync_db_query, sql)