一、问题背景：一个“慢”得不合理的订单接口
先描述一下场景：我们的订单服务（Python 3.10.7，Flask 2.2.3，Gunicorn 20.1.0，单机4核8G）有一个GET /api/orders?user_id=xxx接口，业务逻辑很简单：查Redis缓存（大约1ms）→ 未命中则查MySQL（大约10ms）→ 组装JSON返回。逻辑不复杂，但压测（wrk -t4 -c200 -d30s）数据很难看：
| 指标 | 改造前 |
|---|---|
| QPS | 502 |
| P99 延迟 | 812ms |
| 平均延迟 | 245ms |
| CPU 利用率 | 31% |
典型症状：CPU不高但响应慢，说明线程被IO阻塞了。Gunicorn默认用sync worker，一个worker同时只能处理一个请求。即使开了--workers 8，8个线程全部阻塞在Redis/MySQL的IO等待上，性能自然上不去。
二、环境与版本：为什么选asyncio而不是gevent
先交代环境，方便复现：
- 操作系统：Ubuntu 22.04 LTS
- Python：3.10.7（CPython，非PyPy）
- 框架：Flask 2.2.3 + Gunicorn 20.1.0
- 异步驱动：
redis.asyncio（redis-py 4.5.1）、aiomysql0.1.1 - 压测工具：wrk 4.2.0
关于选型，我特意对比过gevent。gevent用猴子补丁把标准库的socket换成协程版本，侵入小但不可控——一旦某个第三方库内部用了C扩展阻塞调用，照样卡住整个事件循环。asyncio虽然要改代码，但你能清楚看到每个await点，可控性高很多。而且Python 3.10的asyncio已经很成熟，官方文档也推荐。所以我选择asyncio。
三、方案设计：改造Flask路由为异步协程
核心思路：把Flask的同步视图函数改成异步函数，内部用await调用异步Redis/MySQL驱动。Gunicorn需要换一个支持asyncio的worker，这里用uvicorn的UvicornWorker，因为它是基于uvloop的，性能比纯asyncio的worker高约15%。
改造后的架构：
客户端 -> Gunicorn(UvicornWorker) -> ASGI -> Flask异步视图 -> asyncio.gather并发调用Redis/MySQL
注意：Flask本身不支持ASGI，需要用一个适配层。我用了flask[async]（Flask 2.0+自带），它内部把WSGI请求包装成ASGI作用域。但有个坑：Flask的异步视图函数必须返回一个Response对象，不能直接返回字符串，否则会报错。
四、核心实现：Before/After代码对比
Before（同步版本，Gunicorn sync worker）：
# app_sync.py
import time
import redis
import pymysql
from flask import Flask, jsonify, request
app = Flask(__name__)
r = redis.Redis(host='localhost', port=6379, db=0)
def query_mysql(user_id):
conn = pymysql.connect(host='localhost', user='root', password='123456', database='orders')
try:
with conn.cursor() as cursor:
cursor.execute("SELECT order_id, amount FROM orders WHERE user_id=%s ORDER BY created_at DESC LIMIT 5", (user_id,))
return cursor.fetchall()
finally:
conn.close()
@app.route('/api/orders')
def get_orders():
user_id = request.args.get('user_id')
# 1. 查缓存
cache_key = f"orders:{user_id}"
cached = r.get(cache_key)
if cached:
return jsonify(eval(cached)) # 注意：这里用eval仅作演示，生产环境用json
# 2. 缓存未命中，查MySQL
time.sleep(0.01) # 模拟网络延迟
rows = query_mysql(user_id)
# 3. 写回缓存
r.set(cache_key, str(rows), ex=60)
return jsonify(rows)
if __name__ == '__main__':
app.run(port=5000)
启动命令：gunicorn -w 8 -k sync app_sync:app -b 0.0.0.0:5000
After（异步版本，UvicornWorker + asyncio）：
# app_async.py
import asyncio
import json
from flask import Flask, jsonify, request
from redis.asyncio import Redis
import aiomysql
app = Flask(__name__)
r = Redis(host='localhost', port=6379, db=0, decode_responses=True)
# 全局连接池，避免每个请求都创建连接
pool = None
async def init_mysql_pool():
global pool
pool = await aiomysql.create_pool(host='localhost', user='root', password='123456',
database='orders', minsize=5, maxsize=20,
autocommit=True, loop=asyncio.get_event_loop())
async def query_mysql_async(user_id):
async with pool.acquire() as conn:
async with conn.cursor() as cursor:
await cursor.execute("SELECT order_id, amount FROM orders WHERE user_id=%s ORDER BY created_at DESC LIMIT 5", (user_id,))
rows = await cursor.fetchall()
# 转成字典列表
return [{'order_id': row[0], 'amount': row[1]} for row in rows]
@app.route('/api/orders')
async def get_orders():
user_id = request.args.get('user_id')
cache_key = f"orders:{user_id}"
# 1. 异步查缓存
cached = await r.get(cache_key)
if cached:
return jsonify(json.loads(cached))
# 2. 缓存未命中，异步查MySQL，同时写回缓存
rows = await query_mysql_async(user_id)
await r.set(cache_key, json.dumps(rows), ex=60)
return jsonify(rows)
@app.before_serving
async def setup():
await init_mysql_pool()
if __name__ == '__main__':
import uvicorn
uvicorn.run(app, host='0.0.0.0', port=5000, loop='uvloop')
启动命令：gunicorn -w 4 -k uvicorn.workers.UvicornWorker app_async:app -b 0.0.0.0:5000
注意几个关键点：
1. @app.before_serving是Flask 2.2+新增的异步钩子，用于初始化连接池。如果不用这个，直接放在模块顶层await会报错，因为create_pool需要一个运行中的事件循环。
2. aiomysql.create_pool必须传loop=asyncio.get_event_loop()，否则在Gunicorn worker中会因事件循环不一致而报错。
3. worker数量从8降到4，因为每个worker现在可以并发处理大量请求，不再需要那么多进程。
五、踩坑与优化：三个真实遇到的大坑
坑1：事件循环被阻塞——time.sleep是魔鬼
我最初在异步版本里保留了time.sleep(0.01)模拟延迟，结果性能比同步还差。原因：time.sleep是阻塞调用，会卡住整个事件循环，所有并发请求排队等待。必须改用await asyncio.sleep(0.01)。这个错误很隐蔽，因为代码能跑通，但压测数据会让你怀疑人生。
坑2：协程泄漏——未关闭的异步连接
第一次压测时，内存持续上涨，最终OOM。排查发现是aiomysql的pool.acquire()异常时没有释放连接。正确做法是用async with上下文管理器，确保连接归还。另外，Redis连接也要注意decode_responses=True，否则返回的是bytes，JSON序列化会报错。
坑3：混合框架兼容——Flask的jsonify在异步视图里返回非标准对象
Flask 2.2的异步视图要求返回Response对象，但jsonify返回的是Response的子类，没问题。真正的问题是：如果视图函数里有await，Flask会强制走ASGI路径，此时请求上下文（request、g）的访问方式变了。我遇到过一次RuntimeError: Working outside of application context，原因是视图函数内部用了threading.local存数据。解决方案：不要跨await访问request对象，在await之前取出所有需要的参数。
优化：引入asyncio.Queue做限流
压测发现，当并发超过200时，MySQL连接池被打满，报Too many connections。加了一个简单的信号量：
sem = asyncio.Semaphore(50) # 最多同时50个MySQL查询
async def query_mysql_async(user_id):
async with sem:
async with pool.acquire() as conn:
# ... 原有代码
这样既能保护数据库，又不会像线程锁那样死等。
六、效果数据：改造前后对比
压测条件完全一致（wrk -t4 -c200 -d30s，4核8G虚拟机）：
| 指标 | 改造前（sync worker） | 改造后（uvicorn worker） | 提升 |
|---|---|---|---|
| QPS | 502 | 1536 | 3.06x |
| P99 延迟 | 812ms | 213ms | 3.81x |
| 平均延迟 | 245ms | 87ms | 2.82x |
| CPU 利用率 | 31% | 78% | 2.5x |
| 内存占用 | 380MB | 420MB | +10% |
额外测试：用asyncio.gather并发调用Redis和MySQL（缓存未命中场景），把两次网络IO合并为一次等待，P99再降15%：
# 缓存未命中时，并行查MySQL和设置缓存
rows, _ = await asyncio.gather(
query_mysql_async(user_id),
r.set(cache_key, json.dumps(rows), ex=60) # 注意：这里rows还没拿到，实际应该用future
)
正确写法是：
rows_future = query_mysql_async(user_id)
rows = await rows_future
await r.set(cache_key, json.dumps(rows), ex=60)
因为set依赖rows的结果，不能并行。所以这个优化只适用于查询和写缓存互相独立的情况——比如“查询MySQL”和“发送日志到Kafka”可以并行。
七、总结：asyncio不是银弹，但值得一试
这轮改造的核心收益来自两点：一是把8个阻塞线程换成4个事件循环，IO等待不再占满线程；二是用了uvloop（uvicorn默认），事件循环本身提速约10%。如果你也想做类似优化，我的建议是：
- 先确认瓶颈是IO等待。用
cProfile或py-spy看一下，如果CPU利用率高但延迟高，那是计算密集，换asyncio没用，该用多进程。 - 不要全量改造。我们只改了这一个接口，其他同步接口原样保留。Gunicorn支持混合worker，但实际部署时建议拆分成两个服务，避免一个接口阻塞事件循环拖死全局。
- 压测务必看P99。平均延迟低但P99高，说明有长尾请求，通常是因为事件循环被某个慢操作（比如没有超时控制的MySQL查询）卡住。我的经验是给所有
await加asyncio.wait_for超时，比如数据库查询3秒超时，避免拖垮整个服务。
最后，如果你遇到类似“CPU不高但响应慢”的Python服务，先检查连接池配置（maxsize是不是太小）、DNS解析是不是同步的（aiodns）、以及有没有在异步路由里混用同步库。asyncio改造的收益在IO密集场景下非常可观，但坑也很多，建议先在非核心接口上试点。
以上，欢迎在评论区讨论遇到的问题。代码已上传GitHub（链接省略），可复现完整压测流程。
