42｜后台任务与 BackgroundTasks

某些接口需要做一些不紧急的附属操作：用户注册后发送欢迎邮件、删除资源后清理关联文件、操作完成后写审计日志。如果这些操作放在主流程里同步执行，接口响应时间会被拉长。FastAPI 的 BackgroundTasks 让这些操作在响应返回后再执行，不阻塞客户端等待。

一、基本用法

from fastapi import BackgroundTasks

async def send_email(email: str, message: str):
    await asyncio.sleep(2)   # 模拟发送邮件
    print(f"邮件已发送至 {email}: {message}")

@app.post("/users")
async def create_user(
    user: UserCreate,
    background_tasks: BackgroundTasks,
):
    new_user = save_user(user)

    # 注册后台任务，接口会立即返回，邮件在后台发送
    background_tasks.add_task(send_email, user.email, "欢迎注册")

    return {"id": new_user.id}

add_task(函数, *args, **kwargs) 把任务加入队列。响应发送给客户端后，FastAPI 按顺序执行队列中的任务。

二、使用场景

适合用后台任务的操作：

场景	原因
发送邮件/通知	第三方 SMTP 服务慢，用户不需要等待
写审计日志	日志写入失败不应该影响主业务
清理临时文件	文件系统操作可能慢
更新缓存	缓存刷新失败可降级读数据库
推送 Webhook	外部服务不可控，不能阻塞

不适合用后台任务的操作：

场景	原因
数据库主业务写入	失败必须告知用户，不能后台静默失败
需要立即返回结果的操作	后台任务的结果客户端收不到
高可靠性要求的操作	后台任务如果进程崩溃会丢失

三、任务失败的处理

BackgroundTasks 执行失败时：

1. 异常被打印到日志
2. 不影响已返回的响应
3. 不会自动重试

如果需要可靠性，改用专门的异步任务队列（如 Celery、RQ、APScheduler）：

# BackgroundTasks 版本：进程重启任务丢失
background_tasks.add_task(send_email, ...)

# Celery 版本：任务持久化到 Redis，worker 崩溃后重启继续执行
celery_app.send_task("send_email", args=[email, message])

四、与异步函数配合

async def log_operation(user_id: int, action: str):
    await db.execute(insert(AuditLog).values(user_id=user_id, action=action))

@app.delete("/servers/{id}")
async def delete_server(
    id: int,
    background_tasks: BackgroundTasks,
    user_id: str = Depends(get_current_user),
):
    db.execute(delete(Server).where(Server.id == id))
    db.commit()

    background_tasks.add_task(log_operation, user_id, f"delete_server:{id}")
    return {"deleted": id}

五、常见错误

在后台任务中抛异常不处理

async def risky_task():
    raise Exception("出错")

background_tasks.add_task(risky_task)
# 异常只打印日志，可能没人关注

# 改进：加 try/catch，错误时通知运维
async def risky_task():
    try:
        ...
    except Exception as e:
        logger.error("后台任务失败", exc_info=True)
        send_alert("后台任务失败", str(e))

把必须成功的操作放后台

# 错误：支付成功后更新订单状态放后台，失败没人知道
background_tasks.add_task(update_order_status, order_id)

# 正确：核心业务同步执行
update_order_status(order_id)   # 失败时抛异常，接口返回错误

后台任务依赖请求上下文

# 错误：请求结束后 db session 已关闭，后台任务无法使用
def background_job(db: Session):
    db.execute(...)   # Session 已关闭，报错

background_tasks.add_task(background_job, db)

# 正确：后台任务自己创建新的 session
def background_job():
    with get_db() as db:
        db.execute(...)

background_tasks.add_task(background_job)