选FastAPI,因Flask不支持async/await,硬写会报RuntimeWarning或返回空响应;FastAPI原生异步、自动文档、强类型校验,性能与开发效率更优。

Flask 本身不支持 asyncio,硬套 async/await 会卡死
直接在 Flask 路由里写 async def 并用 await 调用异步函数,结果不是变快,而是整个请求阻塞、超时、日志里反复出现 RuntimeWarning: coroutine 'xxx' was never awaited。因为 Flask 是同步 WSGI 框架,它的工作线程根本不认识协程对象——它把 async def 当成普通函数返回一个 coroutine 对象,然后就扔给 WSGI 服务器(比如 Werkzeug 或 Gunicorn)去处理,后者直接忽略或报错。
常见错误写法:
from flask import Flask
app = Flask(__name__)
<h1>❌ 错误:Flask 不会 await 这个协程</h1><p>@app.route('/api')
async def bad_api():
await asyncio.sleep(1) # 这行永远不会执行
return "done"- Flask 的生命周期完全在同步上下文中,
request、session、g等对象都不是线程安全的,更不是协程安全的 - 哪怕你用
asyncio.run()包一层,也会触发新事件循环,导致连接池、全局状态丢失,且每次请求都新建 loop,开销爆炸 - 想“无缝升级”旧 Flask 项目?不存在的。必须换框架,或做明确分层隔离
真正可行的迁移路径:用 FastAPI / Starlette 替代 Flask 路由层
不是“优化 Flask”,而是用兼容性高、生态成熟的异步框架替换其 HTTP 入口。FastAPI 基于 Starlette(底层用 uvicorn),原生支持 async def 路由,且能复用大部分 Flask 业务逻辑——只要那些逻辑本身不依赖 Flask 特有上下文(如 request.args、flask.g)。
- HTTP 层:把
@app.route改成@app.get/@app.post,参数用 Pydantic 模型声明,自动解析 JSON、校验、生成 OpenAPI 文档 - 业务逻辑层:原有
get_user_by_id()、process_order()这类纯函数,只要不调用flask.request就可直接复用;若用了current_app或g,需改造成依赖注入或显式传参 - 数据库层:必须换成异步驱动,如
asyncpg(PostgreSQL)、aiomysql(MySQL)、motor(MongoDB);SQLAlchemy 2.0+的AsyncSession也可用,但别混用 sync/async session
示例迁移片段:
火焰围攻 授权: q15040409-creator.功能概述火焰围攻 授权: q15040409-creator.是一项面向实际任务的技能,主要用于Description.;Flask Web. - Python.;核心要点HTTP?;.Version.1.0.tags. wrap.github.automation.。它将相关步骤、工具调用和结果整理方式集中到统一流程中,帮助使用者更快完成目标并减少重复操作。使用与执行使用时应结合输入条件选择合适的执行方式,核对必要参数、依赖环境与输出内容,并按原始要
立即学习“Python免费学习笔记(深入)”;
# 旧 Flask
@app.route('/user/<int:user_id>', methods=['GET'])
def get_user(user_id):
user = db.query(User).filter(User.id == user_id).first()
return jsonify(user.to_dict())
<h1>新 FastAPI(复用 db 查询逻辑,但用 async session)</h1><p>@app.get("/user/{user_id}")
async def get_user(user_id: int, session: AsyncSession = Depends(get_db)):
stmt = select(User).where(User.id == user_id)
result = await session.execute(stmt)
user = result.scalar_one_or_none()
return user.to_dict() if user else {"error": "not found"}如果必须保留 Flask,只能把异步能力限定在“子任务”
某些场景下(如遗留系统无法整体替换),可以保留 Flask 主框架,仅对特定耗时 I/O 操作做异步化封装,但要注意边界和资源隔离。
- 不要在路由函数里
await,而是在后台启动独立事件循环 + 线程执行异步任务,用asyncio.run_coroutine_threadsafe()或loop.create_task()投递,再用concurrent.futures.Future等待结果 - 所有外部调用必须加超时:
await asyncio.wait_for(fetch_data(), timeout=3),否则一个慢请求拖垮整条线程 - 连接池不能复用:Flask 每次请求新建
aiohttp.ClientSession,会重复 TCP 握手,吞吐提升几乎为零;必须提取为模块级单例,配合atexit或app.teardown_appcontext关闭 - 别碰 Flask 的
g、session、request——它们不是协程局部变量,在异步线程里访问会出RuntimeError: Working outside of application context
性能提升的关键不在 async/await 语法,而在连接复用与并发控制
实测中,把 Flask 改成 FastAPI 后 QPS 提升 6–7 倍,真正起作用的不是 async 关键字,而是三件事:全局复用 aiohttp.ClientSession、数据库连接池大小调到 min_size=10, max_size=50、用 asyncio.Semaphore(100) 控制最大并发请求数。
- 没复用 session:每个请求新建连接 → TLS 握手耗时占总响应 40% 以上 → 吞吐卡在 300 QPS 上不去
- 没设 semaphore:1000 个并发请求全打到下游服务 → 触发限流或连接拒绝 → 大量
ConnectionResetError和重试放大负载 - 没关 debug 模式:
app.debug = True会让 uvicorn 启用重载监听器,每次请求都扫描文件变更 → CPU 占用翻倍
最后提醒一句:asyncio 不是银弹。如果你的瓶颈在 CPU(比如图像缩放、JSON 解析超大 payload),或者数据库慢在没索引、SQL 写得太烂,强行加 async 只会让问题更难定位。先用 cProfile 或 py-spy 看清哪一行吃掉 80% 时间,再决定要不要动 asyncio。

















