Python Flask实现定时任务的不同方法详解

from flask import Flask
from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore
from datetime import datetime

app = Flask(__name__)
app.config['SCHEDULER_JOBSTORES'] = {
    'default': SQLAlchemyJobStore(url='sqlite:///jobs.db')
}

# 创建调度器实例
scheduler = BackgroundScheduler(
    jobstores=app.config['SCHEDULER_JOBSTORES'],
    daemon=True
)

# 定义定时任务函数
def daily_report():
    """每天上午9点执行的日报任务"""
    current_time = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
    print(f"[日报] 执行时间: {current_time}")
    # 这里添加实际的日报生成逻辑

def every_5_minutes():
    """每5分钟执行一次的任务"""
    current_time = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
    print(f"[5分钟任务] 执行时间: {current_time}")
    # 这里添加实际的任务逻辑

def on_startup():
    """应用启动时添加定时任务"""
    # 添加每天9:00执行的任务
    scheduler.add_job(
        id='daily_report_job',
        func=daily_report,
        trigger='cron',
        hour=9,
        minute=0,
        replace_existing=True
    )
    
    # 添加每5分钟执行的任务
    scheduler.add_job(
        id='every_5_minutes_job',
        func=every_5_minutes,
        trigger='interval',
        minutes=5,
        replace_existing=True
    )
    
    # 添加一次性任务（应用启动后10秒执行）
    scheduler.add_job(
        id='startup_check',
        func=lambda: print("应用启动完成!"),
        trigger='date',
        run_date=datetime.now() + timedelta(seconds=10)
    )

# 启动调度器
if not scheduler.running:
    scheduler.start()
    on_startup()

@app.route('/')
def home():
    return """
    <h1>Flask 定时任务示例</h1>
    <p>已启动的定时任务：</p>
    <ul>
        <li>日报任务：每天9:00执行</li>
        <li>5分钟任务：每5分钟执行一次</li>
    </ul>
    <p>查看控制台输出获取执行详情</p>
    """

@app.route('/jobs')
def list_jobs():
    """查看当前所有定时任务"""
    jobs = scheduler.get_jobs()
    job_list = [f"ID: {job.id}, 下次执行: {job.next_run_time}" for job in jobs]
    return "
".join(job_list)

@app.route('/add_job/<minutes>')
def add_job(minutes):
    """动态添加定时任务"""
    try:
        minutes = int(minutes)
        job_id = f"dynamic_job_{minutes}min"
        
        scheduler.add_job(
            id=job_id,
            func=lambda: print(f"动态添加的任务 - 每 {minutes} 分钟执行一次"),
            trigger='interval',
            minutes=minutes,
            replace_existing=True
        )
        
        return f"成功添加每 {minutes} 分钟执行的任务!"
    except ValueError:
        return "参数错误，请输入有效的分钟数"

@app.route('/remove_job/<job_id>')
def remove_job(job_id):
    """移除定时任务"""
    if scheduler.get_job(job_id):
        scheduler.remove_job(job_id)
        return f"已移除任务: {job_id}"
    return "任务不存在"

if __name__ == '__main__':
    app.run(debug=True)

pip install flask apscheduler

trigger='cron', hour=9, minute=0  # 每天9:00执行

trigger='interval', minutes=5  # 每5分钟执行

trigger='date', run_date=datetime.now() + timedelta(seconds=10)

jobstores={
    'default': SQLAlchemyJobStore(url='sqlite:///jobs.db')
}

# 在应用工厂函数中初始化调度器
def create_app():
    app = Flask(__name__)
    
    if os.environ.get('WERKZEUG_RUN_MAIN') == 'true' or not app.debug:
        # 确保只在主进程中初始化调度器
        scheduler = BackgroundScheduler(daemon=True)
        scheduler.start()
        # 添加任务...
    
    return app

# Celery配置示例
from celery import Celery
from celery.schedules import crontab

app = Celery('tasks', broker='redis://localhost:6379/0')

@app.on_after_configure.connect
def setup_periodic_tasks(sender, **kwargs):
    # 每10分钟执行
    sender.add_periodic_task(600, task1.s(), name='每10分钟任务')
    
    # 每天午夜执行
    sender.add_periodic_task(
        crontab(hour=0, minute=0),
        task2.s(),
        name='每日午夜任务'
    )

@app.task
def task1():
    print("执行每10分钟任务")

@app.task
def task2():
    print("执行每日任务")

def job_with_error_handling():
    try:
        # 业务代码
        pass
    except Exception as e:
        app.logger.error(f"任务执行失败: {str(e)}")
        # 发送警报邮件
        send_alert_email(f"任务失败: {str(e)}")
        
def send_alert_email(message):
    # 实现邮件发送逻辑
    pass