APScheduler

news2024/10/23 0:13:50

APScheduler（Advanced Python Scheduler）是一个轻量级的任务调度库，适用于在 Flask 中执行定时任务和后台任务。使用 APScheduler 可以很方便地在 Flask 应用中实现定时任务或周期性任务。

使用 APScheduler 实现后台任务调度

1. 安装 APScheduler

首先安装 APScheduler：

pip install apscheduler

2. Flask 中集成 APScheduler

你可以在 Flask 应用中使用 APScheduler 的 BackgroundScheduler 来调度任务：

示例：在 Flask 中使用 APScheduler

from flask import Flask, jsonify
from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.triggers.interval import IntervalTrigger
import atexit
import time

app = Flask(__name__)

# 创建 BackgroundScheduler 实例
scheduler = BackgroundScheduler()

# 定义一个周期性任务
def scheduled_task():
    print("Scheduled task is running!")
    # 模拟一些处理
    time.sleep(1)

# 添加一个任务，每 10 秒执行一次
scheduler.add_job(func=scheduled_task, trigger=IntervalTrigger(seconds=10))

# 启动调度器
scheduler.start()

# 确保程序退出时调度器关闭
atexit.register(lambda: scheduler.shutdown())

@app.route('/')
def index():
    return jsonify({"message": "APScheduler is running!"})

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=8000)

关键步骤和注意事项

创建调度器：在 Flask 应用初始化时创建 BackgroundScheduler 实例。
添加任务：使用 scheduler.add_job() 方法将任务添加到调度器中。可以使用多种触发器（如 IntervalTrigger、CronTrigger 等）来定义任务的触发条件。
启动调度器：在应用启动时调用 scheduler.start() 来启动调度器。
关闭调度器：使用 atexit.register() 在应用关闭时确保调度器正常停止。

示例：更复杂的调度方式

1. 使用 Cron 表达式

你可以使用 CronTrigger 来定义更精确的调度时间，例如每天的凌晨 2 点：

from apscheduler.triggers.cron import CronTrigger

scheduler.add_job(
    func=scheduled_task,
    trigger=CronTrigger(hour=2, minute=0),
    id='cron_task',
    replace_existing=True
)

2. 添加一次性任务

可以添加一个一次性任务（只执行一次）：

from apscheduler.triggers.date import DateTrigger
from datetime import datetime, timedelta

run_time = datetime.now() + timedelta(seconds=30)
scheduler.add_job(
    func=scheduled_task,
    trigger=DateTrigger(run_date=run_time),
    id='one_time_task',
    replace_existing=True
)

APScheduler 常见配置

在生产环境中，你可能需要对 APScheduler 进行配置，例如使用数据库作为持久化存储或调整线程池大小。配置方式如下：

app.config.update({
    'SCHEDULER_JOBSTORES': {
        'default': {'type': 'sqlalchemy', 'url': 'sqlite:///jobs.sqlite'}
    },
    'SCHEDULER_EXECUTORS': {
        'default': {'type': 'threadpool', 'max_workers': 20}
    },
    'SCHEDULER_JOB_DEFAULTS': {
        'coalesce': False,
        'max_instances': 3
    }
})