FastAPI 后台任务：BackgroundTasks 的使用场景与注意事项

当前位置：点晴教程→知识管理交流 →『技术文档交流』

freeflydom

2026年5月6日 9:7 本文热度 35

你有没有碰到过这种情况：
用户点了注册按钮，结果页面转了好几秒才跳转，背后只是因为要给人家发一封“欢迎加入”的邮件。

用户心里肯定在骂：“这系统怕不是宕机了？” 可你也很无辜，明明只是一个HTTP请求，硬生生把发邮件、写日志这些家务活都干了。

今天咱们就来聊聊 FastAPI 给我们准备的一件贴心小棉袄：BackgroundTasks。
它能让你把耗时的小活儿挪到后台，接口嗖的一声就返回了，用户体验直接上一个台阶。

但别高兴太早，用不好也容易翻车。我先把自己踩过的坑摊开来，咱们边看边避。

🎯 后台任务能帮你干点啥？

先给没接触过的朋友打个比方。你在一家高级餐厅，点完菜之后，服务员马上把单子给你，说“已经安排上了”，你就不用站在那儿等厨师切菜、开火。

FastAPI里的后台任务就是那个贴心的服务员。它把你指定的耗时操作，比如：

📌 发送注册邮件
📌 写操作日志、审计记录
📌 更新缓存、清理临时文件
📌 给第三方推送通知

……统统扔到返回响应之后再去执行。

关键就在于：用户不用为这些“他自己感知不到有什么用”的事情买单。

⚙️ add_task 怎么玩？

用法简单到让人心虚。在你的路径操作函数里，后台任务会被自动注入，你只要往里面添加一个任务函数，并传好参数就行。

看一段最基础的注册发邮件的代码：

def send_welcome_email(email: str, username: str):
    # 假装在发邮件，这里睡一下模拟耗时
    import time; time.sleep(3)
    print(f"邮件已发送给 {email}")
@router.post("/register")
async def register(user: UserCreate, background_tasks: BackgroundTasks):
    # 保存用户逻辑……
    # 后台发送欢迎邮件
    background_tasks.add_task(send_welcome_email, user.email, user.username)
    return {"msg": "注册成功"}

用起来就这么直接。add_task 的第一个参数就是你要延迟执行的函数，后面跟着函数自己的参数，完全不用接触什么消息队列、外部依赖。

以前我刚学会这套时，兴奋得到处用，感觉省了一个亿的服务器资源。但是，很快就挨了现实一巴掌。

⚠️ 翻车实录：后台任务不能持有请求上下文

有个需求是：注册成功后，要把客户端的IP地址写到日志里。我当时脑子一热，直接把 request 对象塞进了后台任务：

# ❌ 错误示范，千万别学！
async def log_ip(request: Request):
    ip = request.client.host
    # 写日志...
    print(ip)
@router.post("/register")
async def register(request: Request, background_tasks: BackgroundTasks):
    background_tasks.add_task(log_ip, request)  # 坑！
    return {"msg": "ok"}

一跑就炸。报错跟你说 “Contextual object not available” 或者 Starlette 运行时的上下文丢失。

道理其实不复杂：
当响应返回后，FastAPI/Starlette 会把当前请求的上下文销毁。后台任务执行时，request对象的那些东西早就没了，你等于拿着一张过期饭票去窗口打饭，人家当然不给。

💡 正确姿势：提前把值抓出来

踩过坑之后，我的解法就一句话：在接口函数里，把所有后台任务需要的“食材”提前准备好，别直接递锅。

正确的写法是提取所需字段，比如IP、用户ID、邮件地址等等，把它们作为普通参数传进去：

def safe_log_ip(ip: str, user_id: int):
    print(f"User {user_id} 来自 IP: {ip}")
@router.post("/register")
async def register(request: Request, background_tasks: BackgroundTasks):
    ip = request.client.host
    user_id = 123  # 假设存完后得到ID
    background_tasks.add_task(safe_log_ip, ip, user_id)
    return {"msg": "ok"}