一、问题背景：同步阻塞如何拖垮整个服务

我们有个内部业务报表系统，前端需要聚合展示订单、库存、用户三个微服务的数据。最初用Flask+requests实现，代码简单，但在压测时暴露严重问题：

  • 300并发下：P95延迟2.3s，P99直接飙到4.5s
  • 数据库连接池：30个连接全部被占满，其他服务超时重试
  • CPU使用率：仅32%，说明瓶颈在IO等待而非计算

核心代码长这样（伪代码描述），三个串行请求：

# 重构前的同步版本（简化）
def aggregate(user_id):
    orders = requests.get(f"http://order-svc/{user_id}", timeout=2).json()
    inventory = requests.get(f"http://inv-svc/{user_id}", timeout=2).json()
    profile = requests.get(f"http://user-svc/{user_id}", timeout=2).json()
    return {"orders": orders, "inventory": inventory, "profile": profile}

每个请求平均耗时600ms，串行执行就是1.8s。但问题是这三个接口互相独立，完全可以并发请求。

二、环境与版本：为什么选asyncio而不是其他方案

生产环境是 Python 3.11.4（Linux容器，4核8G），Flask 2.3.2，部署在K8s中。重构时对比过三种方案：

方案 优点 缺点
线程池（ThreadPoolExecutor） 改动小 线程切换开销，GIL限制，连接池管理复杂
协程（asyncio+httpx） 轻量，IO密集最优 需要重构代码，调试稍难
消息队列异步化 彻底解耦 改动太大，前端要轮询，不适合当前场景

最终选择 asyncio + httpx 0.25.1，理由：Python 3.11的asyncio性能相比3.8提升约30%，httpx支持原生async/await语法，且与requests接口类似，迁移成本可控。

三、方案设计：把串行改成协程并发

重构目标明确：将三个独立API请求从串行改为并发。设计如下：

  1. 用asyncio.gather并发发起三个请求
  2. 设置超时：总超时2.5s，单个请求1.5s
  3. 复用连接：使用httpx.AsyncClient作为长连接池
# 重构后的异步版本（核心实现）
import asyncio
import httpx
from flask import Flask, jsonify

app = Flask(__name__)

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

async def fetch_order(user_id):
    resp = await _client.get(f"http://order-svc/{user_id}")
    return resp.json()

async def fetch_inventory(user_id):
    resp = await _client.get(f"http://inv-svc/{user_id}")
    return resp.json()

async def fetch_profile(user_id):
    resp = await _client.get(f"http://user-svc/{user_id}")
    return resp.json()

async def aggregate_async(user_id):
    # 核心：gather并发，而不是await串行
    results = await asyncio.gather(
        fetch_order(user_id),
        fetch_inventory(user_id),
        fetch_profile(user_id),
        return_exceptions=True  # 防止一个失败拖垮全部
    )
    # 简单异常处理
    return {
        "orders": results[0] if not isinstance(results[0], Exception) else None,
        "inventory": results[1] if not isinstance(results[1], Exception) else None,
        "profile": results[2] if not isinstance(results[2], Exception) else None
    }

@app.route("/api/v1/user//agg")
def aggregate_endpoint(user_id):
    # 同步Flask中运行协程
    loop = asyncio.new_event_loop()
    try:
        asyncio.set_event_loop(loop)
        data = loop.run_until_complete(aggregate_async(user_id))
        return jsonify(data)
    finally:
        loop.close()

这里有个关键点：Flask是同步框架，不能直接await。所以每次请求创建一个临时event loop。但这样每次创建/销毁loop有开销，后面在踩坑部分我会说这个问题的优化方案。

四、踩坑与优化：三个被忽视的细节

坑1：Event Loop策略与临时loop的坑

Python 3.11在Linux默认使用SelectorEventLoop，但每次请求新建loop会导致fd泄漏。压测时发现连接数不断上涨。优化方案：全局复用event loop（见下方代码）。但要注意Flask多线程模式下，全局loop不安全。如果必须线程安全，可以用loop.run_in_executor配合线程局部存储。

