如何在FastAPI（Uvicorn服务器）中实现后台调度任务？

一、直接在FastAPI中集成调度器的可行方案 #

你可以使用APScheduler库直接在FastAPI中集成后台调度任务，这是轻量场景下的首选方案，步骤如下：

1. 安装依赖

pip install apscheduler fastapi uvicorn

2. 代码实现

from fastapi import FastAPI
from apscheduler.schedulers.asyncio import AsyncIOScheduler
from datetime import datetime

app = FastAPI()
# 适配FastAPI异步环境的调度器
scheduler = AsyncIOScheduler(timezone="Asia/Shanghai")

# 定义后台任务
def cleanup_temp_data():
    print(f"清理临时数据任务执行于: {datetime.now()}")

def sync_external_data():
    print(f"同步外部数据任务执行于: {datetime.now()}")

# 启动时初始化调度器
@app.on_event("startup")
async def init_scheduler():
    # 添加定时规则：每分钟清理临时数据，每30分钟同步外部数据
    scheduler.add_job(cleanup_temp_data, "interval", minutes=1)
    scheduler.add_job(sync_external_data, "interval", minutes=30)
    scheduler.start()

# 关闭服务时停止调度器
@app.on_event("shutdown")
async def stop_scheduler():
    scheduler.shutdown()

# 示例API接口
@app.get("/health")
async def health_check():
    return {"status": "running"}

3. 注意事项

• 若任务是CPU密集型，建议改用ThreadPoolScheduler或ProcessPoolScheduler，避免阻塞FastAPI的异步事件循环；
• FastAPI重启会导致调度任务中断，适合对任务连续性要求不高的场景；
• 可以通过APScheduler的存储后端（比如Redis）实现任务持久化，避免重启后丢失任务状态。

二、集成方案不可行时的替代方案 #

如果你的任务存在以下情况，集成方案就不再适用：

• 任务执行时间长、资源占用高，影响API响应性能；
• 任务需要高可用性，API重启不能中断任务；
• 任务逻辑复杂，需要独立的监控、日志和扩缩容能力。

此时的最佳实现方式有两种：

1. 使用Celery+消息队列：FastAPI作为任务生产者，将定时任务发送到Redis/RabbitMQ，Celery Worker独立部署执行任务。这种方式完全解耦API和任务服务，支持分布式执行，适合复杂任务场景。
2. 系统级定时任务：如果任务逻辑简单，直接用Linux的cron或Windows的任务计划程序，调用独立的Python脚本执行任务，无需依赖FastAPI服务。

三、拆分服务是否为正确选择？ #

是否拆分取决于你的业务场景：

• 无需拆分：任务轻量、与API业务强关联（比如定时清理API生成的临时文件）、对任务连续性要求低，此时集成在FastAPI中更简洁，维护成本更低。
• 应该拆分：任务资源消耗大、执行周期长、需要独立扩缩容或高可用性，拆分后可以实现故障隔离，各自优化资源配置，便于独立监控和维护。

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