下一个→ ←上一页

APSchedular Python示例

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

如果将作业放入数据库中，它们将承受调度程序重新启动并保持其状态。在调度器关闭时应该完成的所有任务将在重新启动时完成。

APScheduler的主要优点之一是它可以在多个平台上使用，并且可以取代Windows Task Scheduler或cron守护进程。此外，在撰写本文时，它正在发展中。

APScheduler提供这三种标准计时方法:

cron风格的调度(具有可选择的开始/结束时间)。
基于间隔的执行(以可选的开始时间和结束时间运行任务，间隔均匀)。
一次性延迟执行(在预定的时间只运行一次任务)。

设置:

安装过程非常简单，易于操作。因为apscheduler是一个内置包，所以您应该已经有了它。让我们试一试，如果包还没有安装，那么就安装它。激活虚拟环境后，执行以下命令:

如果没有任何输出，不要惊慌;它只是暗示包没有安装。如果我们执行以下操作，安装它很简单:

基本API:

为了使这一课简单明了，我将简单地介绍APScheduler的四个主要组件中的两个:

调度器
触发

用于添加或删除作业的关键元素是调度程序。大多数用例不需要一个以上的调度器。根据编程环境的不同，许多调度器类型被合并到模块中。

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

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

接下来是包含调度逻辑的触发器组件。每次向调度程序添加作业时;您必须定义触发器，以指定下一次应该在何时运行该触发器。APScheduler中有三个内置触发器:

日期:当您只想在特定时间运行作业一次时，可以使用此选项。
使用时间间隔如果您希望作业在固定的时间内运行。
当您希望在特定时间重复执行任务时，使用cron．

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

第一行:导入

根据我们计划使用的调度程序，我们必须导入所需的组件。在这个实例中，我们将导入BackgroundScheduler。

第二行:调度程序

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

第三行:增加一个工作

add_job函数将用于向调度程序添加作业。有三个主要的可配置参数。

功能:调度器将通过此名称调用指定的函数。你也可以传递一个匿名函数。
触发:你使用的触发器。
参数:触发器设置。每个触发器都有自己的一组参数。

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

            固定播送时间。Add_job (lambda: schedule .print_jobs()，'interval'，seconds=5)
           

第四行:启动调度器:

完成设置后，使用下面的行启动调度器:

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

如果你按Ctrl-C取消KeyboardInterrupt，调度器将继续运行，直到你按Ctrl-Z离开Python Shell或使用关机函数。

与Flask的集成

APScheduler可以简单地集成到Flask中，没有任何问题。创建一个新的名为test.py的Python文件来测试它。

进口

            包括即将到来的导入。从apscheduler.schedulers.background导入BackgroundScheduler
           

工作职能:

这次我们将使用一个真实的函数，而不是使用匿名函数。在新形成的传感器功能中，输入如下代码。count变量只是作为一个示例来显示到目前为止我们计划了多少个作业。

            count = 0 def sensor():全局计数schedule .print_jobs() print(' count: '， count) count += 1
           

调度程序:

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

相反，我们将指定minute选项，以便操作每分钟运行一次。星号符号应用于这种情况。如果给它赋值5，它将每小时在5分钟点被调用一次。

            sched = BackgroundScheduler(daemon=True) schedule .add_job(sensor，'cron'，minute='*') schedule .start()
           

参数args可用于在添加具有形参的函数时传入形参。它将接受元组或列表。假设我们之前使用的函数有两个参数。add job函数的调用方式如下:

            固定播送时间。add_job(传感器,args = [param1, param2], cron,分钟=‘*’)
           

瓶

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

            应用=烧瓶(__name__) if __name__ = =“__main__”:app.run(0.0.0.0,端口= 5000)
           

打开该目录后，将其移动到控制台中test.py文件的位置。运行命令行:

传感器功能可以很容易地改变，以执行您想要的职责。例子包括:

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

只需几行代码就可以完成各种用例。

触发例子:

本部分将给出几个示例，向您展示如何根据自己的需求定制触发器。

间隔:

一些最有用的配置选项包括:

Weeks:周，延迟的持续时间。
天:你要等多久，单位是天。
小时:等待的总时间。
分钟:等待的时间。
秒:等待时间。
开始日期作为间隔计算的起点。
结束日期:可用于触发的最近日期或时间。

下面的代码将使它每两小时运行一次:

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

            固定播送时间。Add_job (job_function， 'interval'， hours=2, start_date='2019-11-11 09:00:00'， end_date='2019-12-25 11:00:00')
           

Cron:

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

因为它提供了更多的设置选择，所以cron触发器要复杂得多。

年——四位数的年
每月(1-12)
月的第一天(1-31)
ISO周为一周(1-53)
星期几——日期或名称(0-6或mon, tue, wed, thu, fri, sat, sun)。每个星期，星期一是第一天。
一小时后(0-23)
间隔分钟(0-59)
第三-第三(0-59)
最早可行的触发日期或时间为开始日期(含)
结束日期-可用于触发的最晚日期或时间(包括)

与interval触发器不同，参数不像结束时那样具有，而是以字符串或int形式接受输入。以下表达式可用于从年到秒的字符串输入字段。

时间表在“2月、3月、4月、11月和12月”的每个第三个星期五“凌晨2点、3点、4点和5点”工作:

            固定播送时间。add_job (job_function, cron,月=“2 - 4、11 - 12”,天= 3日星期五,小时=“2 - 5”)
           

时间表在每个月的最后一个星期天运行任务:

            固定播送时间。Add_job (job_function， 'cron'， day='最后一个太阳')
           

时间表每小时运行一次任务:

结合触发器:

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

将作业安排为每三小时运行一次(仅在周一和周二):

            从apscheduler.triggers.interval导入IntervalTrigger from apscheduler.triggers.cron导入CronTrigger trigger = AndTrigger([IntervalTrigger(hours=3)， CronTrigger(day_of_week='mon,tue')])调度程序。add_job (job_function触发)
           