一、问题背景：一个"看起来很闲"的接口
事情是这样的。我们有个内部服务，提供一个 /api/aggregate 接口，作用很简单：接收一个用户ID，去下游三个服务（用户画像、订单历史、推荐列表）拉数据，拼装后返回。下游都是HTTP接口，平均响应时间 80~150ms。
代码是典型的Flask同步写法，用requests逐个请求：
# before: app.py (Flask 2.2.5 + requests 2.28.2)
import requests
from flask import Flask, jsonify, request
app = Flask(__name__)
PROFILE_URL = "http://profile-svc:8001/user/{}"
ORDER_URL = "http://order-svc:8002/orders?uid={}"
RECO_URL = "http://reco-svc:8003/list/{}"
@app.route("/api/aggregate")
def aggregate():
uid = request.args.get("uid")
profile = requests.get(PROFILE_URL.format(uid), timeout=2).json()
orders = requests.get(ORDER_URL.format(uid), timeout=2).json()
reco = requests.get(RECO_URL.format(uid), timeout=2).json()
return jsonify({"profile": profile, "orders": orders, "reco": reco})
三个请求串行，每个100ms左右，接口总耗时300ms+。用wrk压测（4线程、100连接、跑30秒）：
Running 30s test @ http://127.0.0.1:5000/api/aggregate?uid=123
4 threads and 100 connections
Requests/sec: 121.83
Latency P50: 780ms P99: 1.82s
QPS 120，P99 1.8秒。但看监控，CPU使用率只有8%，进程数1，内存200MB。CPU闲着、请求排队——这就是同步IO的经典症状：每个请求都在等网络，Gunicorn 默认同步worker只能同时处理固定数量请求。当时配的是 gunicorn -w 4，4个worker，每个worker一次处理一个请求，等于并发度只有4。
最直觉的方案是加worker数量，但这是治标：worker是进程，每个占内存，加到32个worker内存直接起飞，而且下游连接数也会爆炸。真正的问题是：这三个下游请求彼此独立，完全可以并发发出。
二、环境与版本
- Python 3.11.6（选3.11是因为asyncio在3.11有显著的任务调度优化）
- aiohttp 3.9.1
- Flask 2.2.5 → 保留Flask做路由层，用
asgiref桥接（后面解释为什么不全换FastAPI） - gunicorn 21.2.0，worker class 换成
gunicorn.workers.gthread.GThreadWorker或直接用 uvicorn - 压测工具：wrk 4.2.0
先说明一个决策：为什么不直接把Flask换成FastAPI？因为这个服务还有十几个同步接口，全量迁移风险大。我的做法是只把这个热点接口改成异步，用 asgiref.sync.async_to_sync 在Flask里跑协程——后来发现这条路在gunicorn下有问题，最终这个接口单独拆了一个uvicorn进程，监听不同端口，网关层做路由。下面会讲这个坑。
三、方案设计
核心思路三条：
- 并发发起下游请求：三个请求用
asyncio.gather并发，总耗时从 3×100ms 降到 max(100ms)。 - 连接复用：aiohttp 的
ClientSession持有连接池，避免每次请求重新TCP握手。默认limit=100，我按压测结果调到limit=200，limit_per_host=50。 - 并发上限保护：不能让无限请求同时打下游。用
asyncio.Semaphore限制同时进行的下游调用总数，超出的排队等待，配合asyncio.wait_for做超时兜底。
架构上，接口拆成独立进程：
网关(Nginx) → /api/aggregate → uvicorn:8100 (asyncio版)
→ 其它路由 → gunicorn:5000 (原Flask)
四、核心实现
4.1 after 代码
# after: async_app.py (aiohttp 3.9.1, Python 3.11.6)
import asyncio
import aiohttp
from aiohttp import web, ClientTimeout
PROFILE_URL = "http://profile-svc:8001/user/{}"
ORDER_URL = "http://order-svc:8002/orders?uid={}"
RECO_URL = "http://reco-svc:8003/list/{}"
# 全局信号量：限制同时在途的下游请求总数
DOWNSTREAM_SEM = asyncio.Semaphore(64)
async def fetch(session: aiohttp.ClientSession, url: str, timeout: float = 2.0):
async with DOWNSTREAM_SEM:
try:
async with session.get(url, timeout=ClientTimeout(total=timeout)) as resp:
resp.raise_for_status()
return await resp.json()
except (aiohttp.ClientError, asyncio.TimeoutError) as e:
# 单个下游失败不拖垮整个接口，返回降级值
return {"error": type(e).__name__, "url": url}
async def aggregate(request: web.Request) -> web.Response:
uid = request.query.get("uid")
if not uid:
raise web.HTTPBadRequest(text="uid required")
session: aiohttp.ClientSession = request.app["session"]
profile, orders, reco = await asyncio.gather(
fetch(session, PROFILE_URL.format(uid)),
fetch(session, ORDER_URL.format(uid)),
fetch(session, RECO_URL.format(uid)),
)
return web.json_response({"profile": profile, "orders": orders, "reco": reco})
async def make_app() -> web.Application:
app = web.Application()
# 连接池参数：limit 总连接数，limit_per_host 单host上限
connector = aiohttp.TCPConnector(
limit=200,
limit_per_host=50,
ttl_dns_cache=300,
enable_cleanup_closed=True,
)
app["session"] = aiohttp.ClientSession(
connector=connector,
timeout=ClientTimeout(total=3.0, connect=0.5),
)
app.router.add_get("/api/aggregate", aggregate)
return app
async def on_cleanup(app):
await app["session"].close()
if __name__ == "__main__":
app = asyncio.run(make_app())
app.on_cleanup.append(on_cleanup)
web.run_app(app, host="0.0.0.0", port=8100, access_log=None)
启动：python async_app.py，或者生产用 uvicorn async_app:make_app --factory --workers 1 --loop uvloop --http httptools --port 8100。注意 workers 必须为 1——asyncio 单进程单事件循环，多worker会各自建连接池，反而稀释了连接复用收益，而且 Semaphore 是进程内的，多进程下限流失效。要扩并发靠单进程内的协程数，要扩CPU靠多机，不要用多worker。
4.2 连接池与超时参数怎么定的
几个关键参数和我的取值依据：
limit=200：压测时从100加到200，QPS从1600涨到2050；到300时下游开始报连接拒绝，所以停在200。limit_per_host=50：三个下游各50，留余量给健康检查。connect=0.5：建连超时0.5s，下游都在内网，正常建连<10ms，0.5s足够，避免慢建连拖住事件循环。total=2.0（单请求）/total=3.0（session级）：下游P99约300ms，2s是给异常留的兜底。Semaphore(64)：这是下游总并发的闸门，不是接口并发。64 = 200连接池里同时活跃的上限，实测超过64后下游P99开始劣化，就定在这。
五、踩坑与优化
坑1：Flask里用async_to_sync，gunicorn sync worker下直接死锁。 我一开始想省事，在Flask路由里 async_to_sync(aggregate)()，本地跑通了，上gunicorn（sync worker）后压测直接hang住。原因是sync worker里没有运行中的事件循环，asgiref每次新建loop，而aiohttp的session是在另一个loop里创建的，跨loop使用连接池报错/挂起。教训：aiohttp的ClientSession必须和事件循环同生命周期，别想着在同步框架里偷懒复用。
坑2：ClientSession在模块顶层创建。 很多人写成 session = aiohttp.ClientSession() 放模块级，Python 3.10+ 下import时没有running loop，直接 RuntimeError: no running event loop。必须放进 make_app() 里、在loop内创建，并用 on_cleanup 关闭。我第一次就是这么写的，服务起不来。
坑3：超时用 asyncio.wait_for 包 session.get，连接没释放。 早期版本我用 wait_for(session.get(url), 2)，超时后协程被cancel，但底层连接有时没归还池子，压测久了报 Connection pool is full。正确做法是给 session.get 传 ClientTimeout，让aiohttp自己管理超时和连接释放，而不是在外面套 wait_for。
优化1：uvloop。 换 uvloop 后同压力下QPS +18%，P99降了约15ms。启动加 --loop uvloop。
优化2：gather 加 return_exceptions=True。 我最初的 fetch 内部try/except已经兜住了，但如果哪天有人改代码漏了，一个下游异常会让整个 gather 抛错。保险起见改成 asyncio.gather(..., return_exceptions=True)，再统一处理。
优化3：DNS缓存。 ttl_dns_cache=300，内网服务名解析虽然快，但高QPS下DNS查询也是开销，缓存5分钟。
六、效果数据
同一台机器（4C8G），wrk 4线程100连接30秒，下游用mock服务固定延迟100ms：
| 指标 | before (Flask+requests, 4 worker) | after (aiohttp+uvloop, 1 worker) |
|---|---|---|
| QPS | 121.83 | 2104.6 |
| P50 延迟 | 780ms | 142ms |
| P99 延迟 | 1.82s | 213ms |
| CPU 使用率 | 8% | 62% |
| 内存 RSS | 201MB | 138MB |
| 下游连接数峰值 | ~4 | ~180（复用，实际TCP连接<60） |
QPS 提升约17倍，P99 从1.8s降到213ms。注意CPU从8%涨到62%——这不是变差了，是之前CPU在空转等IO，现在真的在干活。内存降了是因为单进程替代了4个Flask worker，且连接复用减少了socket缓冲。
真实下游（非mock）环境复测：QPS 1850左右，P99 240ms，比mock略低，因为下游本身P99有波动，但量级一致。
七、总结
这次改造的核心其实就一句话：IO密集的接口，串行改并发，同步改异步，收益是数量级的。但有几个点值得记住：
- 别在同步框架里硬塞asyncio，ClientSession和事件循环必须同生命周期，跨loop必炸。
- 连接池参数（
limit、limit_per_host）和Semaphore要一起调，前者管TCP复用，后者管下游压力，缺一个都会在高压下出问题。 - 超时交给aiohttp的
ClientTimeout，别在外面套wait_for，否则连接泄漏。 - asyncio服务用单worker+uvloop，扩并发靠协程，扩CPU靠多机，别用
--workers N。
如果你的服务里也有这种"多个独立下游串行调用"的接口，可以按这个路子先改一个试试，压测数据会说服你的。代码里那个 fetch 的降级返回也建议保留——异步并发下，一个下游抖动很容易放大成雪崩，单点失败降级比整体超时划算得多。
