一、问题背景：一个拖垮全站的慢接口

我们的订单详情接口 /api/v1/orders/{id} 需要聚合三个下游服务的数据：订单基础信息（MySQL）、用户信息（Redis）、物流轨迹（第三方HTTP）。上线半年一直用 requests 库同步调用，平时并发不高没暴露问题。

大促压测时发现：500并发下，该接口P99延迟达到2.8秒，QPS只有120，直接拖垮了网关。用 cProfile 一分析，发现 73% 的时间花在 requests.get 的阻塞等待上——每次HTTP调用要等DNS解析、TCP握手、TLS协商、响应体下载，而CPU是空闲的。

二、环境与版本

Python: 3.10.11 (CPython, 启用--with-lto优化)
框架: Flask 2.3.2 (Werkzeug 2.3.6)
异步客户端: httpx 0.24.1
压测工具: wrk 4.2.0 (单线程, 4个连接)
服务器: 4核8G 云服务器 (Intel Xeon 8374C, 共享型)
操作系统: Ubuntu 22.04 LTS

注意：Flask是同步框架，不能直接在view函数里 await。所以方案是：用 asyncio.run() 在同步view中跑异步任务，用 httpx.AsyncClient 替代 requests。

三、方案设计：织一张异步的网

核心思路是将三个独立的下游IO从串行改为并发。改造前伪代码：

def get_order_detail(order_id):
    db_data = requests.get(f"http://db-service/orders/{order_id}").json()  # 200ms
    user_data = requests.get(f"http://user-service/users/{db_data['uid']}").json()  # 150ms
    logistics = requests.get(f"http://logistics-service/tracks/{order_id}").json()  # 300ms
    return merge(db_data, user_data, logistics)

总耗时 ≈ 200 + 150 + 300 = 650ms（串行）。

改造后，三个请求并发发出，总耗时 ≈ max(200, 150, 300) = 300ms。但这只是第一步——更重要的是用异步客户端复用TCP连接，减少握手开销。

四、核心实现：asyncio.gather + Semaphore限流

先看关键代码。注意我加了 asyncio.Semaphore(20) 限流，防止瞬间打满下游服务：

import asyncio
import httpx
from functools import lru_cache

# 全局复用AsyncClient，避免每个请求重新建连
@lru_cache(maxsize=1)
def get_async_client() -> httpx.AsyncClient:
    return httpx.AsyncClient(
        timeout=httpx.Timeout(2.0, connect=1.0),  # 总超时2秒，连接超时1秒
        limits=httpx.Limits(max_connections=50, max_keepalive_connections=20),
        headers={"X-Client": "order-api-async"},
    )

# 信号量，控制并发数不超过20
_semaphore = asyncio.Semaphore(20)

async def fetch_json(client: httpx.AsyncClient, url: str) -> dict:
    async with _semaphore:
        try:
            resp = await client.get(url)
            resp.raise_for_status()
            return resp.json()
        except httpx.TimeoutException:
            return {"_error": "timeout", "_url": url}
        except httpx.HTTPStatusError as e:
            return {"_error": f"http_{e.response.status_code}", "_url": url}

async def get_order_async(order_id: str) -> dict:
    client = get_async_client()
    # 并发发起三个请求，asyncio.gather保证等最慢的一个
    results = await asyncio.gather(
        fetch_json(client, f"http://db-service/orders/{order_id}"),
        fetch_json(client, f"http://user-service/users/{order_id}"),  # 实际应从db结果取uid，这里简化
        fetch_json(client, f"http://logistics-service/tracks/{order_id}"),
        return_exceptions=False  # 不强抛，由fetch_json内部捕获
    )
    # 合并逻辑（省略字段校验）
    merged = {}
    for r in results:
        if "_error" not in r:
            merged.update(r)
    return merged

