Python API如何返回成功后异步执行非阻塞长任务？

1. 修正原生Thread的使用方式 #

你之前用threading.Thread没成功，大概率是没处理好线程生命周期或者请求上下文的问题。以Flask为例，正确写法如下：

from flask import Flask, request
import threading
import time

app = Flask(__name__)

def long_running_task(user_id):
    # 这里写长耗时逻辑，注意不要直接引用请求上下文对象
    time.sleep(10)
    print(f"用户{user_id}的后台任务执行完毕")

@app.route('/submit', methods=['POST'])
def submit():
    # 接口验证逻辑
    token = request.json.get('token')
    if not token or token != 'valid_token':
        return '', 401
    
    user_id = request.json.get('user_id')
    # 启动后台线程，传入提前提取好的参数
    thread = threading.Thread(target=long_running_task, args=(user_id,))
    thread.daemon = True  # 可选：设置为守护线程，主进程退出时自动结束线程
    thread.start()  # 必须调用start才会启动线程
    
    # 立即返回204响应
    return '', 204

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

注意：如果任务需要用到请求数据，一定要提前把数据提取出来传给线程函数，不要在线程里直接访问request——请求结束后框架会销毁请求上下文，线程里再引用会报错。

2. 用ThreadPoolExecutor管理后台线程 #

手动创建线程不够灵活，用concurrent.futures.ThreadPoolExecutor可以更稳定地管理线程池：

from flask import Flask, request
from concurrent.futures import ThreadPoolExecutor
import time

app = Flask(__name__)
# 初始化线程池，设置最大工作线程数
executor = ThreadPoolExecutor(max_workers=3)

def long_running_task():
    time.sleep(10)
    print("后台任务执行完成")

@app.route('/submit', methods=['POST'])
def submit():
    # 验证逻辑...
    executor.submit(long_running_task)  # 提交任务到线程池
    return '', 204

3. 生产环境推荐：异步任务队列（Celery） #

如果是生产环境，原生线程方案有明显缺陷：web进程崩溃会丢失未完成任务，无法监控任务状态，也没法重试失败任务。这时用Celery配合消息队列（Redis/RabbitMQ）是更专业的选择：

步骤1：安装依赖 #

pip install celery redis

步骤2：配置Celery和Flask #

from flask import Flask, request
from celery import Celery
import time

app = Flask(__name__)
# 配置Redis作为消息队列和结果存储
app.config['CELERY_BROKER_URL'] = 'redis://localhost:6379/0'
app.config['CELERY_RESULT_BACKEND'] = 'redis://localhost:6379/0'

# 初始化Celery
celery = Celery(app.name, broker=app.config['CELERY_BROKER_URL'])
celery.conf.update(app.config)

# 定义异步任务
@celery.task
def long_running_task(user_id):
    time.sleep(10)
    print(f"用户{user_id}的异步任务执行完成")

@app.route('/submit', methods=['POST'])
def submit():
    # 接口验证
    if not request.json.get('token'):
        return '', 401
    
    user_id = request.json.get('user_id')
    # 提交任务到Celery队列，立即返回响应
    long_running_task.delay(user_id)
    return '', 204

步骤3：启动服务 #

• 先启动本地Redis服务
• 启动Celery Worker：celery -A app.celery worker --loglevel=info
• 启动Flask服务：python app.py

请求接口后会立刻返回204，任务由Celery Worker在后台独立执行，不受web进程的影响。

你之前Thread失败的可能原因 #

• 忘记调用thread.start()：创建Thread对象后必须调用start方法才会真正启动线程
• 在线程中直接使用请求上下文：比如在long_running_task里引用request对象，请求结束后上下文被销毁导致报错
• web服务器的多进程模式：如果用Gunicorn这类服务器的默认多进程模式，请求结束后进程可能被回收，导致线程被强制终止

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