在Python中如何使用Flask 2.x提供的async功能提升高并发性能？-Python教程-PHP中文网

async视图必须配Uvicorn，flask run会静默退化为同步执行；需用uvicorn启动、禁用gunicorn/gevent；async视图中禁用requests，改用httpx.AsyncClient或aiohttp；SQLAlchemy须配置async_engine和AsyncSession；before_request不支持async，应改用ASGI中间件或视图内处理。

在python中如何使用flask 2.x提供的async功能提升高并发性能？

async视图必须配Uvicorn，flask run会静默退化

写了async def视图函数，但用flask run启动，它不会报错，也不会真正异步——而是降级为同步执行，await语句被当成普通调用，整个请求仍占一个线程，毫无并发收益。

根本原因是flask run基于Werkzeug（WSGI），不支持ASGI事件循环。你必须换服务器：

安装uvicorn：pip install uvicorn
启动命令必须是uvicorn main:app（假设main.py里定义了app）
禁用gunicorn --worker-class sync或gevent——它们仍是WSGI模型，无法调度async协程

运行后检查日志：若看到Uvicorn running on http://...且无RuntimeWarning: Running asyncio.run() in a loop that's already running，说明环境就绪。

async视图里不能用requests，必须换aiohttp或httpx.AsyncClient

这是最常踩的坑：requests.get()是阻塞调用，一旦出现在async视图中，会卡死整个Uvicorn worker的事件循环，所有后续请求排队等待，表现就是“突然全接口变慢”或“503超时”。

正确做法是使用原生异步HTTP客户端：

aiohttp（需手动管理session生命周期）
httpx.AsyncClient（推荐，API接近requests，支持HTTP/2）

示例：

import httpx

@app.route("/api")
async def api_proxy():
    async with httpx.AsyncClient() as client:
        resp = await client.get("https://httpbin.org/json")
        return resp.json()

注意：不要在全局创建AsyncClient实例并复用——它不是线程安全的，应在每次请求内新建或通过依赖注入管理生命周期。

Flask Wrap

火焰围攻授权: q15040409-creator.功能概述火焰围攻授权: q15040409-creator.是一项面向实际任务的技能，主要用于Description.；Flask Web. - Python.；核心要点HTTP?；.Version.1.0.tags. wrap.github.automation.。它将相关步骤、工具调用和结果整理方式集中到统一流程中，帮助使用者更快完成目标并减少重复操作。使用与执行使用时应结合输入条件选择合适的执行方式，核对必要参数、依赖环境与输出内容，并按原始要

下载

SQLAlchemy要配async_engine，不能直接用db.session.query()

Flask-SQLAlchemy默认是同步ORM，db.session.query(User).all()会阻塞事件循环。即使你用了Uvicorn，这行代码也会让整个worker停摆。

必须升级到SQLAlchemy 2.0+，并显式配置异步引擎：

数据库驱动换成asyncpg（PostgreSQL）或aiomysql（MySQL）
创建create_async_engine，URL前缀为postgresql+asyncpg://
用AsyncSession替代session，所有查询方法加await

片段示意：

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession

engine = create_async_engine("postgresql+asyncpg://...")
async_session = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)

@app.route("/users")
async def list_users():
    async with async_session() as session:
        users = await session.execute(select(User))
        return [u[0].to_dict() for u in users.all()]

before_request这类钩子不支持async，逻辑得收进视图里

你不能写@app.before_request + async def auth_check()——Flask不会await它，只会把协程对象当普通函数调用，导致未消费的协程堆积、内存缓慢增长，甚至引发RuntimeWarning。

可行方案只有两个：

把预处理逻辑（如JWT校验、权限检查）直接写在每个async视图开头，用await调用
改用ASGI中间件（如uvicorn.middleware.proxy_headers.ProxyHeadersMiddleware），它天然支持async

另外注意：g对象在async视图中可用，但它的生命周期仅限当前请求协程，跨await边界没问题；可放心用g.user = await get_current_user()。

真正的并发收益不在“写了async”，而在整条链路是否全异步：从服务器（Uvicorn）、HTTP客户端（httpx.AsyncClient）、数据库驱动（asyncpg）、到ORM层（SQLAlchemy AsyncSession）。任一环节混入同步调用，都会拖垮整个worker。别指望“大部分异步+一个requests”能提升吞吐——它只会更糟。

Python免费学习笔记（深入）：立即使用
在学习笔记中，你将探索 Python 的核心概念和高级技巧！