asyncio重构Flask API：请求耗时从980ms降至210ms

1. 问题背景：同步阻塞把一个“查询接口”变成了“慢查询”

上个月接手一个订单服务，有个/api/v1/orders/export接口，逻辑很简单：接收订单ID列表，然后依次调用用户服务、商品服务、库存服务，最后组装数据返回。三个上游都是内部RPC，单个延迟都在200-300ms之间。

上线后监控发现：P95延迟980ms，P99直接飙到1.5s。更难受的是，这台2C4G的容器在高峰期CPU只有30%，但Tomcat线程池被打满，大量请求排队。典型症状：线程在等IO，CPU在闲着。

反手看一眼代码，问题一目了然：

# before.py - 同步串行调用（简化版）
def get_orders(ids):
    result = []
    for oid in ids:
        user = requests.get(f"http://user-svc/users/{oid}").json()
        prod = requests.get(f"http://prod-svc/products/{oid}").json()
        stock = requests.get(f"http://stock-svc/stocks/{oid}").json()
        result.append({...})
    return result

三个requests.get是纯串行阻塞，每个都要等TCP握手+响应体返回。一个订单要等三个RTT，10个订单就是30个RTT。这不是代码逻辑问题，是IO模型的问题。

2. 环境与版本：Python 3.10 + FastAPI + httpx

改造前先确认环境，这里直接给出我用的版本组合，避免后面踩坑：

Python 3.10.8
fastapi 0.95.1 (用于提供API，替代原Flask)
uvicorn 0.21.1 (ASGI服务器)
httpx 0.24.1 (异步HTTP客户端)
uvloop 0.17.0 (可选，替换asyncio事件循环)

注意：原接口是Flask写的，但我直接把服务换成了FastAPI。原因有二：一是Flask本身不支持异步视图函数，强行用asyncio.run()会阻塞事件循环；二是FastAPI原生支持async def，配合uvicorn能直接跑异步。如果你坚持用Flask，可以用asyncio.run()包一层，但那样并发效果会打折，后面会解释。

3. 方案设计：协程并发 + 连接复用 + 信号量限流

核心思路很简单：把三个独立的HTTP调用改为asyncio.gather并发执行。但有几个细节必须处理：

1. 连接复用：requests每次新建TCP连接，浪费握手时间。改用httpx.AsyncClient，底层是连接池，默认复用连接。
2. 并发上限：如果一次导出500个订单，每个订单3个并发，瞬间打出1500个并发请求，上游会挂。必须用asyncio.Semaphore限制总并发数。
3. 超时控制：httpx不设置超时默认5s，但上游抖动时，一个慢请求会拖慢整个gather。给每个请求单独加timeout。

设计后的流程：

订单ID列表 → 按批处理（每批50个） → 每批内部用gather并发三个调用 → 组装结果

批大小和信号量都是可调参数。我最终用的是：Semaphore(50)，每批50个订单，实际并发峰值=50*3=150个连接。

4. 核心实现：从同步到异步的完整改造

先看改造后的核心代码。注意这里用了async def，并且所有HTTP调用都走httpx.AsyncClient。

# after.py - 异步并发版本
import asyncio
import httpx
from fastapi import FastAPI

app = FastAPI()
# 全局连接池，复用TCP连接，减少握手开销
client = httpx.AsyncClient(timeout=10.0, limits=httpx.Limits(max_connections=200))

# 信号量控制并发，防止打爆上游
sem = asyncio.Semaphore(50)

async def fetch_one(oid: str):
    """并发获取单个订单的三个数据源"""
    async with sem:  # 限制同时进入的协程数
        # 三个独立请求并发执行，总耗时=max(三个耗时)，而非sum
        user_task = client.get(f"http://user-svc/users/{oid}")
        prod_task = client.get(f"http://prod-svc/products/{oid}")
        stock_task = client.get(f"http://stock-svc/stocks/{oid}")
        user_resp, prod_resp, stock_resp = await asyncio.gather(
            user_task, prod_task, stock_task, 
            return_exceptions=True  # 容错：一个失败不影响其他
        )
        # 注意：这里简化了JSON解析和错误处理，生产环境需要细化
        return {
            "oid": oid,
            "user": user_resp.json() if user_resp.status_code == 200 else None,
            "prod": prod_resp.json() if prod_resp.status_code == 200 else None,
            "stock": stock_resp.json() if stock_resp.status_code == 200 else None,
        }

