正文

如何用 Flask 实现定时任务并轻松监控任务执行状态，避免意外中断？

/0 浏览量
0808

在 Flask 应用中实现定时任务通常需要借助外部库，如 APScheduler，因为它可以提供灵活的定时任务调度功能。同时，为了监控任务执行状态并避免意外中断，我们可以结合使用数据库记录任务状态和日志记录。以下是一个详细的实现方案：

1. 安装 Flask 和 APScheduler

首先，确保你已经安装了 Flask。然后，安装 APScheduler：

pip install Flask APScheduler

2. 创建 Flask 应用

创建一个基本的 Flask 应用，并设置 APScheduler。

from flask import Flask
from apscheduler.schedulers.background import BackgroundScheduler

app = Flask(__name__)

# 创建一个调度器
scheduler = BackgroundScheduler()

def job_function():
    print("执行定时任务")

# 添加定时任务，每天凌晨1点执行
scheduler.add_job(job_function, 'cron', hour=1)

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

3. 记录任务执行状态

为了监控任务执行状态，我们可以使用数据库来存储任务信息。以下是一个简单的数据库模型：

from flask_sqlalchemy import SQLAlchemy

app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///tasks.db'
db = SQLAlchemy(app)

class Task(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(50), nullable=False)
    status = db.Column(db.String(20), nullable=False, default='pending')
    last_run = db.Column(db.DateTime)

    def __init__(self, name):
        self.name = name

# 创建数据库表
db.create_all()

4. 更新任务状态

在 job_function 函数中，更新任务状态：

from datetime import datetime

def job_function():
    task = Task.query.filter_by(name='定时任务').first()
    if task:
        task.status = 'completed'
        task.last_run = datetime.utcnow()
        db.session.commit()
    print("执行定时任务")

5. 监控任务执行状态

你可以通过查询数据库来监控任务执行状态：

@app.route('/tasks')
def tasks():
    tasks = Task.query.all()
    return {'tasks': [{'name': task.name, 'status': task.status, 'last_run': task.last_run} for task in tasks]}

6. 避免意外中断

为了防止意外中断导致任务未完成，我们可以设置一个守护进程来监控 Flask 应用。在 Linux 系统中，可以使用 supervisord 或 systemd。

以下是一个 systemd 服务示例：

[Unit]
Description=Flask App Service

[Service]
User=your_username
WorkingDirectory=/path/to/your/app
ExecStart=/usr/bin/python3 /path/to/your/app/app.py
Restart=always

[Install]
WantedBy=multi-user.target

将上述内容保存为 flask-app.service，然后使用以下命令启动服务：

sudo systemctl start flask-app.service
sudo systemctl enable flask-app.service

通过以上步骤，你可以在 Flask 应用中实现定时任务，并轻松监控任务执行状态，同时避免意外中断。

-- 展开阅读全文 --