Flask中通过API请求调用Celery add_periodic_task触发celery.beat方法问询

默认的celery-beat本身不支持直接通过API调用动态添加周期任务，因为默认调度器的任务配置是加载自本地静态配置、存储在进程内存中的，修改后重启就会丢失，多实例部署时也会出现配置不同步的问题。你需要配合支持动态存储的调度器来完成需求，具体实现逻辑如下：

核心实现步骤 #

• 替换celery默认的PersistentScheduler，改用支持公共存储的动态调度器，可选方案包括基于MySQL的DatabaseScheduler、基于Redis的celery-beatx调度器，所有周期任务规则统一存在公共存储中，beat进程会定期轮询存储自动同步最新的任务列表
• 在Flask侧开发API接口，接收用户传递的时间、日期参数，校验合法性后直接在公共存储中新增/更新对应周期任务的配置即可，不需要主动调用beat进程
• 如果你需要让beat立即重载任务，可以给beat进程发送SIGHUP信号触发配置重载，默认轮询间隔是5分钟，你也可以通过beat_max_loop_interval配置调整轮询频率

示例代码参考 #

首先安装Redis版本的动态调度器依赖：

pip install celery celery-beatx flask

Celery实例配置修改：

from celery import Celery
from celery.schedules import crontab

app = Celery('your_project_name')
# 配置动态调度器和公共存储地址
app.conf.beat_scheduler = 'beatx.scheduler.RedisScheduler'
app.conf.beatx_store = 'redis://localhost:6379/0'
app.conf.broker_url = 'redis://localhost:6379/0'

Flask API接口示例：

from flask import Flask, request, jsonify
from beatx import store

flask_app = Flask(__name__)

@flask_app.route('/add-periodic-task', methods=['POST'])
def add_periodic_task():
    req_data = request.get_json()
    # 校验用户传入的时间、日期参数合法性
    if not all([req_data.get('user_id'), req_data.get('minute'), req_data.get('hour')]):
        return jsonify({"status": "error", "msg": "参数缺失"}), 400
    
    # 生成唯一的任务名称
    task_name = f"user_{req_data['user_id']}_custom_task"
    # 构造周期任务规则，也可以直接使用用户传入的cron表达式
    task_config = {
        "task": "your_custom_celery_task_func", # 替换为你实际的Celery任务函数名
        "schedule": crontab(minute=req_data['minute'], hour=req_data['hour'], day_of_month=req_data.get('day', '*')),
        "args": req_data.get('task_args', []),
        "kwargs": req_data.get('task_kwargs', {})
    }
    # 写入公共存储，beat会自动同步该任务
    store.set(task_name, task_config)
    return jsonify({"status": "success", "task_name": task_name})

注意事项 #

• 不要直接修改当前进程内存中的app.conf.beat_schedule配置，该修改仅对当前进程生效，beat进程无法感知
• 生产环境禁止使用默认的文件存储调度器，多节点部署时会出现任务重复执行、配置不同步的问题
• 需要严格校验用户传入的时间参数合法性，避免非法的调度规则导致beat进程崩溃

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