如何在Flask中先返回重定向响应再执行必需的后台耗时任务

根据你的需求，优先保证重定向响应立即返回、且耗时函数必须执行，可根据应用规模选择以下两种方案：

方案1：轻量后台线程（适合小型低并发应用） #

无需额外依赖，仅通过Python标准库的线程模块即可实现，耗时函数会在后台异步执行，不会阻塞响应返回：

from flask import redirect
import time, random, threading

def quick_processing_function():
    return True

def long_processing_function_that_may_fail():
    time.sleep(15)
    # 原示例笔误修正：random.rand改为random.random
    return random.random() < 0.99

# 原示例笔误修正：路由补全开头斜杠
@app.route('/target')
def target_view():
    result = quick_processing_function()
    if result:
        # 启动独立线程执行耗时任务
        task_thread = threading.Thread(target=long_processing_function_that_may_fail)
        # 设为守护线程，避免应用重启时被未完成的任务卡住进程退出
        task_thread.daemon = True
        task_thread.start()
    return redirect(SOME_EXTERNAL_SITE)

注意：该方案存在局限性，如果Flask进程重启、被回收，未执行完成的任务会丢失，也不支持失败重试，并发量高时线程开销会明显上升。

方案2：异步任务队列（适合生产级高并发应用） #

如果需要保证任务100%执行、支持失败重试、可观测任务状态，推荐使用Celery配合消息中间件实现：

1. 先安装依赖：
   pip install celery redis
2. 初始化Celery实例并注册任务：

# celery_config.py
from celery import Celery
import time, random

# 用Redis做消息中间件，也可替换为RabbitMQ
celery = Celery(
    'async_tasks',
    broker='redis://localhost:6379/0',
    backend='redis://localhost:6379/0'
)

# 把耗时函数注册为Celery任务，可配置重试规则
@celery.task(max_retries=3)
def long_processing_function_that_may_fail():
    time.sleep(15)
    return random.random() < 0.99

3. 修改视图函数，异步提交任务：

from celery_config import long_processing_function_that_may_fail

@app.route('/target')
def target_view():
    result = quick_processing_function()
    if result:
        # 提交任务到队列，立即返回
        long_processing_function_that_may_fail.delay()
    return redirect(SOME_EXTERNAL_SITE)

该方案的优势是任务持久化存储，不会因为Flask进程重启丢失，自动支持失败重试，不会占用Flask的请求worker资源，高并发下性能更稳定，缺点是需要额外部署消息中间件和Celery worker进程，维护成本略高。

不推荐使用Flask自带的after_this_request装饰器实现该需求，该装饰器虽然会在响应发送后执行代码，但仍会占用当前请求的worker进程，高并发场景下会快速耗尽worker资源，导致其他请求无法被处理。

内容的提问来源于stack exchange，提问作者Mark C.