当前位置：首页 > news >正文

如何使用flask做任务调度

news 2025/9/17 2:41:03

如何使用flask做任务调度

若要运用 Flask 开展任务调度工作，一般可采用以下几种办法：

1. 使用 APScheduler 库

APScheduler 是一个功能全面的 Python 任务调度库，它能和 Flask 完美集成，支持固定时间间隔执行、cron 表达式执行等多种调度方式。

下面是一个集成 APScheduler 的 Flask 应用示例：

from flask import Flask
from apscheduler.schedulers.background import BackgroundSchedulerapp = Flask(__name__)
scheduler = BackgroundScheduler()# 定义要调度的任务
def scheduled_task():print("执行定时任务")# 这里可以放置你的业务逻辑# 添加任务到调度器
scheduler.add_job(func=scheduled_task, trigger='interval', seconds=60)# 启动调度器
scheduler.start()@app.route('/')
def index():return "Flask 任务调度示例"if __name__ == '__main__':app.run(debug=True, use_reloader=False)  # 注意：必须设置 use_reloader=False 以避免任务重复执行

这个示例创建了一个每隔 60 秒执行一次的定时任务。你可以通过修改 trigger 参数和相关设置来更改任务的执行频率和方式。

2. 使用 Celery

Celery 是一个强大的分布式任务队列系统，非常适合处理后台任务。结合 Flask 使用时，你可以将耗时的操作放到 Celery 任务中异步执行。

下面是一个使用 Flask 和 Celery 的示例：

from flask import Flask
from celery import Celeryapp = Flask(__name__)
app.config['CELERY_BROKER_URL'] = 'redis://localhost:6379/0'
app.config['CELERY_RESULT_BACKEND'] = 'redis://localhost:6379/0'celery = Celery(app.name, broker=app.config['CELERY_BROKER_URL'])
celery.conf.update(app.config)# 定义 Celery 任务
@celery.task
def long_running_task():print("开始执行长时间运行的任务")# 模拟耗时操作import timetime.sleep(10)print("任务执行完成")return "任务结果"@app.route('/trigger_task')
def trigger_task():# 异步执行任务task = long_running_task.delay()return f"任务已触发，任务ID: {task.id}"if __name__ == '__main__':app.run(debug=True)

要让这个示例正常工作，你需要安装 Redis 服务器并启动 Celery worker：

celery -A your_app_name.celery worker --loglevel=info

3. 使用 Flask-APScheduler 扩展

Flask-APScheduler 是专门为 Flask 设计的 APScheduler 扩展，使用起来更为简便。

下面是使用 Flask-APScheduler 的示例：

from flask import Flask
from flask_apscheduler import APSchedulerapp = Flask(__name__)
scheduler = APScheduler()# 配置调度器
class Config:SCHEDULER_API_ENABLED = Trueapp.config.from_object(Config())# 定义任务
def my_task():print("执行 Flask-APScheduler 任务")# 添加任务
scheduler.add_job(id='my_task', func=my_task, trigger='interval', seconds=30)# 初始化调度器
scheduler.init_app(app)
scheduler.start()@app.route('/')
def index():return "Flask-APScheduler 示例"if __name__ == '__main__':app.run(debug=True)

这个扩展还提供了 REST API 来管理任务，你可以通过访问 /scheduler 端点来查看和操作任务。

4. 使用 BackgroundScheduler 与 Flask 结合

如果你不想使用扩展，可以直接将 BackgroundScheduler 与 Flask 结合使用，这种方式更加灵活。

下面是一个更灵活的集成示例：

from flask import Flask
from apscheduler.schedulers.background import BackgroundScheduler
import atexitapp = Flask(__name__)
scheduler = BackgroundScheduler()def task_function():print("执行后台任务")# 执行你的任务逻辑# 添加任务
scheduler.add_job(func=task_function,trigger='cron',  # 使用 cron 表达式hour=2,          # 每天凌晨2点执行minute=0
)# 启动调度器
scheduler.start()# 确保应用退出时关闭调度器
atexit.register(lambda: scheduler.shutdown())@app.route('/')
def home():return "Flask 与 BackgroundScheduler 集成示例"if __name__ == '__main__':app.run(debug=True, use_reloader=False)  # 禁用重新加载器防止任务重复