Flask与Celery结合实现每月定时任务
在Web开发中，定时任务是一个常见的需求，特别是在需要定期执行数据清洗、报表生成或系统维护等操作时。Flask作为一个轻量级的Web框架，通常用于快速开发，但它本身不直接支持定时任务。幸运的是，我们可以利用Celery这个强大的异步任务队列/作业调度框架来实现这一功能。
环境准备
首先，确保你的Python环境中已经安装了Flask和Celery。如果没有，可以通过pip安装：
bash复制代码 pip install Flask Celery redis
注意：Celery需要一个消息代理来传递消息，这里我们使用Redis作为示例。如果你没有Redis，请先安装并启动Redis服务器。
配置Flask和Celery
创建Flask应用
首先，我们创建一个简单的Flask应用。
python复制代码 # app.py from flask import Flask app = Flask(__name__) @app.route('/') def hello(): return 'Hello, Flask with Celery!' if __name__ == '__main__': app.run(debug=True)
配置Celery
然后，配置Celery以连接到Redis，并指定我们的Flask应用作为Celery的上下文。
python复制代码 # celery_app.py from celery import Celery from app import app # 配置Celery使用Redis作为消息代理 celery = Celery( 'tasks', broker='redis://localhost:6379/0', backend='redis://localhost:6379/0' ) # 将Flask应用上下文绑定到Celery celery.config_from_object(app.config) # 导入任务模块，这样Celery才能自动发现这些任务 # 这里我们稍后定义任务
定义任务
接下来，在另一个文件中定义Celery任务。
python复制代码 # tasks.py from celery_app import celery from datetime import datetime @celery.task def monthly_task(): print(f"Monthly task executed at {datetime.now()}") # 这里可以添加你的任务逻辑 return "Monthly task completed"
设置定时任务
使用Celery的定时任务功能（也称为beat），我们可以安排monthly_task每个月执行一次。这通常是通过配置Celery的beat scheduler来实现的。
python复制代码 # celery_beat_schedule.py from celery.schedules import crontab CELERY_BEAT_SCHEDULE = { 'monthly-task': { 'task': 'tasks.monthly_task', 'schedule': crontab(day_of_month=1, hour=0, minute=0), }, }
注意：这个配置文件需要被Celery加载。这通常是通过在Celery的配置中导入这个模块来实现的，但具体方法可能因你的项目结构而异。
启动Celery Worker和Beat
最后，你需要分别启动Celery的worker和beat。worker负责执行任务，而beat负责按照计划调度任务。
bash复制代码 # 启动Celery Worker celery -A celery_app worker --loglevel=info # 在另一个终端启动Celery Beat celery -A celery_app beat --loglevel=info
确保你的Redis服务器正在运行，并且Celery的worker和beat都配置正确，指向了同一个Redis实例。
总结
通过以上步骤，我们成功地在Flask应用中集成了Celery，并设置了一个每月执行的定时任务。Celery提供了灵活的任务调度和异步任务处理能力，非常适合用于处理Web应用中的后台任务。你可以根据需要扩展任务列表，并调整任务的执行计划。
