FastAPI 教程：详解 FastAPI 中设置定时任务

引言

在Web开发中，定时任务是常见的需求场景，如数据同步、日志清理、通知推送等。FastAPI作为高性能的异步Web框架，虽然本身不提供内置的定时任务功能，但通过结合第三方库或自定义方案，可以轻松实现这一需求。本文将系统介绍FastAPI中设置定时任务的多种方法，涵盖从基础到进阶的完整实现路径。

一、APScheduler方案：轻量级定时任务实现

APScheduler是Python中最流行的定时任务库之一，支持多种触发器类型（间隔、日期、Cron表达式），与FastAPI的集成简单高效。

1.1 基础集成

from fastapi import FastAPI
from apscheduler.schedulers.background import BackgroundScheduler
app = FastAPI()
scheduler = BackgroundScheduler()
scheduler.start()
def job():
    print("定时任务执行")
# 添加间隔任务（每5秒执行一次）
scheduler.add_job(job, 'interval', seconds=5)

1.2 高级配置

• 触发器类型：支持date（单次）、interval（间隔）、cron（类Unix Cron表达式）
• 任务持久化：可通过SQLAlchemyJobStore实现任务持久化
• 异常处理：建议添加misfire_grace_time和coalesce参数处理异常情况

1.3 生产环境建议

• 使用BackgroundScheduler而非BlockingScheduler
• 添加任务重试机制
• 考虑使用asyncio适配器实现异步任务

二、Celery方案：分布式任务队列集成

对于需要分布式处理或复杂任务链的场景，Celery是更专业的选择。

2.1 基础架构

# celery_app.py
from celery import Celery
celery = Celery(
    'tasks',
    broker='redis://localhost:6379/0',
    backend='redis://localhost:6379/1'
)
@celery.task
def scheduled_task():
    print("Celery定时任务执行")

2.2 FastAPI集成

from fastapi import FastAPI
from celery.schedules import crontab
from celery_app import celery
app = FastAPI()
# 配置Celery Beat（定时任务调度器）
celery.conf.beat_schedule = {
    'every-5-seconds': {
        'task': 'celery_app.scheduled_task',
        'schedule': 5.0  # 或使用crontab(minute='*/5')
    }
}
@app.on_event("startup")
async def startup_event():
    worker = celery.Worker()
    worker.start()

2.3 优势对比

特性	APScheduler	Celery
部署复杂度	低	高
分布式支持	否	是
任务持久化	有限	完善
扩展性	一般	强

三、自定义方案：基于ASGI中间件

对于需要深度集成的场景，可以实现自定义ASGI中间件。

3.1 实现原理

import asyncio
from fastapi import FastAPI, Request
from datetime import datetime
class TimerMiddleware:
    def __init__(self, app: FastAPI):
        self.app = app
        self.tasks = []
    async def __call__(self, scope, receive, send):
        if scope["type"] == "lifespan":
            await self.manage_lifespan(scope, receive, send)
        else:
            await self.app(scope, receive, send)
    async def manage_lifespan(self, scope, receive, send):
        async with receive() as message:
            if message["type"] == "lifespan.startup":
                # 启动定时任务
                asyncio.create_task(self.run_periodic_task())
                await send({"type": "lifespan.startup.complete"})
            elif message["type"] == "lifespan.shutdown":
                # 清理资源
                await send({"type": "lifespan.shutdown.complete"})
    async def run_periodic_task(self):
        while True:
            print(f"自定义定时任务执行于 {datetime.now()}")
            await asyncio.sleep(10)  # 每10秒执行一次
app = FastAPI()
app.add_middleware(TimerMiddleware)

3.2 适用场景

• 需要与请求生命周期紧密结合
• 需要自定义任务调度逻辑
• 避免引入额外依赖

四、最佳实践与注意事项

4.1 资源管理

• 确保定时任务不会阻塞主线程
• 合理设置任务间隔，避免资源耗尽
• 生产环境建议使用独立的worker进程

4.2 日志与监控

import logging
from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore
logging.basicConfig()
logging.getLogger('apscheduler').setLevel(logging.DEBUG)
# 配置持久化存储
jobstores = {
    'default': SQLAlchemyJobStore(url='sqlite:///jobs.db')
}
scheduler = BackgroundScheduler(jobstores=jobstores)

4.3 测试策略

• 使用freezegun库模拟时间
• 编写单元测试验证任务执行
• 集成测试验证分布式场景

五、进阶方案：结合消息队列

对于高并发场景，推荐使用消息队列解耦任务生产与消费。

5.1 Redis Pub/Sub实现

import redis
import asyncio
from fastapi import FastAPI
r = redis.Redis()
app = FastAPI()
async def task_consumer():
    pubsub = r.pubsub()
    pubsub.subscribe('scheduled_tasks')
    for message in pubsub.listen():
        if message['type'] == 'message':
            print(f"收到任务: {message['data']}")
@app.on_event("startup")
async def startup():
    asyncio.create_task(task_consumer())
@app.post("/schedule")
async def schedule_task():
    r.publish('scheduled_tasks', 'execute_now')
    return {"status": "task scheduled"}

5.2 Kafka集成方案

• 使用confluent-kafka库
• 实现消费者组保证任务只被处理一次
• 考虑使用Dead Letter Queue处理失败任务

六、性能优化建议

1. 任务分片：对于耗时任务，考虑拆分为多个子任务
2. 并发控制：使用semaphore限制同时执行的任务数
3. 缓存结果：对频繁执行且结果稳定的任务添加缓存
4. 异步IO：确保任务函数使用async/await模式

七、常见问题解决方案

7.1 任务重复执行

• 检查是否多个实例同时运行
• 使用分布式锁（如Redis锁）
• 配置coalesce=True合并积压任务

7.2 时区问题

from pytz import timezone
scheduler.add_job(
    job,
    'cron',
    hour=8,
    minute=30,
    timezone=timezone('Asia/Shanghai')
)

7.3 任务持久化失败

• 检查数据库连接配置
• 确保有足够的磁盘空间
• 配置定期备份策略

结论

FastAPI中的定时任务实现有多种选择，开发者应根据具体场景选择合适方案：

• 简单场景：APScheduler
• 分布式需求：Celery
• 深度集成：自定义中间件
• 高并发场景：消息队列方案

通过合理选择和组合这些方案，可以构建出稳定、高效的定时任务系统。建议从APScheduler开始，随着业务发展逐步引入更复杂的架构。