# 在Flask同步view中调用
def get_order_detail(order_id: str):
    try:
        return asyncio.run(get_order_async(order_id))
    except RuntimeError as e:
        # 处理event loop冲突（见踩坑部分）
        print(f"[WARN] asyncio.run failed: {e}, fallback to sync")
        return fallback_sync(order_id)

改造后视图函数几乎没变，只是把 requests 换成 asyncio.run 包裹的异步函数。但要注意生产环境不要用 asyncio.run 在同步view里高频调用——它每次创建新EventLoop，开销大。更好的方案是全局维护一个EventLoop（见踩坑部分）。

五、踩坑与优化：EventLoop地狱与超时玄学

这里必须吐槽几个坑，全是血泪教训：

坑1：asyncio.run 在Flask多线程下崩溃
Flask默认threaded=True，每请求一个线程。如果两个线程同时调用 asyncio.run()，会报 RuntimeError: asyncio.run() cannot be called from a running event loop。解决方案：全局只创建一个EventLoop，用 loop.run_until_complete() 提交任务：

# 全局单例loop
_loop = asyncio.new_event_loop()
asyncio.set_event_loop(_loop)

def run_async(coro):
    return _loop.run_until_complete(coro)

坑2：httpx超时参数别乱设
我一开始只设了 timeout=2.0，结果发现连接池满了后，排队等待时间也计入2秒超时，导致大量 TimeoutException。正确做法是分开设置：httpx.Timeout(2.0, connect=1.0, pool=0.5)——pool超时是等待空闲连接的时限，设大点（比如5秒）更合理。

坑3：DNS解析在异步中依然阻塞
httpx默认用 loop.getaddrinfo 执行DNS解析，但它是线程池实现的，高并发下线程池默认32会成瓶颈。解决：用 httpx.AsyncClient(trust_env=False) 并预解析IP，或用 uvloop 替代默认EventLoop。

坑4：下游服务突然变慢，全部超时怎么办？
我在 fetch_json 里捕获了所有异常返回 _error 标记，但合并逻辑里如果三个都失败，返回空dict会导致前端异常。所以加了fallback：如果 results 全是error，就返回一个包装的错误结构。

六、效果数据：不看广告看疗效

压测配置：wrk -t4 -c500 -d30s http://localhost:8080/api/v1/orders/12345

指标 改造前 (requests) 改造后 (asyncio+httpx) 提升
平均延迟 1.2s 180ms 6.7x
P99延迟 2.8s 450ms 6.2x
QPS 120 850 7.1x
错误率 5.2% (超时) 0.3% (仅5xx) 平稳
CPU使用率 35% 60% 略升但合理

为什么延迟降了但CPU升了？ 因为原来线程阻塞时CPU空闲，现在线程等待期间EventLoop在轮询socket，CPU利用率上去了。但注意：4核跑到60%说明压力测试机还有余量，真实环境建议加 uvloop 后单worker能扛更多。

连接复用效果：改造前每请求3次TCP握手（总共500并发要1500个连接），改造后连接池复用，实际活跃连接数稳定在20-30之间，DB服务和物流服务的负载也降了。

七、总结：什么时候值得用asyncio？

这次改造让我重新理解了“异步不是银弹”。如果瓶颈在CPU计算，asyncio没用；如果瓶颈在数据库查询，你需要的是连接池+索引优化；但如果你有大量HTTP调用的IO密集体，asyncio+httpx能带来数量级的提升。

关键教训：
- 同步Flask + asyncio是可以的，但别用 asyncio.run，要全局EventLoop。
- 连接池参数必须调优：max_connections 设为下游能承受的极限值，max_keepalive_connections 设为并发数的1/3。
- 超时设置要分层：连接超时、读取超时、池等待超时分开设。
- 压测用wrk，不要用ab——wrk支持并发连接复用，更接近真实场景。

最后，如果你用的是FastAPI（原生异步框架），直接写 async def 视图函数即可，代码更简洁。但Flask存量项目，按我上面的方法改，40行代码就能救活一个服务。有问题欢迎评论区交流。