Python异步重构：用asyncio把Web API响应时间从900ms压到150ms

def get_order_detail(order_id):
    logistics = requests.get(f"http://logistics/{order_id}", timeout=2).json()
    price = requests.get(f"http://price/{order_id}", timeout=2).json()
    stock = requests.get(f"http://inventory/{order_id}", timeout=2).json()
    return merge(logistics, price, stock)

[FastAPI View] → asyncio.run(main()) → create_task ×3
                                    ↓
                    Semaphore(10) 控制并发
                    ↓              ↓              ↓
               物流服务调用    价格服务调用    库存服务调用
                    ↓              ↓              ↓
                各自try/except → 合并结果 → 返回

# 同步版本：三个请求串行，总共耗时≈900ms
import requests
from fastapi import FastAPI

app = FastAPI()

def fetch_logistics(order_id: str) -> dict:
    resp = requests.get(f"http://logistics-service/api/v1/{order_id}",
                        timeout=2)
    return resp.json()

def fetch_price(order_id: str) -> dict:
    resp = requests.get(f"http://price-service/api/v1/{order_id}",
                        timeout=2)
    return resp.json()

def fetch_stock(order_id: str) -> dict:
    resp = requests.get(f"http://stock-service/api/v1/{order_id}",
                        timeout=2)
    return resp.json()

@app.get("/orders/{order_id}")
def get_order(order_id: str):
    # 串行调用，耗时线性叠加
    logistics = fetch_logistics(order_id)
    price = fetch_price(order_id)
    stock = fetch_stock(order_id)
    return {
        "order_id": order_id,
        "logistics": logistics,
        "price": price,
        "stock": stock
    }

# 异步版本：三个请求并发，总耗时≈最大单个耗时(300ms)
import asyncio
import httpx
from fastapi import FastAPI

app = FastAPI()

# 全局连接池，复用TCP连接
client = httpx.AsyncClient(
    timeout=httpx.Timeout(connect=0.5, read=2.0, write=1.0),
    limits=httpx.Limits(max_keepalive_connections=20,
                        max_connections=100)
)

async def fetch_with_semaphore(sem: asyncio.Semaphore,
                               url: str) -> dict | None:
    async with sem:  # 控制并发信号量
        try:
            resp = await client.get(url)
            return resp.json()
        except httpx.TimeoutException:
            return {"error": "timeout"}
        except Exception as exc:
            return {"error": str(exc)}

async def get_order_async(order_id: str) -> dict:
    sem = asyncio.Semaphore(10)  # 限制同时最多10个对外请求
    # 创建三个并发任务
    tasks = [
        asyncio.create_task(
            fetch_with_semaphore(sem,
                f"http://logistics-service/api/v1/{order_id}")),
        asyncio.create_task(
            fetch_with_semaphore(sem,
                f"http://price-service/api/v1/{order_id}")),
        asyncio.create_task(
            fetch_with_semaphore(sem,
                f"http://stock-service/api/v1/{order_id}")),
    ]
    # wait而不是gather，容忍部分失败
    done, pending = await asyncio.wait(tasks, timeout=3.0,
                                       return_when=asyncio.ALL_COMPLETED)
    # 超时未完成的任务直接取消，防止协程泄漏
    for task in pending:
        task.cancel()
    results = []
    for task in done:
        try:
            results.append(task.result())
        except Exception:
            results.append(None)
    # 确保返回三个结果，缺失的填None
    while len(results) < 3:
        results.append(None)
    return {
        "order_id": order_id,
        "logistics": results[0],
        "price": results[1],
        "stock": results[2]
    }

@app.get("/orders/{order_id}")
async def get_order_async_endpoint(order_id: str):
    # FastAPI原生支持async def，直接await
    return await get_order_async(order_id)

wrk -t16 -c200 -d60s --latency http://localhost:8000/orders/ORD-20231115-001