### 在 FastAPI 中实现定时任务与批处理业务逻辑 FastAPI 是一个基于 Python 的现代 Web 框架，支持异步请求处理和类型提示，适用于构建高性能的 API 服务。在实际业务中，经常需要实现定时任务和批处理逻辑，例如定时同步数据、生成报表、清理缓存等操作。以下是如何在 FastAPI 中实现定时任务与批处理业务逻辑的详细说明。 #### 使用 `APScheduler` 实现定时任务 FastAPI 可以结合 `APScheduler`（Advanced Python Scheduler）来实现定时任务调度。`APScheduler` 提供了三种调度器：`BlockingScheduler`、`BackgroundScheduler` 等，适用于不同的运行环境。以下是一个使用 `BackgroundScheduler` 在 FastAPI 中定时执行任务的示例： ```python from fastapi import FastAPI from apscheduler.schedulers.background import BackgroundScheduler import time app = FastAPI() # 定义一个简单的定时任务函数 def batch_task(): print(f"执行定时任务: {time.strftime('%Y-%m-%d %H:%M:%S')}") # 初始化调度器 scheduler = BackgroundScheduler() # 添加定时任务（每 10 秒执行一次） scheduler.add_job(batch_task, 'interval', seconds=10) # 启动调度器 @app.on_event("startup") def startup_event(): scheduler.start() # 停止调度器 @app.on_event("shutdown") def shutdown_event(): scheduler.shutdown() @app.get("/") def read_root(): return {"message": "定时任务已启动"} ``` 此实现方式允许开发者将定时任务逻辑与业务逻辑分离，提高了系统的可维护性和可扩展性[^2]。 #### 实现批处理业务逻辑 在 FastAPI 中，批处理任务通常可以封装为一个函数，并通过定时任务触发执行。例如，可以实现一个数据清洗或数据聚合的函数，定期从数据库中提取数据进行处理： ```python from fastapi import FastAPI from apscheduler.schedulers.background import BackgroundScheduler import time import random app = FastAPI() # 模拟数据库数据 data_store = [random.randint(1, 100) for _ in range(1000)] # 批处理逻辑：计算数据总和 def process_batch_data(): total = sum(data_store) print(f"批处理结果: 数据总和为 {total}") # 定义定时任务 def scheduled_task(): print(f"开始执行批处理: {time.strftime('%Y-%m-%d %H:%M:%S')}") process_batch_data() # 初始化调度器 scheduler = BackgroundScheduler() # 添加定时任务（每天 00:00 执行） scheduler.add_job(scheduled_task, 'cron', hour=0, minute=0) @app.on_event("startup") def start_scheduler(): scheduler.start() @app.on_event("shutdown") def stop_scheduler(): scheduler.shutdown() @app.get("/trigger-batch") def trigger_batch(): process_batch_data() return {"message": "批处理已手动触发"} ``` 该实现方式将业务逻辑封装在 `process_batch_data()` 函数中，既可以通过定时任务自动触发，也可以通过 API 接口手动调用，增强了系统的灵活性和可测试性[^3]。 #### 结合异步任务队列实现高性能批处理 对于计算密集型或耗时较长的批处理任务，建议使用异步任务队列（如 Celery）与 FastAPI 集成。Celery 支持任务异步执行、任务重试、任务队列持久化等功能，适用于大规模数据处理场景。以下是一个使用 Celery 的示例： ```python # tasks.py from celery import Celery import time app = Celery('tasks', broker='redis://localhost:6379/0') @app.task def long_running_task(): time.sleep(10) # 模拟长时间任务 print("批处理任务已完成") return "任务完成" ``` ```python # main.py from fastapi import FastAPI from tasks import long_running_task app = FastAPI() @app.get("/start-task") def start_task(): task = long_running_task.delay() return {"task_id": task.id} ``` 通过 Celery，可以将批处理任务从主线程中解耦，避免阻塞 Web 服务，提高整体性能和响应速度。