FastAPI BackgroundTask失效求助：任务未在后台执行

使用FastAPI 0.99.1版本，年初BackgroundTask功能可正常工作，但当前添加的后台任务会像普通函数一样阻塞执行，必须等任务完成才返回响应，不符合「先返回响应、再后台执行任务」的预期。

测试排查代码 #

路由文件 (app/routers/pdf.py) #

from fastapi import APIRouter, BackgroundTasks, Depends, status

from app.config.dbconfig import get_db
from app.services.pdf import ServicePdf
from app.utils.service_result import ServiceResult, handle_result

router = APIRouter(
    prefix="/pdf",
    tags=["pdf"],
    responses={404: {"description": "Not found"}},
)


@router.post("/test", status_code=status.HTTP_201_CREATED)
async def parse_candidate_pdf(
    background_tasks: BackgroundTasks,
    db: get_db = Depends(),
):
    background_tasks.add_task(ServicePdf(db).test, "name")
    return handle_result(ServiceResult(None), status.HTTP_201_CREATED)


@router.post("/test-asyncio", status_code=status.HTTP_201_CREATED)
async def parse_candidate_pdf_asyncio(
    background_tasks: BackgroundTasks,
    db: get_db = Depends(),
):
    background_tasks.add_task(ServicePdf(db).test_asyncio, "name")
    return handle_result(ServiceResult(None), status.HTTP_201_CREATED)

服务文件 (app/services/pdf.py) #

import asyncio
import time

from app.services.main import AppService

class ServicePdf(AppService):
    def test(self, name):
        time.sleep(5)
        print("Awake now!", name)

    async def test_asyncio(self, name):
        await asyncio.sleep(5)
        print("Awake now!", name)

流程对比 #

• 实际执行流程：进入路由 → 执行完后台任务 → 返回响应
• 预期执行流程：进入路由 → 返回响应 → 后台任务继续执行

补充验证 #

新建仅包含基础逻辑的测试项目，同步、异步后台任务均可正常工作，排除FastAPI版本问题。

基础测试项目代码 (main.py) #

import time
import asyncio
from fastapi import FastAPI, BackgroundTasks

app = FastAPI()


async def process_data_asyncio(data: str):
    await asyncio.sleep(5)
    print(f"Processed data: {data}")


def process_data(data: str):
    time.sleep(5)
    print(f"Processed data: {data}")


@app.post("/process-data-asyncio/")
async def parse_asyncio(data: str, background_tasks: BackgroundTasks):
    background_tasks.add_task(process_data_asyncio, data)
    return {"message": "Data processing started in the background."}


@app.post("/process-data/")
async def parse(data: str, background_tasks: BackgroundTasks):
    background_tasks.add_task(process_data, data)
    return {"message": "Data processing started in the background."}


@app.get("/")
def home():
    return "api"

启动命令 #

uvicorn main:app --reload

排查方向 #

1. 数据库依赖的阻塞问题
   检查get_db依赖是否包含同步阻塞操作（比如同步数据库连接的建立）。如果get_db在获取连接时卡住，会导致整个请求流程阻塞，无法提前返回响应。建议：

   • 若使用异步数据库驱动（如asyncpg），确保get_db是异步生成器；
   • 若使用同步驱动，用asynccontextmanager配合ThreadPoolExecutor包装，避免阻塞事件循环。

2. 响应处理函数的阻塞逻辑
   检查handle_result或ServiceResult是否存在同步阻塞操作（如同步写入数据库、文件IO）。如果返回响应前需要等待这类操作完成，会导致后台任务被提前触发（FastAPI会在响应返回后调度后台任务，若响应过程被阻塞，任务可能被提前执行）。

3. 父类AppService的影响
   检查ServicePdf的父类AppService的__init__或初始化逻辑是否包含同步阻塞操作，比如实例化时执行了耗时的同步任务，导致ServicePdf(db)实例化过程卡住，进而阻塞整个请求流程。

4. Uvicorn运行配置差异
   对比当前项目与测试项目的Uvicorn启动参数，是否使用了不同的工作线程/进程配置（如--workers参数设置不当）。同步任务的调度依赖Uvicorn的线程池配置，若当前项目的线程池被耗尽或配置错误，会导致后台任务无法异步执行。

5. 全局中间件的阻塞
   检查项目中是否存在全局中间件，是否在请求处理流程中执行了同步阻塞代码，导致事件循环被卡住，无法及时返回响应并调度后台任务。

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