APScheduler 是一个 Python 库,用于在你的应用程序中添加作业调度功能。它支持多种不同的作业存储和运行作业的方式,APScheduler 是一个功能强大且灵活的任务调度库,支持复杂的调度策略,包括但不限于 cron 表达式。通过合理利用 APScheduler,可以轻松实现复杂的定时任务调度逻辑,以满足应用程序需求。

class apscheduler.triggers.cron.CronTrigger(year=None, month=None, day=None, week=None, day_of_week=None, hour=None, minute=None, second=None, start_date=None, end_date=None, timezone=None, jitter=None)

父类:apscheduler.triggers.base.BaseTrigger

当前时间与所有指定的时间约束匹配时触发,类似于Linux cron定时任务程序。

参数:

year(int | str) - 4位数年份
month(int | str) - 月(1-12)
day(int | str) - (1-31)日
week(int | str) - ISO week(1-53)
day_of_week(int | str) - 工作日的数字或名称(0-6或星期一,星期二,星期三,星期四,星期五,星期五,星期日)
hour (int | str) - 小时(0-23)
minute(int | str) - 分钟(0-59)
second(int | str) - second(0-59)
start_date(datetime | str) - 触发的最早可能日期/时间(包括)
end_date(datetime | str) - 触发的最新可能日期/时间(包括)
timezone(datetime.tzinfo | str) - 用于日期/时间计算的时区(默认为调度程序时区)
jitter(int | None) - jitter最多提前或延迟作业执行几秒钟。

注意:第一个工作日总是星期一。

apscheduler.triggers.cron介绍

它是APScheduler中最强大的内置触发器。您可以在每个字段上指定各种不同的表达式,在确定下一个执行时间时,它会找到满足每个字段中条件的最早可能时间。此行为类似于大多数类UNIX操作系统中的“Cron”实用程序。

您还可以分别通过start_date和 end_date参数指定cron样式计划的开始日期和结束日期。它们可以作为日期/日期时间对象或文本( ISO 8601格式)给出。

与crontab表达式不同,您可以省略不需要的参数。如果省略的参数是大于指定的参数的默认是为*,但除了week和 day_of_week,它们的默认值也是*。例如,仅指定day=1, minute=20,这就等于year='*', month='*', day=1, week='*', day_of_week='*', hour='*', minute=20, second=0,工作将在每个月的第一天以每小时20分钟的时间执行。

注意:在APScheduler 2.0中更改了省略字段的行为。省略的字段以前总是默认为*

cron表达式介绍

*:触发所有值
*/aa从最小值开始,触发每个值
a-b:触发a-b范围内的任何值(必须小于b)
a-b/c:触发在a-b之间的每个c
xth y:第y个工作日中第x天发触发
last x:在一个月中第x个工作日中的最后一天触发
last:在月末的最后一天触发
x,y,z:触发任何匹配的表达式; 可以组合任意数量的任何上述表达式。

注意:xth y,last x,last这3个是用在day参数中,其它所有参数都可以使用。

夏时制时间问题

cron触发器与所谓的“挂钟”时间一起工作。因此,如果所选时区遵守DST(夏时制),您应该知道在进入或离开DST时它可能会导致cron触发器出现意外行为。从标准时间切换到夏时制时,时钟会向前移动一小时或半小时,具体取决于时区。同样,当切换回标准时间时,时钟向后移动一小时或半小时。这将导致一些时间段根本不存在或重复。如果您的日程安排会在其中任何一个时段执行作业,则可能会比预期更频繁地执行或更少执行。这不是一个错误。如果您希望避免这种情况,请使用不遵守DST的时区,例如UTC。

例如,以下时间表可能存在问题:

#在欧洲/赫尔辛基时区,3月的最后一个周日上午根本不会执行
#同样,它将在10月的最后一个星期天上午执行两次
sched.add_job(job_function, 'cron', hour=3, minute=30)

使用示例代码

from apscheduler.schedulers.blocking import BlockingScheduler

def job_function():
    print "Hello World"
sched = BlockingScheduler()
#在6月,7月,8月,11月和12月的第三个星期五的00:00,01:00,02:00和03:00执行job_function
sched.add_job(job_function, 'cron', month='6-8,11-12', day='3rd fri', hour='0-3')
sched.start()

您可以使用start_dateend_date限制计划运行的总时间:

#在2014-05-30 00:00:00之前,从周一到周五每天5:30 (am) 运行。
sched.add_job(job_function, 'cron', day_of_week='mon-fri', hour=5, minute=30, end_date='2014-05-30')

通过scheduled_job()装饰器使用@sched.scheduled_job('cron', id='my_job_id', day='last sun')

def some_decorated_task():
    print("I am printed at 00:00:00 on the last Sunday of every month!")

使用标准crontab表达式使用

sched.add_job(job_function, CronTrigger.from_crontab('0 0 1-15 may-aug *'))

jitter选项使您可以将随机添加到执行时间。如果您有多个服务器并且不希望它们在同一时刻运行作业,或者您希望阻止作业在繁忙时段运行,那么这可能很有用:

#运行 `job_function` 在[-120,+120]秒中随机选择一个额外延迟时间。
sched.add_job(job_function, 'cron', hour='*', jitter=120)

推荐文档