FastAPI BackgroundTasks 极简学习手册

一、一句话搞懂BackgroundTasks

BackgroundTasks就是：用户请求后先快速返回结果，然后在后台悄悄完成那些不着急的工作。

就像你在奶茶店点单，服务员先给你一个取餐号（快速响应），然后再慢慢给你做奶茶（后台任务）。

二、适用场景（什么时候用）

✅ 适合做的事情：

发邮件（用户不需要等邮件发完才能看到注册成功）
写日志（用户不需要等日志写完就能继续操作）
处理数据（上传文件后立即返回，后台慢慢转码）
清理临时文件（用户下载完文件后，后台删除临时文件）

❌ 不适合做的事情：

支付处理（需要实时确认结果的关键任务）
大型数据计算（耗时太长会占用服务器资源）

三、核心区别：阻塞 vs 非阻塞

类型	通俗理解	像什么场景	执行时间
阻塞执行	排队一个一个做	食堂打饭，一个人打完下一个才能打	总时间=所有任务时间相加
非阻塞执行	同时做多个任务	手机后台同时下载多个App	总时间=最慢的那个任务时间

1. 阻塞执行（默认方式）

用普通函数 def 定义任务，任务会按顺序执行，前一个任务做完才能做下一个。

from fastapi import FastAPI, BackgroundTasks
import time

app = FastAPI()

def make_coffee(coffee_type):
    print(f"开始做{coffee_type}...")
    time.sleep(2)  # 模拟做咖啡需要2秒
    print(f"{coffee_type}做好啦！")

@app.post("/order")
async def order_coffee(background_tasks: BackgroundTasks):
    background_tasks.add_task(make_coffee, "拿铁")
    background_tasks.add_task(make_coffee, "美式")
    return {"message": "你的订单已提交，咖啡正在制作中"}

执行结果：先做拿铁（2秒），再做美式（2秒），总耗时4秒。

2. 非阻塞执行（并发方式）

用异步函数 async def 定义任务，配合 asyncio.create_task() 实现多个任务同时执行。

from fastapi import FastAPI, BackgroundTasks
import asyncio

app = FastAPI()

async def download_file(file_name):
    print(f"开始下载{file_name}...")
    await asyncio.sleep(3)  # 模拟下载需要3秒
    print(f"{file_name}下载完成！")

def start_download(file_name):
    asyncio.create_task(download_file(file_name))  # 启动异步任务

@app.post("/download")
async def download_files(background_tasks: BackgroundTasks):
    background_tasks.add_task(start_download, "电影.mp4")
    background_tasks.add_task(start_download, "音乐.mp3")
    return {"message": "下载任务已提交，后台同时下载"}

执行结果：电影和音乐同时开始下载，总耗时3秒（和最慢的任务时间一样）。

3. CPU密集型任务处理（避免阻塞主线程）

对于像大量计算这样的CPU密集型任务，建议放到线程池中执行，避免影响服务器响应速度。

from fastapi import FastAPI, BackgroundTasks
from fastapi.concurrency import run_in_threadpool

app = FastAPI()

def heavy_calculation():
    print("开始进行复杂计算...")
    result = sum(range(10**7))  # 模拟大量计算
    print(f"计算完成，结果是{result}")

@app.post("/calculate")
async def run_calculation(background_tasks: BackgroundTasks):
    background_tasks.add_task(run_in_threadpool, heavy_calculation)
    return {"message": "计算任务已提交，后台进行中"}

四、关键注意事项（避坑指南）

异步函数不能直接用：不能直接把 async def 函数传给 add_task()，需要用 asyncio.create_task() 包装一下。
任务超时问题：BackgroundTasks适合做轻量级任务（通常<3秒），超过5秒的任务建议用Celery等专业工具。
异常不会影响主线程：后台任务执行失败不会影响用户请求，但也不会通知你，所以最好在任务内部自己处理异常和记录日志。
不要依赖请求资源：后台任务不能直接访问已关闭的数据库连接或请求上下文，要自己管理需要的资源。