一、问题背景：一个被同步IO拖死的API
去年接手了一个内部服务，功能很简单：接收客户端请求，查一次PostgreSQL，再调一次下游的用户画像服务，拼装结果返回。业务逻辑不到50行，但每天要扛200万次调用。
原来的实现是Flask + requests + psycopg2，典型的同步栈。上线初期没问题，随着调用方增多，问题开始暴露：
- 压测QPS卡在120上不去，4核机器CPU利用率只有15%
- P99延迟1.2s，高峰期偶发超时
- 运维加了两次机器，QPS线性增长但成本吃不消
打开代码一看就明白了：每个请求要串行做两次网络IO——一次数据库查询（平均15ms），一次下游HTTP调用（平均40ms）。同步模型下，线程在这55ms里全程阻塞，什么也干不了。4核机器开200个worker线程，理论并发上限就是200，QPS = 200 / 0.055 ≈ 3600，但实际因为GIL和线程切换开销，跑到120就到顶了。
问题本质很清楚：这是个IO密集型服务，却被同步阻塞模型限制了并发能力。Python的GIL让多线程在CPU密集场景没用，但在IO密集场景，asyncio的事件循环才是正解——单线程就能撑起几千个并发协程。
二、环境与版本
先交代下环境，避免版本差异导致的行为不一致：
- Python 3.11.6（3.11的asyncio有C实现的Task优化，比3.8快不少）
- aiohttp 3.9.1（Web框架 + 客户端）
- asyncpg 0.29.0（PostgreSQL异步驱动，比aiopg快，因为它是直接基于C实现的协议）
- PostgreSQL 14.10
- wrk 4.2.0 压测
- 机器：4核8G，Ubuntu 22.04
选aiohttp而不是FastAPI，是因为这个服务不需要OpenAPI文档，也不想引入pydantic的校验开销（虽然影响很小），aiohttp更轻，启动快，依赖少。如果团队已经在用FastAPI，方案完全通用，把路由层换掉就行。
三、方案设计
改造的核心思路只有一条：把两次串行IO改成协程并发。
原来的同步逻辑是：
查DB (15ms) → 调下游 (40ms) → 组装返回 总计 55ms
这两次IO之间没有数据依赖（下游调用只需要用户ID，不需要DB结果），完全可以并发：
┌→ 查DB (15ms) ─┐
请求 →──┤ ├→ 组装返回 总计 max(15, 40) = 40ms
└→ 调下游 (40ms)┘
单请求延迟从55ms降到40ms，同时单线程可以挂起成千上万个协程，并发能力不再受线程数限制。
架构上分三层：
1. Web层：aiohttp的Application，路由处理函数声明为async def
2. 业务层：用asyncio.gather并发编排DB查询和下游调用
3. 资源层：asyncpg连接池 + aiohttp ClientSession，全局复用，避免每次请求建连接
连接池参数是关键，后面单独讲。
四、核心实现
Before：同步版本
# app_sync.py
import requests
import psycopg2
from psycopg2 import pool
from flask import Flask, jsonify, request
app = Flask(__name__)
# 同步连接池
db_pool = pool.ThreadedConnectionPool(
minconn=5, maxconn=50,
host="127.0.0.1", port=5432,
database="userdb", user="app", password="***"
)
DOWNSTREAM_URL = "http://profile-svc:8080/profile"
@app.route("/api/user/")
def get_user(uid):
conn = db_pool.getconn()
try:
with conn.cursor() as cur:
cur.execute("SELECT id, name, level FROM users WHERE id = %s", (uid,))
row = cur.fetchone()
if not row:
return jsonify({"error": "not found"}), 404
user = {"id": row[0], "name": row[1], "level": row[2]}
finally:
db_pool.putconn(conn)
# 串行调用下游
resp = requests.get(DOWNSTREAM_URL, params={"uid": uid}, timeout=2)
profile = resp.json() if resp.ok else {}
return jsonify({**user, "profile": profile})
if __name__ == "__main__":
app.run(host="0.0.0.0", port=8000, threaded=True)
这段代码本身没毛病，逻辑清晰，但每次请求两次网络IO全程阻塞，worker线程就这么被占着。
After：异步版本
# app_async.py
import asyncio
import asyncpg
from aiohttp import web, ClientSession, ClientTimeout
DB_DSN = "postgresql://app:***@127.0.0.1:5432/userdb"
DOWNSTREAM_URL = "http://profile-svc:8080/profile"
# 全局资源，在on_startup中初始化
db_pool: asyncpg.Pool = None
http_session: ClientSession = None
async def get_user(request: web.Request) -> web.Response:
uid = int(request.match_info["uid"])
async def fetch_user():
async with db_pool.acquire() as conn:
row = await conn.fetchrow(
"SELECT id, name, level FROM users WHERE id = $1", uid
)
if row is None:
return None
return {"id": row["id"], "name": row["name"], "level": row["level"]}
async def fetch_profile():
try:
async with http_session.get(
DOWNSTREAM_URL, params={"uid": uid},
timeout=ClientTimeout(total=2)
) as resp:
if resp.status == 200:
return await resp.json()
return {}
except asyncio.TimeoutError:
return {}
# 两次IO并发执行
user, profile = await asyncio.gather(fetch_user(), fetch_profile())
if user is None:
return web.json_response({"error": "not found"}, status=404)
return web.json_response({**user, "profile": profile})
async def on_startup(app: web.Application):
global db_pool, http_session
db_pool = await asyncpg.create_pool(
dsn=DB_DSN,
min_size=10,
max_size=50, # 根据PG的max_connections和实例数算
max_inactive_connection_lifetime=300,
command_timeout=2,
)
http_session = ClientSession(
connector=aiohttp.TCPConnector(
limit=200, # 总连接上限
limit_per_host=100,
ttl_dns_cache=300,
keepalive_timeout=30,
),
timeout=ClientTimeout(total=2),
)
async def on_cleanup(app: web.Application):
await db_pool.close()
await http_session.close()
def create_app() -> web.Application:
app = web.Application()
app.router.add_get("/api/user/{uid:\\d+}", get_user)
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=8000, access_log=None)
几个关键点：
asyncio.gather让两次IO真正并发，总耗时取决于慢的那个，而不是两者之和。- 连接池全局复用，不在请求里建连接。asyncpg的
create_pool是懒加载的，min_size个连接启动时建好，max_size是并发上限。 - ClientSession必须全局唯一。这是aiohttp最容易踩的坑，每个请求建一个session会导致连接无法复用，还会泄漏。aiohttp的文档明确要求session在应用生命周期内共享。
on_startup/on_cleanup钩子管理资源生命周期，避免在模块顶层用await（模块加载时没有事件循环）。
五、踩坑与优化
改造过程不是一次成功的，记录几个真实的坑：
坑1：ClientSession放在请求里创建。 第一版图省事，在fetch_profile里async with ClientSession() as s，压测时QPS只有400，而且CLOSE_WAIT状态的连接暴涨。原因是每次请求都新建连接池，keepalive完全失效。改成全局session后QPS直接翻倍。这是aiohttp新手最容易犯的错。
坑2：连接池参数拍脑袋。 一开始max_size=100，觉得越大越好。结果PG的max_connections默认100，服务启动几个worker就把PG连接打满了，报too many clients。后来按公式算：max_size × worker数 + 预留 < max_connections。4个worker进程，每个max_size=50，加上预留，把PG调到300才够。另外command_timeout=2一定要设，避免慢查询把连接池占死。
坑3：没给下游调用设超时。 aiohttp的默认timeout是5分钟，下游服务一旦抖动，协程全部挂起，虽然不阻塞事件循环，但连接池会被占满，后续请求排队。显式设ClientTimeout(total=2)后，下游超时的请求快速失败，整体可用性反而提升了。
坑4：asyncio.gather默认不取消。 如果其中一个任务抛异常，另一个还会继续跑。我们的场景里下游失败可以降级返回空，所以用return_exceptions=False（默认）配合内部try/except就够了。但如果两个任务都是强依赖，记得处理异常传播，避免"孤儿协程"。
优化1：uvloop替换默认事件循环。 在__main__里加两行：
import uvloop
uvloop.install()
uvloop基于libuv，事件循环的吞吐比标准库的selector实现高不少，实测QPS再涨约15%。注意uvloop在Windows上不支持，生产环境是Linux没影响。
优化2：关闭access_log。 aiohttp默认每个请求打一行日志，高QPS下日志IO本身就成了瓶颈。access_log=None关掉，日志交给上游的nginx记录。
优化3：json_response的序列化。 aiohttp默认用标准库json，数据量大时可以考虑orjson，但我们的响应体很小（<1KB），收益不明显，没折腾。
六、效果数据
压测命令：wrk -t4 -c500 -d60s http://127.0.0.1:8000/api/user/12345，500并发连接，持续60秒，取稳定段数据。
| 指标 | Before (Flask同步) | After (aiohttp异步) | 提升 |
|---|---|---|---|
| QPS | 120 | 1800 | 15x |
| P50延迟 | 380ms | 22ms | 17x |
| P99延迟 | 1200ms | 85ms | 14x |
| CPU利用率 | 15% | 68% | - |
| 内存占用 | 1.2GB (200线程) | 180MB | 85%↓ |
| 机器数 | 4台 | 2台 | 50%↓ |
几点说明：
- QPS到1800时CPU到68%，瓶颈转移到了PG（单条查询15ms，50个连接的池子理论上限约3300 QPS），再往上要优化SQL或加缓存，这不是asyncio能解决的。
- 延迟的降低一部分来自并发编排（55ms→40ms），更大一部分来自没有线程切换和排队等待。
- 内存从1.2G降到180M，主要是线程栈没了。200个线程每个默认8M栈（虽然实际提交没那么多），加上Flask的请求上下文，开销很可观。
单请求的耗时对比（用日志打点统计的平均值）：同步版55ms（DB 15ms + 下游40ms串行），异步版41ms（并发后取max），和设计预期一致。
七、总结
这次改造的结论很朴素：IO密集型服务，同步阻塞模型就是原罪，asyncio是Python里成本最低的解药。但asyncio不是银弹，几点经验值得记下来：
- 协程里绝对不能有阻塞调用。一次
time.sleep或者同步的requests.get，整个事件循环就卡住了，所有并发协程一起陪葬。改造时要逐个审查依赖库，确认是异步版本（asyncpg而非psycopg2，aiohttp而非requests，aiofiles而非open）。 - 连接池和Session必须全局复用，参数要按"下游容量 × 实例数"反推，不能拍脑袋。
- 超时是生命线。异步模型下，没有超时的协程会静默占着资源，比同步模型更难排查。
- 性能瓶颈会转移。asyncio解决了Web层的并发，瓶颈就跑到数据库或下游去了，要有预期。
最后，如果团队从零起步且需要OpenAPI、依赖注入这些，FastAPI + asyncpg + httpx的组合更现代，方案完全一样。但如果是存量Flask服务要低成本改造，aiohttp这套是最小改动路径。代码已经上了生产，跑了半年，稳定性没问题。
本文代码基于Python 3.11.6 + aiohttp 3.9.1 + asyncpg 0.29.0，压测数据来自内网环境，不同硬件和网络下绝对值会有差异，但量级关系可参考。
