在 Web API 的实际开发中，经常有一些操作不需要阻塞主请求的响应——比如发送邮件、写审计日志、触发数据同步等。FastAPI 提供了内置的 BackgroundTasks 机制，允许你在返回响应之后异步执行这些轻量级后台任务，而无需引入 Celery 等重量级任务队列。本文系统讲解 BackgroundTasks 的核心知识点与最佳实践。

为什么需要后台任务

考虑一个用户注册接口：主流程是验证数据、创建用户、返回 201。如果在此过程中还要「发送欢迎邮件」，直接在主请求里调用邮件服务会拖慢响应速度——用户得等邮件发完才能看到”注册成功”。更优雅的做法是：先返回响应，再后台发送邮件。

BackgroundTasks 正是为此而生。它适合执行不依赖返回值、允许异步完成的操作。

基本用法

注册最简单的后台任务

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
from fastapi import FastAPI, BackgroundTasks

app = FastAPI()


def write_log(message: str):
    """模拟一个耗时的日志写入操作"""
    with open("app.log", "a") as f:
        f.write(f"{message}\n")


@app.post("/send-notification/{email}")
async def send_notification(email: str, background_tasks: BackgroundTasks):
    # 将 write_log 注册为后台任务，传入参数
    background_tasks.add_task(write_log, f"通知已发送至 {email}")
    # 主逻辑立即返回，后台任务在响应返回后执行
    return {"message": "通知发送成功"}

关键点：

  1. 在路由函数参数中声明 background_tasks: BackgroundTasks，FastAPI 会自动注入；
  2. 通过 add_task(task_func, *args, **kwargs) 注册后台任务；
  3. FastAPI 在响应已返回给客户端之后才执行这些任务。

添加多个后台任务

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
@app.post("/users/register")
async def register_user(
    username: str,
    email: str,
    background_tasks: BackgroundTasks,
):
    # 模拟创建用户...
    user_id = 12345

    # 注册多个后台任务
    background_tasks.add_task(send_welcome_email, email, username)
    background_tasks.add_task(write_audit_log, user_id, "register")
    background_tasks.add_task(update_search_index, user_id)

    return {"message": "注册成功", "user_id": user_id}


async def send_welcome_email(email: str, username: str):
    """发送欢迎邮件（模拟耗时操作）"""
    await asyncio.sleep(2)  # 模拟网络延迟
    print(f"已发送欢迎邮件至 {email}，欢迎 {username}！")


def write_audit_log(user_id: int, action: str):
    """写入审计日志"""
    with open("audit.log", "a") as f:
        f.write(f"[{__import__('datetime').datetime.now()}] user_id={user_id} action={action}\n")


def update_search_index(user_id: int):
    """更新搜索索引"""
    print(f"正在更新用户 {user_id} 的搜索索引...")

多个任务会按照 add_task 的注册顺序依次执行。

支持异步函数和同步函数

add_task 同时支持同步函数（def）和异步函数（async def），框架会自动适配：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
# 同步任务
def sync_task(param: str):
    print(f"同步任务：{param}")

# 异步任务
async def async_task(param: str):
    await asyncio.sleep(1)
    print(f"异步任务：{param}")

@app.get("/mixed-tasks")
async def mixed_tasks(background_tasks: BackgroundTasks):
    background_tasks.add_task(sync_task, "hello")
    background_tasks.add_task(async_task, "world")
    return {"status": "ok"}

建议：如果后台任务有 I/O 操作（数据库写入、HTTP 请求、文件写入），用 async def 以获得更好的并发性能。

在依赖中使用 BackgroundTasks

有时候后台任务需要和依赖注入结合——比如在依赖函数中根据某些条件注册任务。可以在依赖函数里注入 BackgroundTasks：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
from fastapi import Depends


def audit_dependency(
    background_tasks: BackgroundTasks,
    x_request_id: str | None = None,
):
    """作为依赖，自动为每个请求添加审计日志"""
    if x_request_id:
        background_tasks.add_task(
            write_log, f"request_id={x_request_id} 已处理"
        )
    return x_request_id


@app.get("/api/data")
async def get_data(request_id: str = Depends(audit_dependency)):
    return {"data": "some data", "request_id": request_id}

当请求带有 x-request-id 头时，响应返回后自动写入日志。这非常适合「全局审计」、「请求耗时记录」等场景。

实战场景

场景一：注册后发送验证邮件

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
from pydantic import BaseModel, EmailStr

class RegisterRequest(BaseModel):
    username: str
    email: EmailStr
    password: str