坑2：超时控制必须精细

最初我设置timeout=2.0统一超时，结果发现如果order服务2秒无响应，整个gather要等满2秒才返回。改为三个超时分层：

  • 连接超时：1.0s（快速失败）
  • 读取超时：1.5s（单请求最大等待）
  • 总等待：用asyncio.wait_for包裹gather，设置2.5s总上限

坑3：连接池复用提升30%性能

如果每次请求都新建AsyncClient，握手开销巨大。改为模块级全局_client，压测后发现首字节时间从45ms降到12ms。但注意生产环境要加max_keepalive_connections限制，否则空闲连接耗尽。

优化后的完整代码：

# 优化后的最终版本：全局loop + 精细超时
import asyncio
import httpx
from flask import Flask, jsonify

app = Flask(__name__)

# 全局事件循环（单线程部署场景下安全）
_loop = asyncio.new_event_loop()
asyncio.set_event_loop(_loop)

_client = httpx.AsyncClient(
    timeout=httpx.Timeout(connect=1.0, read=1.5, write=1.0, pool=1.0),
    limits=httpx.Limits(max_connections=100, max_keepalive_connections=20)
)

async def fetch_service(svc_name, user_id):
    url = f"http://{svc_name}-svc/{user_id}"
    resp = await _client.get(url)
    return resp.json()

async def aggregate_async(user_id):
    # 用wait_for控制总超时
    try:
        results = await asyncio.wait_for(
            asyncio.gather(
                fetch_service("order", user_id),
                fetch_service("inv", user_id),
                fetch_service("user", user_id),
                return_exceptions=True
            ),
            timeout=2.5
        )
        return {
            "orders": results[0] if not isinstance(results[0], Exception) else None,
            "inventory": results[1] if not isinstance(results[1], Exception) else None,
            "profile": results[2] if not isinstance(results[2], Exception) else None
        }
    except asyncio.TimeoutError:
        return {"error": "aggregate timeout"}, 504

@app.route("/api/v1/user//agg")
def aggregate_endpoint(user_id):
    # 直接复用全局loop，避免重复创建
    data = _loop.run_until_complete(aggregate_async(user_id))
    return jsonify(data)

# 应用退出时清理
@app.teardown_appcontext
def close_loop(exception=None):
    # 实际生产建议用信号处理优雅关闭
    pass

五、效果数据：从180 RPS到1030 RPS

用locust 2.20.1压测，环境：4核8G容器，三个mock服务部署在本地（模拟真实网络延迟600ms）。

指标 同步版本 异步版本 提升
P50延迟 1.8s 0.35s 5.1x
P95延迟 2.3s 0.4s 5.7x
P99延迟 4.5s 0.7s 6.4x
吞吐量（RPS） 180 1030 5.7x
CPU使用率 32% 45% 合理上升
数据库连接占用 30/30 8/30 释放75%

还有一个重要观察：错误率从2.1%降至0.3%。原因在于超时控制更精确，不会因为一个服务卡住导致整个请求超时。

六、总结：什么时候该用asyncio？

这次重构的核心收益不是「用了异步」这个动作，而是消除了串行等待。如果你遇到以下情况，建议考虑asyncio：

  • IO密集型：大量HTTP/DB/文件操作，CPU占用低于50%
  • 独立请求多：一个接口需要调用多个外部服务，且互相无依赖
  • 高并发场景：单机RPS超过200，且P95延迟敏感

但要注意：CPU密集型场景别用asyncio，那是多进程的领域。另外，如果团队不熟悉协程，调试成本可能高于性能收益——建议先用concurrent.futures.ThreadPoolExecutor做小规模试点，性能不够再上asyncio。

最后提一句：Python 3.12/3.13的asyncio还在持续优化，未来协程调度开销会更低。但核心思想不变——让IO等待变成可挂起的协程调度，这是Python异步编程的精髓。