Flask 中的 apscheduler 执行两次 [重复]

Python

Flask是一个流行的Python Web框架，提供了许多方便的功能和扩展，使得开发Web应用变得简单而高效。其中一个强大的扩展是apscheduler，它允许我们在后台执行定时任务。本文将介绍如何使用Flask中的apscheduler来执行两次重复任务，并提供一个简单的案例代码。

Flask中的apscheduler扩展

Flask中的apscheduler扩展是基于APScheduler库的封装，它提供了一种简单而灵活的方式来安排和执行定时任务。通过apscheduler扩展，我们可以轻松地在Flask应用中设置和管理定时任务，无需编写复杂的代码。

安装和配置apscheduler

首先，我们需要安装apscheduler扩展。可以使用pip命令来安装：

pip install apscheduler

安装完成后，在Flask应用中导入apscheduler扩展：

Python
from flask_apscheduler import APScheduler

接下来，我们需要配置apscheduler。在Flask应用的配置文件中，添加以下配置：

Python
JOBS = [
    {
        'id': 'job1',
        'func': 'app:task',
        'trigger': 'interval',
        'seconds': 60  # 每隔60秒执行一次任务
    }
]
SCHEDULER_API_ENABLED = True

上述配置中，我们定义了一个名为job1的定时任务，它将执行名为task的函数，触发器设置为每隔60秒执行一次。你可以根据需求自定义触发器的时间间隔。

定义定时任务

在Flask应用中，我们需要定义要执行的具体任务。在本例中，我们将创建一个简单的任务函数，打印一条消息到控制台：

Python
from datetime import datetime
def task():
    print("定时任务执行于:", datetime.now())

以上代码定义了一个名为task的函数，它将打印当前时间到控制台。

添加apscheduler到Flask应用

接下来，我们需要将apscheduler扩展添加到Flask应用中。在Flask应用的工厂函数中，创建一个APScheduler实例，并将其初始化为Flask应用的扩展：

Python
def create_app():
    app = Flask(__name__)
    
    # 其他应用配置...
    
    # 添加apscheduler扩展
    scheduler = APScheduler()
    scheduler.init_app(app)
    
    # 添加定时任务
    scheduler.add_job(id='job1', func=task, trigger='interval', seconds=60)
    
    # 启动定时任务
    scheduler.start()
    
    return app

以上代码将在Flask应用中添加了一个名为job1的定时任务，它将执行我们之前定义的task函数，并设置触发器为每隔60秒执行一次。最后，调用scheduler.start()方法来启动定时任务。

测试定时任务

现在，我们可以运行Flask应用并测试定时任务是否按照预期执行。在终端中运行以下命令启动Flask应用：

flask run

在控制台中，你将看到每隔60秒打印一条消息，显示当前时间。这表明定时任务已经成功设置和执行。

本文介绍了如何使用Flask中的apscheduler扩展来执行两次重复任务。我们首先安装和配置了apscheduler扩展，然后定义了要执行的定时任务。最后，将apscheduler扩展添加到Flask应用中，并启动定时任务。通过这个简单的案例，你可以在自己的Flask应用中轻松地使用apscheduler来执行定时任务。

Python
from flask import Flask
from flask_apscheduler import APScheduler
from datetime import datetime
def task():
    print("定时任务执行于:", datetime.now())
def create_app():
    app = Flask(__name__)
    # 其他应用配置...
    # 添加apscheduler扩展
    scheduler = APScheduler()
    scheduler.init_app(app)
    # 添加定时任务
    scheduler.add_job(id='job1', func=task, trigger='interval', seconds=60)
    # 启动定时任务
    scheduler.start()
    return app
app = create_app()
if __name__ == "__mAIn__":
    app.run()