一、问题背景：一个被串行调用拖死的聚合接口
我们有个商品详情聚合接口 /api/v1/product/detail，逻辑很朴素：拿到 product_id 后，依次调用三个下游服务——商品基础信息（HTTP）、库存服务（HTTP）、促销服务（HTTP），然后拼装返回。三个下游的平均响应时间分别是 40ms、35ms、45ms，加起来 120ms 左右，理论上单请求耗时也就 130ms 上下。
但线上监控显示 P99 是 1.2 秒，单机 QPS 卡在 120 左右，高峰期 CPU 利用率只有 15%，线程池却排满了。原因很简单：Flask 默认的多线程模型下，每个请求的大部分时间都在 requests.get() 上阻塞等 IO，线程全被占住，新请求只能排队。4核机器开了 32 个 worker 线程，理论上限就是 32 / 0.13s ≈ 246 QPS，实际因为下游抖动和 GIL 争抢，只跑到 120。
这不是加机器能优雅解决的问题——加一倍机器 QPS 翻倍，但成本也翻倍，而且线程上下文切换的开销会越来越难看。真正的解法是让等待 IO 的时间不再占用线程，也就是异步。
二、环境与版本
先把环境交代清楚，异步这块版本差异影响很大：
- Python 3.10.12（3.10 的 asyncio 已经比较稳定，
asyncio.run和TaskGroup之外的东西基本够用） - Flask 2.2.5（老服务，暂不整体迁移到 FastAPI，采用渐进式改造）
- aiohttp 3.9.5
- uvloop 0.19.0（Linux 下的高性能事件循环）
- gunicorn 21.2.0 + gevent 23.9.1（过渡期用）
- 压测工具：wrk 4.2.0，
wrk -t8 -c200 -d60s
机器规格：4 核 8G，CentOS 7.9，下游服务在同一内网，RTT 约 0.5ms。
三、方案设计：不整体重写，先隔离异步边界
一开始考虑过直接把整个服务迁到 FastAPI，但服务里有二十多个接口，迁移成本和回归测试量太大。最后定的方案是渐进式：只把最热的这个聚合接口改成异步，其余接口保持不变，通过一个"异步执行器"在 Flask 里跑事件循环。
核心思路：
- 在应用启动时创建一个全局事件循环，跑在独立线程里（避免和 Flask 的 WSGI 线程冲突）；
- 聚合接口把协程提交到这个循环执行，用
asyncio.run_coroutine_threadsafe拿结果； - 三个下游调用改成
aiohttp并发发起，用asyncio.gather聚合； - 全局复用一个
aiohttp.ClientSession，配好连接池。
这样改动的代码面很小，其他接口零影响，出问题可以一键切回同步实现（加个开关）。
另一个关键决策是并发控制。下游服务扛不住无限并发，所以对每个下游用 asyncio.Semaphore 限流，商品服务 200 并发、库存 150、促销 100，这个数字是压测调出来的，后面细说。
四、核心实现
4.1 异步执行器（Flask 里跑事件循环）
# async_runtime.py
import asyncio
import threading
from concurrent.futures import Future
class AsyncRuntime:
def __init__(self):
self._loop = None
self._thread = None
self._ready = threading.Event()
def start(self):
def _run():
# Linux 下用 uvloop 替换默认事件循环，实测吞吐提升约 15%
import uvloop
asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
self._loop = asyncio.new_event_loop()
asyncio.set_event_loop(self._loop)
self._ready.set()
self._loop.run_forever()
self._thread = threading.Thread(target=_run, name="async-loop", daemon=True)
self._thread.start()
self._ready.wait(timeout=5)
if self._loop is None:
raise RuntimeError("async event loop failed to start")
def submit(self, coro, timeout=3.0):
if self._loop is None:
raise RuntimeError("runtime not started")
fut: Future = asyncio.run_coroutine_threadsafe(coro, self._loop)
return fut.result(timeout=timeout)
def stop(self):
if self._loop:
self._loop.call_soon_threadsafe(self._loop.stop)
runtime = AsyncRuntime()
这里有个细节：run_coroutine_threadsafe 返回的是 concurrent.futures.Future，.result(timeout) 是阻塞调用的。所以在 Flask 的 worker 线程里，这个线程依然会被占住——但没关系，因为它占的是一个轻量线程，而真正的 IO 等待发生在事件循环线程里，一个循环线程可以同时处理成百上千个协程。Flask 那侧只要把线程数调大一点（比如 64），瓶颈就不在它身上了。
4.2 聚合接口的异步实现
# product_service.py
import asyncio
import aiohttp
from aiohttp import ClientTimeout
# 全局连接池配置：这是调优的重头戏，见 4.3
CONNECTOR = aiohttp.TCPConnector(
limit=300, # 总连接数上限
limit_per_host=100, # 单 host 上限
ttl_dns_cache=300, # DNS 缓存 5 分钟
keepalive_timeout=30,
)
TIMEOUT = ClientTimeout(total=2.0, connect=0.5, sock_read=1.5)
_session: aiohttp.ClientSession | None = None
# 每个下游的并发闸门，数字来自压测
SEM_PRODUCT = asyncio.Semaphore(200)
SEM_STOCK = asyncio.Semaphore(150)
SEM_PROMO = asyncio.Semaphore(100)
def get_session() -> aiohttp.ClientSession:
global _session
if _session is None or _session.closed:
_session = aiohttp.ClientSession(connector=CONNECTOR, timeout=TIMEOUT)
return _session
async def _fetch(session, sem, url, params):
async with sem: # 超过闸门就排队，保护下游
async with session.get(url, params=params) as resp:
resp.raise_for_status()
return await resp.json()
async def fetch_product_detail(product_id: int) -> dict:
session = get_session()
# 三个下游并发发起，总耗时 ≈ max(40, 35, 45) 而非 120
product_task = _fetch(session, SEM_PRODUCT,
"http://product-svc/internal/detail", {"id": product_id})
stock_task = _fetch(session, SEM_STOCK,
"http://stock-svc/internal/stock", {"id": product_id})
promo_task = _fetch(session, SEM_PROMO,
"http://promo-svc/internal/promo", {"id": product_id})
# return_exceptions=True 让单个下游失败不拖垮整个接口
product, stock, promo = await asyncio.gather(
product_task, stock_task, promo_task, return_exceptions=True
)
if isinstance(product, Exception):
raise RuntimeError(f"product service failed: {product!r}")
return {
"product": product,
"stock": None if isinstance(stock, Exception) else stock,
"promo": None if isinstance(promo, Exception) else promo,
}
Flask 侧的视图函数几乎没动：
@app.route("/api/v1/product/detail")
def product_detail():
product_id = int(request.args["id"])
data = runtime.submit(fetch_product_detail(product_id), timeout=2.5)
return jsonify({"code": 0, "data": data})
4.3 连接池参数：别小看这几个数字
limit=300 / limit_per_host=100 不是拍脑袋来的。压测时先用 limit=100，QPS 到 900 就上不去了，aiohttp 开始报 Connection pool is full, discarding connection 的 warning。原因是单请求要占 3 个连接（三个下游各一个），并发 200 时理论上需要 600 个连接，但每个 host 只有 100 个上限，实际把每个下游卡在 100 并发。改成 limit_per_host=100 配合下游闸门（200/150/100）后，连接数和闸门上限对齐，warning 消失，QPS 直接上到 1500+。
keepalive_timeout=30 也是调过的：默认是 15 秒，压测时下游连接复用率不够，每次重建连接多花 1~2ms，累计起来 P99 会高 20ms 左右。改成 30 秒后连接复用率从 78% 提到 96%。
五、踩坑与优化
坑一：全局 ClientSession 在 fork 后失效。 gunicorn 用 gevent worker 时，如果 ClientSession 在 master 进程里创建，fork 到 worker 后连接池句柄是坏的，表现为随机 RuntimeError: Event loop is closed。解法是延迟到第一次请求时再创建（上面的 get_session 就是懒加载），并注册 atexit 关闭。
坑二：asyncio.gather 不带 return_exceptions 会连坐。 最初没加这个参数，促销服务偶发 500 时，整个商品详情接口直接 500。改成 return_exceptions=True 后，促销挂了就返回 promo: null，前端降级展示，可用性从 99.2% 提到 99.95%。
坑三：超时设置要分层。 一开始只设了 ClientTimeout(total=2.0)，结果下游"慢但不断"的时候，请求会一直挂着直到总超时，把信号量占满。后来拆成 connect=0.5, sock_read=1.5，连接慢和读慢分开处理，信号量占用时间明显下降。另外 Flask 侧的 submit(timeout=2.5) 要略大于 aiohttp 的总超时，否则事件循环里还在跑、外面已经抛超时了，日志会对不上。
坑四：日志和 trace_id 的传递。 异步后一个请求跨了多个协程，threading.local 失效。改用 contextvars.ContextVar 存 trace_id，在 fetch_product_detail 入口 set，_fetch 里 get，日志才能串起来。这块不复杂但容易漏，漏了之后排查线上问题会很痛苦。
优化：uvloop 和 DNS 缓存。 换 uvloop 后同样的压测脚本 QPS 从 1560 提到 1800 左右。ttl_dns_cache=300 把 DNS 解析从每次请求一次降到 5 分钟一次，内网 DNS 虽然快，但省下来的 0.3ms 在 P99 上还是能看出来。
六、效果数据
同一台 4 核 8G 机器，wrk 压测 60 秒，并发 200：
| 指标 | 改造前（Flask 同步） | 改造后（asyncio+aiohttp） |
|---|---|---|
| QPS | 120 | 1800 |
| P50 延迟 | 210ms | 68ms |
| P99 延迟 | 1200ms | 180ms |
| CPU 利用率 | 15% | 62% |
| 内存占用 | 380MB | 420MB |
| 下游错误率 | 0.8% | 0.05% |
QPS 提升 15 倍，P99 从 1.2 秒降到 180ms。CPU 利用率上来了是好事——说明机器终于被用起来了，而不是在等 IO。内存涨了 40MB，主要是连接池和事件循环的开销，可接受。
下游错误率反而降了，因为信号量限流把突发流量挡在了下游前面，下游不再被瞬时打爆。
七、总结
这次改造最大的体会是：异步不是银弹，边界要划清楚。我们没有整体迁移框架，只在最热的接口上切开一个异步边界，用 run_coroutine_threadsafe 桥接同步和异步世界，改动量控制在 300 行以内，风险可控。
几个可以直接抄走的经验：
Semaphore的数值一定要压测调，宁小勿大，保护下游比榨干自己更重要；ClientSession全局复用 + 懒加载，连接池limit_per_host要和并发闸门对齐；gather(return_exceptions=True)是聚合接口的标配，单点失败要能降级；- 超时分层（connect / sock_read / total），别只设一个 total；
contextvars替代threading.local，不然日志没法看。
下一步打算把库存扣减这类写接口也迁过来，但那涉及事务和幂等，坑会更多，等踩完再写一篇。
