Celery任务在Flask应用启动前自动执行，如何设置为API触发？

问题原因 #

启动Celery Worker时任务就自动执行，通常是以下两种情况导致：

• Redis队列中存在之前遗留的未处理任务，Worker启动后立即消费这些任务
• 导入的模块（比如report_main.py）中存在直接执行的代码，模块加载时就触发了任务逻辑

解决方案 #

1. 清理Redis队列 #

执行命令清空Redis中的所有任务队列，避免Worker启动后处理旧任务：

redis-cli FLUSHDB

2. 检查模块级代码执行 #

打开report_main.py，确保generate_report函数没有在模块级别被直接调用。比如不要出现类似下面的代码（除非放在if __name__ == "__main__"块内）：

# 错误示例：模块级别直接调用函数
generate_report(123, 456, 789)

3. 优化代码（移除不安全的全局变量） #

你代码中使用的全局变量在多进程/多线程环境下容易出问题，建议直接在路由函数中获取参数后传给任务：

from flask import Flask, jsonify, request
from celery import Celery
from report_main import generate_report

app = Flask(__name__)
app.config['CELERY_BROKER_URL'] = 'redis://localhost:6379'
# 可选：配置结果后端，用于查询任务状态和结果
app.config['CELERY_RESULT_BACKEND'] = 'redis://localhost:6379'

celery = Celery(app.name, 
                broker=app.config['CELERY_BROKER_URL'],
                backend=app.config['CELERY_RESULT_BACKEND'])

@app.route("/")
def home():
    return "Report generation"

@app.route("/home")
def home_home():
    return jsonify(listening="The model is live")

@app.route("/report", methods=["POST", "GET"])
def upload():
    machine_id = request.args.get("machine_id")
    scan_id = request.args.get("scan_id")
    org_id = request.args.get("org_id")

    # 校验必填参数
    if not all([machine_id, scan_id, org_id]):
        return jsonify(error="缺少必填参数：machine_id、scan_id、org_id"), 400

    task = intiate_report_generation.delay(org_id, machine_id, scan_id)

    return jsonify(success="报告生成已启动", task_id=task.id)

@celery.task
def intiate_report_generation(org_id, machine_id, scan_id):
    report_url, heatmap_urls = generate_report(org_id, machine_id, scan_id)
    # 可选：返回任务结果，方便后续查询
    return {"report_url": report_url, "heatmap_urls": heatmap_urls}

if __name__ == "__main__":
    app.run()

4. 正确的启动步骤 #

1. 确保Redis服务正在运行：redis-server
2. 启动Celery Worker（注意指定正确的Celery实例路径）：

celery -A run.celery worker -l info

3. 启动Flask应用：

python run.py

完成以上配置后，只有当调用/report端点并传入完整参数时，Celery任务才会被触发执行。

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