如何实现Flask API异步后台请求？解决同步阻塞问题

你的代码问题在于直接await异步函数导致请求阻塞——Flask的异步路由虽然支持async/await语法，但await asyncio.gather(...)会强制等待异步任务完成才返回响应，完全失去了后台执行的意义。下面提供两种可行的解决方案：

方案1：使用异步任务提交（需异步服务器支持） #

利用asyncio.create_task将异步任务提交到事件循环，不等待其完成直接返回响应。但需要用支持异步的WSGI服务器（如Hypercorn、Uvicorn）替代Flask默认的Werkzeug服务器。

修改后的代码：

from flask import Flask, jsonify
import asyncio

app = Flask(__name__)

async def async_function():
    await asyncio.sleep(5)
    print("Background function completed")
    
@app.route('/async')
async def async_route():
    # 提交任务到事件循环，不等待完成
    asyncio.create_task(async_function())
    return jsonify({"result": "Function registered to run in the background after 5 seconds."})

if __name__ == '__main__':
    # 注意：这里改用hypercorn启动（需先安装：pip install hypercorn）
    import hypercorn
    hypercorn.run(app, host="127.0.0.1", port=5000)

运行方式：

1. 安装依赖：pip install hypercorn
2. 执行脚本，或直接用命令行：hypercorn your_app_file_name:app

方案2：线程池执行异步任务（兼容默认服务器） #

如果不想更换服务器，可以用ThreadPoolExecutor将异步任务放到独立线程中运行，主线程（处理请求）直接返回响应，完全不阻塞。

修改后的代码：

from flask import Flask, jsonify
import asyncio
from concurrent.futures import ThreadPoolExecutor

app = Flask(__name__)
# 初始化线程池，可根据需求调整线程数
executor = ThreadPoolExecutor(max_workers=4)

async def async_function():
    await asyncio.sleep(5)
    print("Background function completed")

def run_async_task():
    # 为线程创建独立的事件循环并执行异步任务
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    loop.run_until_complete(async_function())
    loop.close()

@app.route('/async')
def async_route():
    # 提交任务到线程池
    executor.submit(run_async_task)
    return jsonify({"result": "Function registered to run in the background after 5 seconds."})

if __name__ == '__main__':
    app.run(debug=True)

关键说明 #

• 方案1更符合异步编程范式，但依赖异步服务器；方案2兼容性更强，适合快速改造现有同步Flask项目。
• 生产环境中如果任务复杂（如需要持久化、重试），可以考虑用Celery等任务队列框架，但上述两种方案足以应对轻量后台任务需求。

内容的提问来源于stack exchange，提问作者kfirwe