async def process_batch(oids: list):
    """每批并发处理，批大小由调用方控制"""
    tasks = [fetch_one(oid) for oid in oids]
    results = await asyncio.gather(*tasks)
    return results

@app.get("/api/v1/orders/export")
async def export_orders(ids: str):
    """ids: 逗号分隔的订单ID列表"""
    oid_list = ids.split(",")
    # 按50个一批，避免一次性创建太多协程
    batch_size = 50
    all_results = []
    for i in range(0, len(oid_list), batch_size):
        batch = oid_list[i:i+batch_size]
        results = await process_batch(batch)
        all_results.extend(results)
    return {"data": all_results, "total": len(all_results)}

对比一下before和after的时间复杂度：

• before：串行调用，耗时 = N × (t_user + t_prod + t_stock)，N是订单数。
• after：每批内并发，单批耗时 ≈ max(t_user, t_prod, t_stock) + 调度开销。批间串行，但批大小可调。

5. 踩坑与优化：不是所有同步代码都能“无脑改async”

我踩过的坑，逐个说，这些细节比核心代码更值钱。

坑1：requests库不能在协程里用。
requests是纯同步阻塞库，如果直接放在async def里调用，它依然会阻塞线程，协程直接失效。必须换成httpx或aiohttp。我选httpx是因为它API和requests几乎一样，迁移成本低，而且支持HTTP/2（虽然内部服务用的是HTTP/1.1）。

坑2：asyncio.Semaphore要放在全局，不能放函数内部。
如果每次请求都创建新的Semaphore(50)，那相当于没用限流，多个请求同时进来，每个请求内部有50个并发，总量还是爆。我的做法是全局只创建一个sem，所有协程共享。

坑3：uvicorn必须设置--workers 1，否则多进程下asyncio混用会出问题。
如果你用uvicorn main:app --workers 4，会启动4个进程，每个进程有独立的事件循环和连接池。这本身没问题，但需要注意：如果用了全局client对象，每个进程各有一份，连接池隔离。我实测4个worker时，吞吐确实翻倍，但P99延迟不稳定，因为进程间负载不均。最终我保留1个worker，靠协程本身的并发能力就够用了，还省内存。

坑4：uvloop替换默认事件循环，性能有提升但不是必须。
在Linux上，把asyncio的事件循环换成uvloop，能减少EPoll调用的开销。实测数据：纯异步IO场景，延迟降低约8-12%。代码如下：

import uvloop
asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())

注意：必须在uvicorn.run()之前调用。Windows不支持uvloop，生产环境是Linux所以没问题。

坑5：return_exceptions=True必须加。
如果gather里某个请求超时抛异常，不加这个参数，整个gather都会取消，导致其他两个正常请求的结果也丢失。加了之后，异常会被封装成对象，需要自己判断类型。我上面的代码只是简化版，生产环境里我用try/except包裹更细粒度。

6. 效果数据：压测结果与资源占用对比

使用locust做压测，模拟200并发用户，持续5分钟，数据如下：

数据说明：异步版的延迟曲线更平滑，没有长尾。同步版在200并发下线程池（默认200线程）打满，大量请求等待排队。异步版只用1个进程1个线程，却扛住了同样的并发。

额外发现：连接池复用带来的收益比并发本身更大。我把httpx的max_connections从50调到200，P95又降了30ms。因为复用TCP连接省去了每次三次握手的时间，而在内网环境下，RTT本身接近0.1ms，但连接建立要1ms左右，积少成多。

7. 总结：什么场景适合asyncio，什么场景别硬用

这次改造的核心收益来自IO密集型的多路复用。如果你的API里有多个独立的HTTP/RPC调用，且之间没有依赖关系，用asyncio.gather并发是最高性价比的优化。但如果你的业务是CPU密集（比如图像处理、大量计算），asyncio帮不上忙，应该用多进程或concurrent.futures.ProcessPoolExecutor。

另外，现有代码迁移到asyncio的成本要评估：
- 如果上游库都是同步的（比如requests、pymysql），必须换异步驱动（httpx、asyncpg），这部分改动量不小。
- 如果业务逻辑复杂，涉及多表事务、复杂计算，异步会让代码可读性下降，反而得不偿失。

我这次的改造范围控制在API层，没有动底层数据访问，所以风险可控。最后提醒一句：先压测定位瓶颈，确认是IO阻塞再上asyncio。如果你接口本身只调一次数据库，延迟低，那改异步收益不大，别为了炫技而重构。