### Flask 中使用 Celery Beat 的配置与任务调度详解 在 Flask 项目中集成 Celery 和 Celery Beat，可以实现定时任务的调度功能。以下是详细的配置和任务调度说明： #### 1. 安装依赖 首先需要安装 Celery 和 Redis（作为消息队列和结果存储）。可以通过以下命令安装： ```bash pip install celery redis flask ``` #### 2. 配置 Celery 应用 创建一个 `celery_app.py` 文件来定义 Celery 应用，并设置 Redis 为 Broker 和 Backend： ```python from celery import Celery def make_celery(app_name): # 创建 Celery 实例 celery = Celery( app_name, broker='redis://localhost:6379/0', # 使用 Redis 作为 Broker backend='redis://localhost:6379/1' # 使用 Redis 作为 Backend ) return celery # 初始化 Celery celery_app = make_celery('flask_celery') ``` #### 3. 配置 Celery Beat Celery Beat 是 Celery 的定时任务调度器。需要在 Celery 配置中添加定时任务的调度信息。 在 `celery_app.py` 中添加以下内容： ```python from celery.schedules import crontab # 配置定时任务 celery_app.conf.beat_schedule = { 'add-every-30-seconds': { # 定时任务名称 'task': 'tasks.add', # 调用的任务函数 'schedule': 30.0, # 每 30 秒执行一次 'args': (16, 16) # 传递给任务的参数 }, 'send-email-every-monday-morning': { # 另一个定时任务 'task': 'tasks.send_email', 'schedule': crontab(hour=7, minute=30, day_of_week=1), # 每周一早上 7:30 执行 }, } ``` #### 4. 定义任务 在 `tasks.py` 文件中定义具体的任务函数。例如： ```python from celery_app import celery_app @celery_app.task def add(x, y): return x + y @celery_app.task def send_email(): print("发送邮件任务正在执行...") # 这里可以添加发送邮件的逻辑 ``` #### 5. 启动 Celery 和 Celery Beat 启动 Celery Worker 和 Celery Beat 以运行任务和调度器。 - 启动 Celery Worker： ```bash celery -A celery_app worker --loglevel=info ``` - 启动 Celery Beat： ```bash celery -A celery_app beat --loglevel=info ``` #### 6. 集成到 Flask 项目 将 Celery 集成到 Flask 项目中，确保两者共享相同的配置。可以在 Flask 应用初始化时加载 Celery 配置： ```python from flask import Flask from celery_app import celery_app app = Flask(__name__) # 将 Flask 配置传递给 Celery app.config.update( CELERY_BROKER_URL='redis://localhost:6379/0', CELERY_RESULT_BACKEND='redis://localhost:6379/1' ) # 更新 Celery 配置 celery_app.conf.update(app.config) ``` #### 7. 示例：动态添加任务 如果需要动态添加任务，可以参考以下代码： ```python from celery_app import celery_app @celery_app.task def dynamic_task(message): print(f"动态任务正在执行...消息: {message}") ``` 通过 Celery Beat 动态添加任务时，可以修改 `beat_schedule` 并重新启动 Celery Beat[^2]。 --- ### 注意事项 - 确保 Redis 服务已启动并正确配置。 - 如果需要持久化任务调度信息，可以将 Celery Beat 的调度表存储在数据库中[^4]。 - 在生产环境中，建议使用更安全的 Broker 和 Backend 配置。 ---