一、问题背景：一个被下游拖死的接口
去年接手的风控系统里有个 /risk/query 接口，逻辑很直白：收到用户ID后，依次调用三个下游服务拿数据，合并后返回。
- 用户画像服务：平均 90ms
- 黑名单服务：平均 70ms
- 行为评分服务：平均 110ms
三个是串行写的，加起来理论 270ms。但线上 P99 能到 1.8s，因为下游偶尔抖动，一个慢请求会把整个线程卡住。Flask 默认多线程模型，gunicorn 配了 --workers 4 --threads 8，一共 32 个并发槽位。压测数据惨不忍睹：
Concurrency 50, Duration 60s
QPS: 118
P50: 412ms
P99: 1820ms
CPU: 78% (4 cores)
问题本质有两个：一是串行调用浪费了等待时间，二是同步阻塞模型下线程数就是并发上限。下游 90% 的时间在等网络 IO，CPU 根本没干活。
二、环境与版本
不写版本号的性能文章都是耍流氓：
- Python 3.11.6（3.11 的 asyncio 有专门优化，比 3.8 快不少）
- Flask 3.0.0 + gunicorn 21.2.0（保留，作为入口）
- aiohttp 3.9.1
- uvloop 0.19.0（关键，Linux 下事件循环加速）
- 压测：wrk 4.2.0
- 机器：4C8G，Ubuntu 22.04，内网同机房
三、方案设计
核心思路：把 IO 等待并发化，把线程模型换成事件循环。
架构上做了两层改造：
- 接入层：gunicorn 换成
uvicorn风格的 ASGI 部署不现实（改动太大），所以保留 Flask 作为入口，但内部通过asyncio.run_coroutine_threadsafe把请求转发给一个常驻事件循环。更彻底的做法是直接用 FastAPI，后面会讲为什么没选。 - 下游调用层：三个下游用
asyncio.gather并发，共享一个aiohttp.ClientSession连接池。
超时策略：总超时 500ms，单下游 300ms。任何一路超时走降级，返回默认值而不是报错——风控接口可用性优先。
连接池配置：
connector = aiohttp.TCPConnector(
limit=200, # 总连接数
limit_per_host=50, # 单host连接数
ttl_dns_cache=300, # DNS缓存5分钟
enable_cleanup_closed=True,
)
timeout = aiohttp.ClientTimeout(total=0.5, connect=0.1, sock_read=0.3)
四、核心实现
Before：同步串行版本
# before.py
import requests
from flask import Flask, jsonify, request
app = Flask(__name__)
SESS = requests.Session()
SESS.mount("http://", requests.adapters.HTTPAdapter(
pool_connections=32, pool_maxsize=32, max_retries=1))
def fetch(url, uid):
try:
r = SESS.get(f"{url}?uid={uid}", timeout=0.5)
return r.json()
except Exception:
return {}
@app.route("/risk/query")
def query():
uid = request.args.get("uid")
profile = fetch("http://profile.svc/get", uid) # ~90ms
black = fetch("http://black.svc/check", uid) # ~70ms
score = fetch("http://score.svc/get", uid) # ~110ms
return jsonify({
"uid": uid,
"level": decide(profile, black, score),
})
三个 fetch 顺序执行，总耗时是三者之和。
After：asyncio 并发版本
# after.py
import asyncio
import aiohttp
from flask import Flask, jsonify, request
app = Flask(__name__)
# 常驻事件循环，独立线程跑
LOOP = asyncio.new_event_loop()
def _start_loop(loop):
asyncio.set_event_loop(loop)
loop.run_forever()
import threading
threading.Thread(target=_start_loop, args=(LOOP,), daemon=True).start()
CONNECTOR = aiohttp.TCPConnector(
limit=200, limit_per_host=50,
ttl_dns_cache=300, enable_cleanup_closed=True,
)
TIMEOUT = aiohttp.ClientTimeout(total=0.5, connect=0.1, sock_read=0.3)
async def fetch(session, url, uid):
try:
async with session.get(f"{url}?uid={uid}") as resp:
return await resp.json()
except Exception:
return {}
async def query_async(uid):
async with aiohttp.ClientSession(
connector=CONNECTOR, timeout=TIMEOUT
) as session:
profile, black, score = await asyncio.gather(
fetch(session, "http://profile.svc/get", uid),
fetch(session, "http://black.svc/check", uid),
fetch(session, "http://score.svc/get", uid),
)
return decide(profile, black, score)
@app.route("/risk/query")
def query():
uid = request.args.get("uid")
fut = asyncio.run_coroutine_threadsafe(query_async(uid), LOOP)
level = fut.result(timeout=0.6)
return jsonify({"uid": uid, "level": level})
关键点：ClientSession 每次请求都新建其实有点浪费，但复用 session 要处理跨线程问题。实测下来新建 session 开销约 0.2ms，相对 200ms 的 IO 可以忽略，代码更干净。如果追求极致，可以把 session 也常驻在事件循环里。
五、踩坑与优化
坑1：uvloop 没生效。 一开始忘了在事件循环启动前 import uvloop; uvloop.install()，QPS 卡在 800。加上之后直接到 1000+。uvloop 对高频短连接的收益非常明显。
坑2：asyncio.gather 默认不取消。 如果一路超时抛异常，其他协程不会被取消，会继续跑完。改成 gather(..., return_exceptions=True) 配合内部 try/except，确保超时的那路不影响整体。
坑3：DNS 解析成瓶颈。 下游用的是 k8s service 域名，每次请求都解析。开 ttl_dns_cache=300 后 P99 降了 60ms。这个参数很多人会忽略。
坑4：连接池 limit 太小。 最初设 limit=50，压测到 800 QPS 就上不去了，connector._acquired 一直打满。调到 200 后解决。经验值：limit ≈ 目标QPS × 平均RT，1100 × 0.2s = 220，取 200 略保守。
坑5：Flask 的 fut.result() 阻塞了工作线程。 这其实是妥协方案。真正优雅的是全量换 FastAPI/Starlette，但改造成本高。当前方案下 gunicorn 至少要配 --threads 16，不然线程还是会被 result() 阻塞。我们最终配的是 --workers 2 --threads 16。
六、效果数据
同一台机器，同样 50 并发压 60 秒：
| 指标 | Before | After | 提升 |
|---|---|---|---|
| QPS | 118 | 1103 | 9.3x |
| P50 | 412ms | 88ms | 4.7x |
| P99 | 1820ms | 210ms | 8.7x |
| CPU | 78% | 51% | -35% |
| 内存 | 320MB | 410MB | +28% |
内存涨了 90MB，主要是连接池和事件循环的开销，完全可接受。CPU 反而降了，因为线程上下文切换少了。
下游抖动场景（人为给 score 服务加 300ms 延迟）：
- Before P99：2400ms
- After P99：340ms（超时降级生效）
七、总结
这次改造的核心不是“用了 asyncio”，而是认清了瓶颈在 IO 等待而非计算。如果下游是 CPU 密集，asyncio 一点用没有，反而增加复杂度。
几点经验：
- 并发调用下游，
gather是标配，但别忘了return_exceptions和超时。 - 连接池参数必须压测调优，
limit_per_host是重点。 uvloop在 Linux 上必装，收益立竿见影。- Flask 里嵌 asyncio 是权宜之计，长期看该上 FastAPI。但如果你有个跑了两年的 Flask 项目，这个方案能在一天内上线，风险可控。
- 降级比报错重要。风控接口挂了比返回保守结果严重得多。
最后贴一下压测命令，方便复现：
wrk -t4 -c50 -d60s --latency \
"http://localhost:8000/risk/query?uid=test123"
下一步打算把 session 常驻事件循环，再压一轮看能不能破 1300。有结果回来更新。
