在 FastAPI 项目中，通过 `main.py` 文件管理定时任务文件中的任务是一种常见的做法，尤其是在需要集中控制任务调度逻辑的场景下。通常可以通过将定时任务模块（如使用 `APScheduler` 或 `Celery`）作为 FastAPI 应用的一部分进行初始化，并在应用启动时加载这些任务。 以下是一个基于 `APScheduler` 的实现方式，展示了如何在 `main.py` 中加载并管理外部定时任务模块： --- ### 使用 `APScheduler` 管理外部定时任务模块 #### 示例目录结构： ``` my_fastapi_project/ │ ├── main.py ├── tasks/ │ ├── __init__.py │ └── cron_tasks.py ``` #### `tasks/cron_tasks.py` 内容如下： ```python from datetime import datetime def log_current_time(): print(f"当前时间: {datetime.now()}") ``` #### `main.py` 中集成并调用该任务： ```python from fastapi import FastAPI from apscheduler.schedulers.asyncio import AsyncIOScheduler from tasks.cron_tasks import log_current_time app = FastAPI() scheduler = AsyncIOScheduler() @app.on_event("startup") async def startup_event(): scheduler.add_job(log_current_time, 'interval', seconds=10) scheduler.start() @app.on_event("shutdown") async def shutdown_event(): scheduler.shutdown() ``` 上述代码中，`main.py` 负责初始化调度器并从 `cron_tasks.py` 导入任务函数，随后将其注册到调度器中[^2]。 --- ### 使用 `Celery` 管理异步任务模块 若使用 `Celery` 实现定时任务，则可以通过 `celery beat` 调度机制统一管理多个任务模块。 #### 示例：`main.py` 启动时触发 Celery 模块化任务 假设 `tasks/long_task.py` 定义了耗时任务： ```python # tasks/long_task.py from celery import Celery celery_app = Celery('tasks', broker='redis://localhost:6379/0') @celery_app.task def long_running_task(name): print(f"开始执行任务: {name}") return f"任务 {name} 完成" ``` 在 `main.py` 中可以调用该任务： ```python from fastapi import FastAPI from tasks.long_task import long_running_task app = FastAPI() @app.get("/run-task") async def run_task(): task = long_running_task.delay("test_task") return {"task_id": task.id} ``` 这种方式允许 `main.py` 作为任务入口点，而具体的任务定义分布在多个模块中，便于维护和扩展[^1]。 --- ### 动态加载与热更新支持 如果希望在运行时动态加载或卸载任务，可以在 `main.py` 中引入反射机制或配置中心，根据配置信息动态导入任务模块并注册到调度器中。例如： ```python import importlib TASK_MODULES = ["tasks.cron_tasks", "tasks.other_tasks"] @app.on_event("startup") async def load_scheduled_tasks(): for module_path in TASK_MODULES: module = importlib.import_module(module_path) if hasattr(module, "register_tasks"): module.register_tasks(scheduler) ``` 这样，每个任务模块可以自行定义 `register_tasks(scheduler)` 函数用于注册其内部的任务，实现松耦合的设计[^4]。 ---