asyncio重构 Flask 订单查询接口：从 1.2s 到 180ms 的改造记录

# before.py
import requests
from flask import Flask, jsonify

app = Flask(__name__)

SESSION = requests.Session()
SESSION.mount("http://", requests.adapters.HTTPAdapter(pool_connections=20, pool_maxsize=50))

USER_URL = "http://user-svc:8001/user/{}"
STOCK_URL = "http://stock-svc:8002/stock/{}"
COUPON_URL = "http://coupon-svc:8003/coupon/{}"
LOGISTICS_URL = "http://logistics-svc:8004/logistics/{}"
RISK_URL = "http://risk-svc:8005/risk/{}"

@app.route("/api/order/")
def get_order(order_id):
    user = SESSION.get(USER_URL.format(order_id), timeout=0.3).json()
    stock = SESSION.get(STOCK_URL.format(order_id), timeout=0.3).json()
    coupon = SESSION.get(COUPON_URL.format(order_id), timeout=0.3).json()
    logistics = SESSION.get(LOGISTICS_URL.format(order_id), timeout=0.4).json()
    risk = SESSION.get(RISK_URL.format(order_id), timeout=0.4).json()

    return jsonify({
        "order_id": order_id,
        "user": user,
        "stock": stock,
        "coupon": coupon,
        "logistics": logistics,
        "risk": risk,
    })

# after.py
import asyncio
import threading
import httpx
from flask import Flask, jsonify

app = Flask(__name__)

# ---- 常驻事件循环，应用启动时初始化 ----
_loop = asyncio.new_event_loop()
_loop_thread = threading.Thread(target=_loop.run_forever, daemon=True)
_loop_thread.start()

# ---- 全局复用的 AsyncClient，连接池只建一次 ----
_client: httpx.AsyncClient = None

def _init_client():
    global _client
    _client = httpx.AsyncClient(
        timeout=httpx.Timeout(connect=0.2, read=0.4, write=0.4, pool=0.5),
        limits=httpx.Limits(max_connections=200, max_keepalive_connections=50),
        http2=False,  # 下游都是 HTTP/1.1，关掉省握手
    )

_init_client()

URLS = {
    "user": "http://user-svc:8001/user/{}",
    "stock": "http://stock-svc:8002/stock/{}",
    "coupon": "http://coupon-svc:8003/coupon/{}",
    "logistics": "http://logistics-svc:8004/logistics/{}",
    "risk": "http://risk-svc:8005/risk/{}",
}

# 每个下游单独超时，避免慢服务拖垮整体
TIMEOUTS = {"user": 0.3, "stock": 0.3, "coupon": 0.3, "logistics": 0.4, "risk": 0.35}
KEYS = list(URLS.keys())

async def _fetch_one(key: str, order_id: str):
    try:
        r = await _client.get(
            URLS[key].format(order_id),
            timeout=TIMEOUTS[key],
        )
        r.raise_for_status()
        return key, r.json()
    except Exception as e:
        # 降级：单点失败返回 None，不抛给上层
        return key, {"__error__": str(e)}

async def _fetch_all(order_id: str):
    results = await asyncio.gather(
        *[_fetch_one(k, order_id) for k in KEYS],
        return_exceptions=False,
    )
    return dict(results)

@app.route("/api/order/")
def get_order(order_id):
    # 关键：把协程丢进常驻 loop，阻塞等结果
    fut = asyncio.run_coroutine_threadsafe(_fetch_all(order_id), _loop)
    try:
        data = fut.result(timeout=1.0)  # 兜底超时，防止 loop 卡死
    except Exception:
        return jsonify({"order_id": order_id, "error": "upstream_timeout"}), 504

    return jsonify({"order_id": order_id, **data})

# ---- 进程退出时清理 ----
@app.teardown_appcontext
def _noop(exc):
    pass

import atexit
@atexit.register
def _shutdown():
    if _client:
        asyncio.run_coroutine_threadsafe(_client.aclose(), _loop)
    _loop.call_soon_threadsafe(_loop.stop)

gunicorn -k gthread -w 4 --threads 8 --worker-connections 1000 \
  --timeout 5 --graceful-timeout 3 app:app