APSchedular Python 示例

您可以使用 Advanced Python Scheduler (APScheduler) Python 包，将 Python 代码安排在稍后执行，可以只执行一次，也可以经常执行。您可以灵活地按需添加新任务并删除旧任务。

如果您将任务保存在数据库中，它们将能够承受调度器的重启并保留其状态。所有在调度器关闭期间本应完成的任务，将在调度器重新启动后完成。

APScheduler 的一个主要优点是它可以在多个平台上使用，并且可以替代 Windows 任务计划程序或 cron 守护进程。此外，在撰写本文时，它仍在开发中。

APScheduler 提供以下三种标准计时方法

• Cron 风格调度（具有可选的开始/结束时间）。
• 基于间隔的执行（在可选的开始和结束时间，以固定的间隔运行任务）。
• 一次性延迟执行（仅一次，在预定的时间运行任务）。

设置

安装过程非常简单易懂。由于 apscheduler 是一个内置包，您应该已经拥有它了。让我们尝试一下，如果包还没有安装，我们就安装它。激活您的虚拟环境后，运行以下命令

如果没有输出，请不要惊慌；这仅仅意味着包尚未安装。如果我们进行以下操作，安装它将很简单

基本 API

为了让本教程简短明了，我将只介绍 APScheduler 的四个主要组件中的两个

• 调度器
• 诱因

您将用来添加或删除作业的关键组件就是调度器。大多数用例不需要一个以上的调度器。根据编程环境，许多调度器类型都包含在模块中。

例如：BlockingScheduler, AsyncIOScheduler, TornadoScheduler, QtScheduler 等。

在本教程中，我们将使用 BackgroundScheduler，因为我们希望任务在 Flask 服务器的后台执行。

接下来是触发器组件，它包含了调度逻辑。每次将任务添加到调度器时；您都必须定义一个触发器，该触发器指定下次何时运行。APScheduler 内置三种触发器

• Date：当您只想在特定时间运行一次作业时使用此选项。
• 当您希望作业以固定的时间间隔运行，请使用 **Interval**。
• 当您希望在特定时间(s)重复执行任务时，请使用 **cron**。

稍后，我们将使用 interval 和 cron 触发器来展示它们之间的区别。让我们看看如何编写三行代码来构建我们自己的任务调度作业。

第一行：导入

根据我们要使用的调度器，我们必须导入所需的组件。在这种情况下，我们将导入 BackgroundScheduler。

第二行：调度器

创建 BackgroundScheduler 时，daemon 参数应设置为 True。通过这样做，我们可以在退出 Flask 程序时终止线程。

第三行：添加作业

将使用 add_job 函数将作业添加到调度器。有三个主要的可配置参数。

• Function：调度器将按名称调用指定的函数。您也可以传递一个匿名函数。
• Trigger：您正在使用的触发器。
• Parameters：触发器的设置。每个触发器都有自己的一组参数。

看下面的代码行，它每五秒更新一次调度器的可用作业列表。

第四行：启动调度器

完成设置后，使用以下行启动调度器

如果您在 Python shell 中执行代码，您应该能看到以下输出

如果您按 Ctrl-C 取消 KeyboardInterrupt，调度器将继续运行，直到您按 Ctrl-Z 退出 Python Shell 或使用 shutdown 函数。

与 Flask 集成

APScheduler 可以轻松地与 Flask 集成，没有任何问题。创建一个名为 test.py 的新 Python 文件进行测试。

进口

作业函数

这次，我们不使用匿名函数，而是使用一个真实的函数。在新创建的 sensor 函数中，输入以下代码。count 变量仅用作示例，以显示我们迄今为止计划了多少作业。

调度器

除了使用 cron 触发器而不是 interval 之外，我们将以与前几个阶段相同的方式定义调度器。

取而代之的是，我们将指定 minute 选项，以便操作每分钟运行一次。在此情况下使用星号。如果您将其设置为 5，它将在每小时的五分钟点被调用一次。

当添加具有参数的函数时，可以使用 args 参数来传递参数。它将接受一个元组或列表。假设我们之前使用的函数有两个参数。add_job 函数可以这样调用

Flask

剩余的任务是定义运行 Flask 服务器所需的函数。

打开控制台后，将目录切换到 test.py 文件所在的位置。运行命令

sensor 函数可以轻松修改以执行您想要的任务。示例包括

• 自动化从数据库检索数据的过程。
• 每天对服务器进行分析。
• 计划模型训练和更新的特定时间。

只需几行代码即可实现各种用例。

触发器示例

在本部分中，我们将提供一些示例，展示如何根据您的需求自定义触发器。

Interval

一些最有用的配置选项包括以下内容

• Weeks：周，延迟的持续时间。
• Days：需要等待的天数。
• Hours：总共需要等待的小时数。
• Minutes：需要等待的分钟数。
• Seconds：等待的持续时间。
• start_date 作为区间计算的起点。
• end_date：触发器可以使用的最后日期或时间。

以下代码将使其每两个小时运行一次

为了确保作业仅在开始和结束时间戳之间执行，您可以使用 start 和 end 参数。

Cron

为了确保作业仅在开始和结束时间戳之间执行，您可以使用 start 和 end 参数。

由于它提供了更多的设置选项，cron 触发器更加复杂。

• year - 四位数的年份
• month - 月份（1-12）
• day - 月份中的第一天（1-31）
• week - ISO 周（1-53）
• day_of_week - 一周中的某一天（0-6 或 mon、tue、wed、thu、fri、sat、sun）。每周一为第一天。
• hour - 小时（0-23）
• minute - 分钟（0-59）
• second - 秒（0-59）
• start_date - 触发的最早日期或时间（包含）
• end_date - 触发的最后日期或时间（包含）

与 interval 触发器不同，参数的末尾没有 *，并且接受字符串或整数作为输入。从 year 到 second 的字符串输入字段可接受以下表达式。

计划作业在“2 a.m.、3 a.m.、4 a.m. 和 5 a.m.”的“2 月、3 月、4 月、11 月和 12 月”的每个第三个星期五运行

计划每月最后一个星期日运行任务

计划每小时运行任务

组合触发器

根据您的用例，如果出现需要的情况，可以使用 AndTrigger 和 OrTrigger 来组合触发器。

计划作业每三个小时运行一次（仅限周一和周二）

结论

我们的第一步是安装 APScheduler 模块。

然后，我们详细检查了模块的两个基本部分：调度器和触发器。我们还测试了调度器的四行 Python Shell 实现。

下一步是将其集成到基本的 Flask 服务器中。我们使用了一个真实的函数，而不是一个匿名函数。

之后，我们了解了配置触发器组件参数的一些示例。我们还可以使用 AndTrigger 和 OrTrigger 组合 interval 和 cron 触发器。