1. 问题背景：一个聚合接口为何拖垮了整个详情页

公司有一个BFF（Backend For Frontend）服务，其中一个/api/v1/product/detail接口承担了首页和详情页的全部流量。这个接口的逻辑很简单：拿到商品ID后，依次调用三个内部HTTP服务获取基础信息、库存和促销活动，最后组装返回。

压测结果惨不忍睹：

指标 优化前
P50 820ms
P95 1.5s
P99 2.1s
吞吐量 180 req/s

监控面板显示Tomcat线程池被打满（这虽然是个Python服务，但Gunicorn的同步Worker同样致命）。核心原因一目了然：三个下游HTTP请求是串行执行的。每个下游服务本身只要150-250ms，但加在一起就不可接受了。

2. 环境与版本：Python 3.10带来的原生协程红利

这次改造前，团队内部其实对是否引入asyncio有分歧。有人觉得用concurrent.futures.ThreadPoolExecutor把三个请求丢到线程池里就完事了。但考虑到线程切换开销和GIL限制，以及后续要接入WebSocket长连接，我们决定一步到位。

我的验证环境：

Python: 3.10.11 (原生asyncio，无需uvloop)
aiohttp: 3.8.4
asyncpg: 0.27.0 (注意：项目中的Redis读取也用redis.asyncio替换了)
Gunicorn: 20.1.0 (worker_class改为'uvicorn.workers.UvicornWorker')

关键点：Python 3.10的asyncio已经非常成熟，TaskGroup（3.11才有的特性）用不上，所以用asyncio.gather。

3. 方案设计：用协程将串行IO改为并发IO

我们先看一下优化前的同步代码（核心逻辑）：

# 优化前：同步串行调用 (Flask 2.2.5, Gunicorn同步Worker)
import requests
import time

def get_product_detail(product_id: str) -> dict:
    start = time.perf_counter()
    # 1. 调用商品基础服务
    base_info = requests.get(
        f"http://product-service/api/v1/products/{product_id}",
        timeout=0.5
    ).json()
    # 2. 调用库存服务 (依赖商品ID)
    stock_info = requests.get(
        f"http://stock-service/api/v1/stocks/{product_id}",
        timeout=0.5
    ).json()
    # 3. 调用促销服务
    promo_info = requests.get(
        f"http://promo-service/api/v1/promos/{product_id}",
        timeout=0.5
    ).json()
    # 组装返回
    result = {
        "product_id": product_id,
        "name": base_info["name"],
        "stock": stock_info["available"],
        "promo": promo_info.get("discount", 1.0)
    }
    return result

这三个请求之间其实没有数据依赖。它们唯一的共同点是都依赖product_id。所以完全可以用asyncio.gather并发发起。

4. 核心实现：asyncio + aiohttp 重构

改造后，我把整个接口从Flask迁移到了FastAPI（FastAPI 0.100.0，天然支持异步视图函数）。如果你坚持用Flask，可以用asyncio.run()包一层，但那样性能会打折扣，因为Flask的同步WSGI处理会阻塞事件循环。

# 优化后：asyncio并发调用 (FastAPI 0.100.0, uvicorn 0.23.2)
import asyncio
import aiohttp
from fastapi import FastAPI
import asyncpg

app = FastAPI()
# 全局复用连接池
pg_pool = None
semaphore = asyncio.Semaphore(50)  # 限制并发，防止压垮下游

async def fetch_json(session: aiohttp.ClientSession, url: str, timeout: float = 0.8):
    """带超时和信号量的GET请求"""
    async with semaphore:  # 控制并发数
        try:
            async with session.get(url, timeout=aiohttp.ClientTimeout(total=timeout)) as resp:
                if resp.status == 200:
                    return await resp.json()
                else:
                    return None
        except asyncio.TimeoutError:
            return None  # 降级处理

@app.get("/api/v1/product/detail")
async def get_product_detail(product_id: str):
    # 并发请求三个服务
    async with aiohttp.ClientSession() as session:
        tasks = [
            fetch_json(session, f"http://product-service/api/v1/products/{product_id}"),
            fetch_json(session, f"http://stock-service/api/v1/stocks/{product_id}"),
            fetch_json(session, f"http://promo-service/api/v1/promos/{product_id}")
        ]
        base_info, stock_info, promo_info = await asyncio.gather(*tasks)

    # 后续组装，如果有DB操作则用asyncpg
    # 注意：这里故意省略了DB调用，实际项目中用asyncpg.fetchrow
    return {
        "product_id": product_id,
        "name": base_info.get("name") if base_info else "unknown",
        "stock": stock_info.get("available") if stock_info else 0,
        "promo": promo_info.get("discount") if promo_info else 1.0
    }

5. 踩坑与优化：三个必须注意的细节

坑1：Gunicorn的Worker必须换掉。如果你还用gunicorn -w 4 -b :8000 app:app跑Flask，那asyncio一点用都没有。因为同步Worker会阻塞事件循环。我最终用的是：

gunicorn main:app -w 4 -k uvicorn.workers.UvicornWorker --bind 0.0.0.0:8000 --timeout 60

坑2：aiohttp.ClientSession不能每次请求都创建。我在第一版代码里每个请求都async with aiohttp.ClientSession()，结果压测发现性能只提升了30%。原因是创建Session的开销很大（SSL握手、连接池初始化）。正确做法是全局单例：

# app启动时创建
session = aiohttp.ClientSession()
# 关闭时确保释放
@app.on_event("shutdown")
async def close_session():
    await session.close()

坑3：最容易忽略的——asyncio.gather默认不取消其他任务。如果其中一个请求超时（比如促销服务卡了3秒），fetch_json里的asyncio.TimeoutError会让这个任务返回None，但另外两个任务会继续执行。这没问题。但如果你用了asyncio.wait_for包裹整个gather，那超时后所有任务都会被取消，可能丢失已有响应。我这里选择的是每个子任务自己控制超时，而不是整体超时。

6. 性能对比数据：压测结果

用wrk -t8 -c100 -d30s压测，环境是4核8G的Docker容器（跟生产一致）：

指标 优化前 (同步Flask) 优化后 (asyncio+FastAPI) 提升
P50 820ms 96ms 8.5x
P95 1.5s 180ms 8.3x
P99 2.1s 280ms 7.5x
吞吐量 180 req/s 1450 req/s 8.1x
错误率 2.3% 0.1% -

还有一个意外收获：因为asyncio.Semaphore(50)限制了并发，下游服务的负载从之前的每秒3000次请求降到了每秒1500次，它们的CPU使用率反而下降了，GC暂停也变少了。

7. 总结与反思

这次改造最大的收益不是代码层面的，而是思维层面的——Python的异步模型不是银弹，但它非常适合IO密集型任务。如果你的接口里有超过2个无依赖的IO操作（HTTP、DB、Redis），无脑用asyncio就对了。

但也要泼一盆冷水：如果你的服务里CPU密集计算占大头（比如图像处理、加密），那asyncio帮不了你，需要上multiprocessing或C扩展。另外，异步代码的调试成本确实更高，建议把DEBUG级别的日志打开，配合asyncio.get_event_loop().set_debug(True)。

最后留个彩蛋：如果你升级到Python 3.11，把asyncio.gather换成asyncio.TaskGroup，代码会更清晰。但我们在生产环境选择保守，3.10 + 手动gather更稳定。