@app.post("/register")
async def register(
    body: RegisterRequest,
    background_tasks: BackgroundTasks,
):
    # 1. 写入数据库（主流程）
    user = await db.create_user(body.username, body.email, body.password)

    # 2. 异步发送验证邮件
    background_tasks.add_task(
        send_verification_email,
        to_email=body.email,
        user_id=user.id,
        token=user.generate_token(),
    )

    return {"code": 201, "message": "注册成功，验证邮件已发送"}

场景二：数据导入后异步处理

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
import csv
from io import StringIO


@app.post("/users/import")
async def import_users(
    file: UploadFile,
    background_tasks: BackgroundTasks,
):
    content = await file.read()
    reader = csv.DictReader(StringIO(content.decode()))
    users = [row for row in reader]

    # 先批量写入
    await db.bulk_insert_users(users)

    # 后台处理：数据清洗、索引重建
    background_tasks.add_task(clean_imported_data, users)
    background_tasks.add_task(rebuild_search_index)

    return {"message": f"已导入 {len(users)} 条数据，后台处理中"}

场景三：聚合多个后台操作

当后台逻辑复杂时，建议把多个操作封装成一个函数：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
async def post_register_workflow(user_id: int, email: str, username: str):
    """注册后的完整工作流"""
    await send_welcome_email(email, username)
    await record_analytics_event("user_signup", user_id)
    await invalidate_cache("top_users")


@app.post("/register")
async def register(
    body: RegisterRequest,
    background_tasks: BackgroundTasks,
):
    user = await db.create_user(**body.dict())
    # 一个 entry point，内部串行
    background_tasks.add_task(post_register_workflow, user.id, body.email, body.username)
    return {"message": "注册成功"}

这样路由函数保持简洁，复杂逻辑收敛在工作流函数中。

注意事项与限制

1. 后台任务在响应返回后执行，但仍在同一个进程内

BackgroundTasks 运行在当前 Web 进程的事件循环中。如果进程崩溃或重启（如容器重启、uvicorn 被 kill），未完成的任务会丢失。它不提供持久化保证。

2. 不适合长时间运行的任务

如果一个后台任务需要运行几分钟，它会阻塞同一进程中的其他后台任务。对于长耗时任务，应改用 Celery、ARQ、Dramatiq 等任务队列。

3. 不能返回结果给客户端

后台任务的返回值无法传递给客户端——响应在任务执行之前就已发出。如果需要让客户端感知结果，改用 WebSocket 或轮询 + 状态表（pending/processing/done）。

4. 不要在后台任务中修改已返回的响应

响应对象在 BackgroundTasks 运行前就已经序列化并发送，此时修改路由函数中返回的字典或 Pydantic 对象不会有任何效果。

5. 与 Starlette BackgroundTask 的关系

FastAPI 的 BackgroundTasks 底层来自 Starlette。如果你需要更精细的控制，可以直接使用 starlette.background.BackgroundTask：

1
2
3
4
5
6
7
8
9
10
11
from starlette.background import BackgroundTask
from fastapi.responses import JSONResponse


@app.get("/custom-bg")
async def custom_bg():
    task = BackgroundTask(write_log, "自定义背景任务")
    return JSONResponse(
        content={"status": "ok"},
        background=task,
    )

这在需要动态决定返回内容＋附加后台任务时比较有用。

何时改用 Celery / 任务队列

场景 推荐方案
发送邮件、短信、推送通知 BackgroundTasks
写入日志 / 审计记录 BackgroundTasks
简单的缓存刷新 BackgroundTasks
耗时 < 5 秒的数据处理 BackgroundTasks
定时任务（cron） Celery Beat / APScheduler
重试 + 失败补偿 Celery / ARQ
大规模批量处理 Celery / Dramatiq
需要任务进度追踪 Celery + 状态表

简单来说：**BackgroundTasks 适合「发完响应后顺手做掉」的轻量操作；一旦涉及可靠性、重试、调度，就用专业的任务队列。**

小结

BackgroundTasks 是 FastAPI 中最轻量级的异步执行方式：

  1. 在路由参数中声明 background_tasks: BackgroundTasks；
  2. 用 add_task(func, *args, **kwargs) 注册后台任务；
  3. 任务在响应已发送后执行，支持 async def 和 def；
  4. 适合发送通知、写日志、简单数据清洗等非关键路径操作；
  5. 不要用于需要持久化、重试或长时间运行的任务——那些交给 Celery。

掌握 BackgroundTasks 后，你可以轻松优化接口响应速度，把耗时操作从「阻塞」变为「后台处理」，提升用户体验和系统吞吐量。