1. 问题背景：一个“看起来不慢”的查询接口
上个月接手了一个订单服务，核心逻辑是：接收前端请求 -> 调用内部用户服务(HTTP) -> 调用商品服务(HTTP) -> 聚合数据返回。单次调用内部服务耗时约80ms，两个串行就是160ms。当时线上QPS峰值150，单机4核8G，CPU空闲但响应时间飘到800ms+。
用ab压测结果如下：
Concurrency Level: 50
Time taken: 10.01s
Requests per second: 121.37
TP99: 812ms
显然瓶颈不在计算，而在同步等待I/O。每个请求占用一个线程，而线程在等待HTTP响应时完全阻塞。方案很明确：用asyncio把阻塞I/O变成非阻塞协程。
2. 环境与版本：别用Flask跑asyncio
先说结论：Flask不支持asyncio，它的WSGI模型是同步的。所以我用aiohttp写了一个纯异步API服务，前端Nginx做反向代理，保留原Flask服务做内部接口兼容。
| 组件 | 版本 |
|---|---|
| Python | 3.10.11 (CPython) |
| aiohttp | 3.8.5 |
| Flask (旧服务) | 2.2.5 |
| gunicorn (旧部署) | 20.1.0 |
| uvloop | 0.17.0 |
| 压测工具 | wrk / ab |
关键决策：不混用asyncio.run()和Flask，而是整个服务用aiohttp.web重写。因为Flask的线程池和事件循环混在一起会有非常隐蔽的bug（后面踩坑部分详述）。
3. 方案设计：三个核心点
改造不是简单地把requests.get换成aiohttp.get，需要系统性设计：
- 连接池复用：每次请求新建连接是灾难，必须用
aiohttp.TCPConnector维护连接池，设置limit=200（超过会排队），ttl_dns_cache=300减少DNS解析。 - 并发限流：内部服务有负载上限，全量并发会把下游打挂。用
asyncio.Semaphore(100)控制最大并发HTTP调用数。 - 串行改并行：原来用户服务和商品服务是串行调用，改成
asyncio.gather()并行发起，总耗时从160ms降到80ms。
4. 核心实现：Before & After代码
Before (Flask + requests，同步阻塞)
# app_sync.py - 改造前
from flask import Flask, jsonify
import requests
import time
app = Flask(__name__)
def fetch_user(user_id):
# 模拟下游服务，实际为HTTP调用
resp = requests.get(f"http://user-service/users/{user_id}", timeout=1)
return resp.json()
def fetch_order(order_id):
resp = requests.get(f"http://order-service/orders/{order_id}", timeout=1)
return resp.json()
@app.route("/api/order/")
def get_order(order_id):
# 串行调用两个下游服务
user_data = fetch_user(order_id)
order_data = fetch_order(order_id)
# 聚合逻辑
result = {"user": user_data, "order": order_data, "ts": time.time()}
return jsonify(result)
if __name__ == "__main__":
app.run(threaded=True, processes=1)
After (aiohttp + asyncio，协程并发)
# app_async.py - 改造后
from aiohttp import web
import aiohttp
import asyncio
import uvloop
import time
asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
# 全局连接池，limit=200表示最多200个并发连接
_connector = aiohttp.TCPConnector(limit=200, ttl_dns_cache=300, enable_cleanup_closed=True)
_semaphore = asyncio.Semaphore(100) # 限制下游并发
async def fetch(session, url):
# 使用信号量控制并发
async with _semaphore:
async with session.get(url, timeout=aiohttp.ClientTimeout(total=2)) as resp:
return await resp.json()
async def fetch_user(session, user_id):
return await fetch(session, f"http://user-service/users/{user_id}")
async def fetch_order(session, order_id):
return await fetch(session, f"http://order-service/orders/{order_id}")
async def handler(request):
order_id = request.match_info["order_id"]
async with aiohttp.ClientSession(connector=_connector) as session:
# 并行调用两个下游服务，总耗时约80ms而非160ms
user_data, order_data = await asyncio.gather(
fetch_user(session, order_id),
fetch_order(session, order_id)
)
result = {"user": user_data, "order": order_data, "ts": time.time()}
return web.json_response(result)
if __name__ == "__main__":
app = web.Application()
app.router.add_get("/api/order/{order_id}", handler)
web.run_app(app, host="0.0.0.0", port=8080)
关键点解释：
- uvloop 替换默认事件循环，提升约15%的吞吐（官方基准测试数据）。
- TCPConnector(limit=200) 而不是默认的 limit=100，在4核8G机器上测出来200是最优值，过高会触发TIME_WAIT堆积。
- asyncio.gather() 是性能提升的核心：两个80ms的串行调用变成并行后，单请求耗时直接减半。
5. 踩坑与优化：三个真实生产事故
坑1：把连接池放在请求内创建
第一个版本我在handler里写async with aiohttp.ClientSession() as session，导致每个请求都新建连接池。压测时QPS反而降到80，因为连接建立和销毁的开销远大于节省的时间。修复：把ClientSession提升到模块级别，配合connector复用。
坑2：Semaphore 和 Connection Pool 的相互作用
设置_semaphore = Semaphore(200)，但TCPConnector(limit=100)。结果因为信号量放行200个协程，但连接池只有100个连接，导致一半协程排队等待连接池，反而增加了延迟。经验公式：semaphore_limit <= connector_limit，通常取connector_limit * 0.8。
坑3：asyncio.TimeoutError 和 aiohttp.ClientTimeout 的坑
asyncio.wait_for(session.get(url), timeout=2) 这种写法会取消协程，但底层socket未必立即关闭，导致连接泄漏。修复：使用aiohttp.ClientTimeout(total=2)，它会在超时后正确清理连接。
6. 效果数据：wrk压测对比
在同一台4核8G CentOS 7.9机器上，用wrk -t4 -c50 -d30s压测：
| 指标 | Flask+requests | aiohttp+asyncio | 提升倍数 |
|---|---|---|---|
| QPS | 121 | 2147 | 17.7x |
| TP99 | 812ms | 45ms | 18x |
| TP999 | 2.1s | 89ms | 23.6x |
| CPU利用率 | 28% | 85% | - |
| 内存占用 | 420MB | 310MB | - |
解释：QPS提升主要来自两个因素——(1) 并行调用下游服务让单请求耗时减半；(2) 协程切换比线程切换轻量得多，4核可以轻松跑2000+并发协程。内存下降是因为协程栈比线程栈小得多（默认线程栈8MB vs 协程栈~几KB）。
7. 总结与建议
适用场景：如果你的API是I/O密集型（HTTP调用、数据库查询、文件读写），并且QPS卡在几百上不去，asyncio是首选方案。但如果是CPU密集型（图像处理、加密解密），asyncio帮助不大，应该用多进程。
三条经验：
1. 别在Flask里混用asyncio，要么全异步（aiohttp/FastAPI），要么全同步（gunicorn多worker）。混用会触发RuntimeError: no running event loop。
2. 连接池和信号量必须一起调优，只调一个参数没有意义。建议用locust或wrk做全参数矩阵压测。
3. 生产环境务必加uvloop，收益虽小但零成本。另外建议开启aiohttp的debug=True（仅测试环境）排查连接泄漏。
最后留个问题：如果你的下游服务是数据库操作，asyncio方案中应该用asyncpg还是aiomysql？两者性能差异巨大，我下一篇文章会专门对比。欢迎评论区讨论。
