如何让FastAPI在客户端断开后仍完成StreamingResponse的流函数运行？

方法1：用Python threading模块创建独立线程 #

直接把生成器的执行逻辑放到单独线程里，脱离请求上下文绑定，客户端断开连接后线程仍会继续跑完。

示例代码：

import threading
from fastapi import FastAPI

app = FastAPI()

def run_generator():
    # 这里替换成你的生成器逻辑，比如耗时的数据处理
    for i in range(10):
        print(f"处理进度：{i+1}/10")
        # 模拟耗时操作
        import time
        time.sleep(1)
    print("生成器任务执行完毕")

@app.get("/start-task")
async def start_task():
    # 启动非守护线程，避免主线程退出时被强制终止
    thread = threading.Thread(target=run_generator, daemon=False)
    thread.start()
    return {"message": "任务已启动，将在后台持续执行"}

如果需要批量处理任务，建议用concurrent.futures.ThreadPoolExecutor来管理线程，防止线程泄漏。

方法2：用asyncio创建脱离请求上下文的异步任务 #

如果你的生成器是异步实现的，可以把任务提交到全局事件循环，脱离请求的取消绑定。

示例代码：

import asyncio
from fastapi import FastAPI

app = FastAPI()
# 获取ASGI服务器的全局事件循环
loop = asyncio.get_event_loop()

async def async_generator():
    for i in range(10):
        print(f"异步处理进度：{i+1}/10")
        await asyncio.sleep(1)
    print("异步生成器任务执行完毕")

@app.get("/start-async-task")
async def start_async_task():
    # 提交任务到全局循环，不受请求取消影响
    loop.create_task(async_generator())
    return {"message": "异步任务已启动，后台继续执行"}

这种方式只适用于异步生成器，且依赖ASGI服务器（比如uvicorn）的全局事件循环稳定性。

方法3：用multiprocessing创建子进程 #

如果生成器是CPU密集型任务，用子进程更合适——进程完全独立，不受主进程中请求取消的影响。

示例代码：

import multiprocessing
from fastapi import FastAPI

app = FastAPI()

def cpu_bound_generator():
    total = 0
    # 模拟CPU密集计算
    for i in range(1_000_000):
        total += i
    print(f"CPU密集任务完成，计算结果：{total}")

@app.get("/start-process-task")
async def start_process_task():
    process = multiprocessing.Process(target=cpu_bound_generator)
    process.start()
    return {"message": "CPU密集任务已启动，后台执行"}

如果需要传递任务结果，可以用multiprocessing.Queue或Pipe；注意不要创建过多子进程，避免耗尽系统资源。

关键注意事项 #

• 所有后台任务都要加异常捕获，避免崩溃导致资源泄漏。
• 如果需要保存任务结果或状态，直接写入本地文件、数据库即可，不用额外中间件。
• 生产环境建议给后台任务加日志记录，方便排查问题。

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