在Heroku运行Flask应用时，能否用APScheduler动态添加任务？

完全可以实现运行时调用scheduler.add_job()添加任务，你的MongoDBJobStore思路是正确的核心方向——通过共享的持久化JobStore，让Flask应用和clock进程的调度器同步任务状态。下面是具体的实现步骤：

1. 核心原理说明 #

Heroku上的web dyno（跑Flask）和clock dyno（跑APScheduler）是独立进程，不能直接共享内存中的调度器实例。但通过MongoDBJobStore，所有任务都会被持久化到MongoDB中：

• Flask进程添加任务时，直接将任务写入MongoDB
• clock进程的调度器会定期从MongoDB刷新任务列表，自动执行新添加的任务

2. 配置MongoDBJobStore #

第一步：安装依赖 #

确保两个进程都安装了所需包：

pip install apscheduler flask pymongo gunicorn

第二步：编写clock.py #

这是clock dyno的启动文件，初始化调度器并绑定MongoDBJobStore：

from apscheduler.schedulers.blocking import BlockingScheduler
from apscheduler.jobstores.mongodb import MongoDBJobStore
import os
from your_task_module import run_scheduled_task  # 导入你的任务执行函数

# 从Heroku环境变量读取MongoDB连接信息
MONGODB_URI = os.environ["MONGODB_URI"]

# 配置JobStore，指定数据库和集合
jobstores = {
    "default": MongoDBJobStore(
        uri=MONGODB_URI,
        database="task_scheduler",
        collection="jobs"
    )
}

# 初始化阻塞式调度器（适合clock进程）
scheduler = BlockingScheduler(jobstores=jobstores)

if __name__ == "__main__":
    try:
        scheduler.start()
    except (KeyboardInterrupt, SystemExit):
        pass

第三步：Flask应用中配置相同的JobStore #

在你的Flask app文件（比如app.py）中，初始化一个后台调度器，使用完全相同的MongoDBJobStore配置：

from flask import Flask, request, jsonify
from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.jobstores.mongodb import MongoDBJobStore
import os
from your_task_module import run_scheduled_task

app = Flask(__name__)

MONGODB_URI = os.environ["MONGODB_URI"]

# 和clock.py一致的JobStore配置
jobstores = {
    "default": MongoDBJobStore(
        uri=MONGODB_URI,
        database="task_scheduler",
        collection="jobs"
    )
}

# 初始化后台调度器（不阻塞Flask进程）
scheduler = BackgroundScheduler(jobstores=jobstores)
scheduler.start()

# 实现添加任务的API接口
@app.route("/api/add-task", methods=["POST"])
def add_scheduled_task():
    req_data = request.get_json()
    
    # 从请求中获取任务参数（根据你的需求调整）
    task_id = req_data.get("task_id")
    target_hour = req_data.get("hour")
    target_minute = req_data.get("minute")
    task_params = req_data.get("params", [])

    # 动态添加任务到JobStore（自动同步到MongoDB）
    scheduler.add_job(
        func=run_scheduled_task,
        trigger="cron",
        hour=target_hour,
        minute=target_minute,
        args=task_params,
        id=task_id,
        replace_existing=True  # 重复ID时覆盖原有任务
    )

    return jsonify({"status": "success", "message": "任务已成功调度"})

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

3. 任务函数的注意事项 #

• 任务函数（比如run_scheduled_task）必须放在独立的模块中（比如your_task_module.py），确保Flask和clock进程都能导入执行
• 函数不能依赖Flask请求上下文或web进程的独有资源，因为clock进程是独立运行的

4. Heroku部署配置 #

Procfile配置 #

在项目根目录创建Procfile，定义两个dyno的启动命令：

web: gunicorn app:app
clock: python clock.py

环境变量配置 #

在Heroku的配置变量中添加：

• MONGODB_URI：你的MongoDB连接字符串（比如Mongo Atlas的URI）
• 其他你需要的环境变量

启动dyno #

部署后，同时启动web和clock dyno：

heroku ps:scale web=1 clock=1

5. 额外优化点 #

• 可以添加任务删除、查询的API接口，调用scheduler.remove_job(task_id)或scheduler.get_job(task_id)即可
• 调整JobStore的refresh_interval参数（默认60秒），控制clock进程刷新任务列表的频率
• 给任务添加错误处理，比如配置APScheduler的executors和job_defaults设置重试机制

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