一、问题背景：一个“慢得离谱”的聚合接口
事情是这样的。我们有个内部服务叫/api/v1/aggregate，职责很简单：接收一个商品ID，同时从三个下游服务拉数据——商品基础信息、库存状态、价格信息，拼装后返回。
原来的实现是Flask + requests，三个下游调用串行执行：
# before: app.py (Flask 2.3.2 + requests 2.31.0)
import requests
from flask import Flask, jsonify
app = Flask(__name__)
@app.route("/api/v1/aggregate/")
def aggregate(item_id):
# 串行调用三个下游
item = requests.get(f"http://item-svc/items/{item_id}", timeout=2).json()
stock = requests.get(f"http://stock-svc/stock/{item_id}", timeout=2).json()
price = requests.get(f"http://price-svc/price/{item_id}", timeout=2).json()
return jsonify({"item": item, "stock": stock, "price": price})
单看代码没毛病，但压测数据很难看：
- QPS: 120（4 worker，gunicorn sync模式）
- P99延迟: 2300ms
- 下游平均响应时间其实只有80ms左右，但串行三次就是240ms起步，加上Python GIL和同步IO等待，worker全被堵死
问题本质很清楚：同步阻塞IO导致worker在等待网络响应期间完全空转。每个请求要等3次RTT，worker数量有限，并发能力被死死卡住。
二、环境与版本
先交代一下环境，避免版本差异导致行为不一致：
| 组件 | 版本 |
|---|---|
| Python | 3.11.6 |
| Flask | 2.3.2（改造后仅保留少量同步接口） |
| aiohttp | 3.9.1 |
| uvloop | 0.19.0 |
| gunicorn | 21.2.0 |
| wrk | 4.2.0（压测工具） |
| 机器 | 4C8G，下游服务同机房 |
Python 3.11的asyncio在性能上有明显优化（TaskGroup、异常组等），所以没有再考虑3.8/3.9。uvloop在3.11上兼容良好，实测比默认事件循环快约20%。
三、方案设计：为什么选asyncio而不是多线程/多进程
摆在面前的选择有几个：
- 多线程 + requests：改造成本低，但GIL下线程切换开销大，且requests的Session不是线程安全的，需要每线程一个Session，连接复用率低。
- 多进程：加worker数量，但内存翻倍，且单请求延迟没改善。
- asyncio + aiohttp：单线程事件循环，IO等待期间可以切换处理其他请求，连接池复用，内存占用低。
我选了方案3。关键设计点：
- 用aiohttp的
ClientSession做全局连接池，而不是每次请求新建连接 - 三个下游用
asyncio.gather并发调用，总延迟从“三次之和”变成“三次最大值” - 给每个下游单独设置超时（
ClientTimeout），避免一个慢下游拖垮整个请求 - Web框架层面：Flask是同步的，不能直接跑协程。我用了
aiohttp自带的Web服务器重写这个接口（也可以用FastAPI，但不想引入额外依赖），保留Flask处理其他同步接口，两者用gunicorn的uvicorn worker或分开部署
实际上我最终是把聚合接口单独拆成一个aiohttp应用，Flask只保留管理类接口。这样职责清晰，也避免在Flask里硬塞async的别扭。
四、核心实现
4.1 连接池与Session管理
第一个坑：一开始我在每个handler里async with aiohttp.ClientSession()，结果压测时连接数暴涨，TIME_WAIT一堆。正确做法是全局复用一个Session，并配置连接池参数。
# after: aggregate_service.py (aiohttp 3.9.1)
import asyncio
import aiohttp
from aiohttp import web, ClientTimeout, TCPConnector
# 全局Session，在app启动时创建
SESSION: aiohttp.ClientSession | None = None
# 连接池配置：总连接数上限、每host上限、DNS缓存
CONNECTOR_LIMIT = 200
CONNECTOR_LIMIT_PER_HOST = 100
DNS_TTL = 300 # 秒
# 单个下游的超时：连接3s，总读取2s
DOWNSTREAM_TIMEOUT = ClientTimeout(total=2.0, connect=0.5, sock_read=1.5)
async def fetch_json(session: aiohttp.ClientSession, url: str) -> dict:
"""带超时和重试的单次下游调用"""
for attempt in range(2): # 简单重试一次
try:
async with session.get(url, timeout=DOWNSTREAM_TIMEOUT) as resp:
resp.raise_for_status()
return await resp.json()
except (aiohttp.ClientError, asyncio.TimeoutError) as e:
if attempt == 1:
raise
await asyncio.sleep(0.05) # 短暂退避
async def aggregate_handler(request: web.Request) -> web.Response:
item_id = request.match_info["item_id"]
s = request.app["session"]
# 三个下游并发调用，gather会等全部完成（或第一个异常）
item, stock, price = await asyncio.gather(
fetch_json(s, f"http://item-svc/items/{item_id}"),
fetch_json(s, f"http://stock-svc/stock/{item_id}"),
fetch_json(s, f"http://price-svc/price/{item_id}"),
return_exceptions=False, # 任一失败直接抛，由上层统一处理
)
return web.json_response({"item": item, "stock": stock, "price": price})
async def on_startup(app: web.Application):
connector = TCPConnector(
limit=CONNECTOR_LIMIT,
limit_per_host=CONNECTOR_LIMIT_PER_HOST,
ttl_dns_cache=DNS_TTL,
enable_cleanup_closed=True, # 清理半关闭连接，防泄漏
)
app["session"] = aiohttp.ClientSession(connector=connector)
async def on_cleanup(app: web.Application):
await app["session"].close()
def create_app() -> web.Application:
app = web.Application()
app.router.add_get("/api/v1/aggregate/{item_id}", aggregate_handler)
app.on_startup.append(on_startup)
app.on_cleanup.append(on_cleanup)
return app
if __name__ == "__main__":
web.run_app(create_app(), host="0.0.0.0", port=8080)
4.2 启动脚本：uvloop + gunicorn
# 用uvloop加速事件循环，4个worker（与CPU核数一致）
gunicorn aggregate_service:create_app \
--worker-class aiohttp.GunicornUVLoopWebWorker \
--workers 4 \
--bind 0.0.0.0:8080 \
--access-logfile - \
--timeout 10
注意worker数不是越多越好。asyncio单worker本身就能扛很高并发，worker数设成CPU核数（4）即可，再多反而增加上下文切换和连接池碎片化。
五、踩坑与优化
坑1：asyncio.gather的异常传播。默认return_exceptions=False时，任一协程抛异常会立即向上抛，但其他未完成的协程不会被取消，会继续在后台跑完，造成资源浪费。如果你的下游允许部分失败，建议用return_exceptions=True然后手动判断；如果要求全部成功，至少在异常处理里显式取消剩余任务。我最终在handler外层加了统一异常处理，返回502。
坑2：超时设置太粗。一开始只设了ClientTimeout(total=2)，结果某个下游TCP握手就卡了1.8秒，剩下0.2秒根本不够读数据。拆成connect=0.5 + sock_read=1.5后，连接阶段的慢下游能被快速识别并重试。
坑3：连接池limit_per_host设太小。默认是0（不限制），但我在第一版设了20，压测到800 QPS时出现大量“连接池等待”，延迟飙升。调到100后消失。经验值：limit_per_host至少设为“预期QPS × 下游平均延迟(秒) × 1.5”。
坑4：没开enable_cleanup_closed。下游偶发主动断连时，aiohttp会留下半关闭连接，长时间运行后连接数缓慢上涨。加上这个参数后稳定了。
优化点：给fetch_json加了轻量重试（仅一次，50ms退避），下游偶发超时导致的502从0.8%降到0.1%。
六、效果数据
压测命令（同机房，下游服务mock为固定80ms延迟）：
wrk -t4 -c200 -d60s http://localhost:8080/api/v1/aggregate/12345
| 指标 | Before (Flask+requests) | After (aiohttp+asyncio) | 变化 |
|---|---|---|---|
| QPS | 120 | 1800 | 15x |
| P50延迟 | 820ms | 95ms | -88% |
| P99延迟 | 2300ms | 180ms | -92% |
| 内存占用（4 worker） | 480MB | 410MB | -15% |
| 下游连接数峰值 | 约900（频繁新建） | 稳定在280左右 | 连接复用 |
QPS没有到理论值（200并发/0.095s≈2100），主要受限于单机4核和下游mock服务的处理能力，但已经远超预期。P99的180ms基本等于“最慢下游的P99 + 少量调度开销”，说明并发调用确实生效了。
七、总结
这次改造的核心就一句话：把“等待”从阻塞变成可调度。同步代码里三个80ms的串行调用，在asyncio下并发后总延迟≈80ms，加上事件循环调度开销，P50做到95ms是合理的。
几点建议给准备做类似改造的同学：
- 不要硬在Flask里塞async，拆成独立aiohttp/FastAPI服务更干净，运维上也更容易灰度。
- 连接池参数一定要压测调，
limit、limit_per_host、超时三件套，默认值在生产环境基本不能用。 asyncio.gather的异常语义要搞清楚，否则线上会出现“请求已返回但后台任务还在跑”的诡异现象。- uvloop值得上，一行配置20%收益，何乐不为。
代码已脱敏，可直接跑通（下游地址换成你自己的即可）。如果这篇文章帮你少踩一个坑，那就值了。
