服务器重启后如何重新创建FastAPI中的后台任务

FastAPI的BackgroundTasks是绑定到单个请求上下文设计的，在启动事件（startup）里直接使用它不会生效——因为启动阶段没有请求，FastAPI不会自动处理这些任务。你需要直接用Python的异步/同步任务机制实现，无需依赖BackgroundTasks。

核心思路 #

1. 把推送通知的逻辑封装成独立的函数（同步或异步），脱离BackgroundTasks依赖
2. 在启动事件中从数据库读取待恢复的任务数据
3. 针对每个任务，根据预设的延迟时间，直接通过事件循环（异步）或线程（同步）启动任务

示例代码（异步场景） #

假设你的任务是异步函数，FastAPI应用采用异步模式：

from fastapi import FastAPI
import asyncio
from your_db_module import get_pending_notification_tasks  # 自定义数据库查询函数

app = FastAPI()

# 封装异步推送任务逻辑
async def send_push_notification(user_id: int, message: str, delay: int):
    await asyncio.sleep(delay)  # 等待指定时间
    # 这里写实际的推送逻辑，比如调用推送API
    print(f"给用户 {user_id} 发送推送：{message}")

@app.on_event("startup")
async def restore_background_tasks():
    # 从数据库获取所有未完成的待推送任务
    pending_tasks = await get_pending_notification_tasks()
    
    for task in pending_tasks:
        # 直接用asyncio.create_task启动异步任务
        asyncio.create_task(
            send_push_notification(
                user_id=task.user_id,
                message=task.message,
                delay=task.remaining_delay  # 数据库中存储的剩余等待时间
            )
        )
    print(f"已恢复 {len(pending_tasks)} 个后台推送任务")

示例代码（同步场景） #

如果你的推送逻辑是同步函数，或者FastAPI采用同步模式：

from fastapi import FastAPI
import threading
import time
from your_db_module import get_pending_notification_tasks

app = FastAPI()

# 封装同步推送任务逻辑
def send_push_notification(user_id: int, message: str, delay: int):
    time.sleep(delay)
    print(f"给用户 {user_id} 发送推送：{message}")

@app.on_event("startup")
def restore_background_tasks():
    pending_tasks = get_pending_notification_tasks()
    
    for task in pending_tasks:
        # 启动线程执行同步任务
        threading.Thread(
            target=send_push_notification,
            args=(task.user_id, task.message, task.remaining_delay),
            daemon=True  # 设置为守护线程，避免阻止服务器关闭
        ).start()
    print(f"已恢复 {len(pending_tasks)} 个后台推送任务")

关键注意事项 #

• 任务状态管理：数据库中要记录任务的执行状态（比如是否已完成），避免重启后重复执行。任务完成后记得更新数据库状态。
• 守护线程/任务：同步场景下用daemon=True，异步场景下无需额外设置（asyncio任务随事件循环结束而终止），确保服务器关闭时这些任务不会阻塞退出。
• 错误处理：建议在任务函数中添加异常捕获，避免单个任务失败导致整个逻辑崩溃。

内容的提问来源于stack exchange，提问作者Dmitriy Lunev