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"}你看,把依赖上下文的数据提前抽取成普通值,任务函数就不需要关心请求了。这个习惯养成了,能避开90%的相关报错。
🔁 那还需要Celery吗?
可能有朋友会问:听说搞后台任务都用Celery,BackgroundTasks是不是太轻量了?
这好比出门买菜:如果只是楼下便利店买瓶酱油,你没必要发动汽车、等红绿灯、找停车位。
BackgroundTasks就是穿上拖鞋就能下楼的事儿,不需要消息队列,不需要额外进程,部署简单到哭。
但是它也有明确的局限:
📌 任务运行在同一个事件循环或线程池里,无法横向扩展
📌 适合轻量且不介意偶尔丢失的操作(服务重启任务就丢了)