【Python 模块】APScheduler 实现定时任务

概述

在项目中,我们可能遇到有定时任务的需求。

  • 定时执行任务:例如每天早上 8:00 定时推送早报。
  • 间隔执行任务:比如:爬虫间隔多少时间去爬取一次。

模块简介

它是一个轻量级的 Python 定时任务调度框架。有四种组件,分别是:调度器(scheduler)作业存储(job store)触发器(trigger)执行器(executor),这里只介绍 触发器(trigger)。同时,它还支持异步执行、后台执行调度任务。

触发器

APScheduler 有三种内建的 触发器。

  • date 触发器:表示特定时间点触发,只执行一次。
  • interval 触发器:固定时间间隔触发
  • cron 触发器:在特定时间周期触发,最强大的触发器,能实现每天固定时间执行功能。

date 触发器

参数:

参数 说明
run_date (datetime 或 str) 作业的运行日期或时间
timezone (datetime.tzinfo 或 str) 指定时区
from apscheduler.schedulers.blocking import BlockingScheduler
from datetime import datetime

def func():
    print(datetime.now().strftime("%Y-%m-%d %H:%M:%S"))

scheduler = BlockingScheduler()
# 2020-03-16 17:50:00  指定时间运行一次
scheduler.add_job(func,next_run_time=datetime(2020, 3, 16, 17, 50))
scheduler.start()

interval 触发器

以下都是可选参数(如果不填,就是 当前开始 间隔一秒 执行一次 但不会执行太久,会自动停止):

参数 说明
weeks (int) 间隔几周
days (int) 间隔几天
hours (int) 间隔几小时
days (int) 间隔几天
minutes (int) 间隔几分钟
seconds (int) 间隔多少秒
start_date (datetime 或 str) 开始日期
end_date (datetime 或 str) 结束日期
timezone (datetime.tzinfo 或str) 时区
from apscheduler.schedulers.blocking import BlockingScheduler
from datetime import datetime

def func():
    print(datetime.now().strftime("%Y-%m-%d %H:%M:%S"))

scheduler = BlockingScheduler()
# 每 6 秒执行一次
scheduler.add_job(func,trigger = "interval",seconds=6)
scheduler.start()

cron 触发器

特定时间周期性地触发,和Linux crontab格式兼容。它是功能最强大的触发器。

参数:

header 1 header 2
year (int 或 str) 年,4位数字
month (int 或 str) 月 (范围1-12)
day (int 或 str) 日 (范围1-31
week (int 或 str) 周 (范围1-7)
day_of_week (int 或 str) 周内第几天或者星期几 (范围0-6 或者 mon,tue,wed,thu,fri,sat,sun)
hour (int 或 str) 时 (范围0-23)
minute (int 或 str) 分 (范围0-59)
second (int 或 str) 秒 (范围0-59)
start_date (datetime 或 str) 最早开始日期(包含)
end_date (datetime 或 str) 最晚结束时间(包含)
timezone (datetime.tzinfo 或str) 指定时区
from apscheduler.schedulers.blocking import BlockingScheduler
from datetime import datetime

def func():
    print(datetime.now().strftime("%Y-%m-%d %H:%M:%S"))

scheduler = BlockingScheduler()
# 在 1到3月 和 7到9月的 8到12 和 16到20 的30分 执行
scheduler.add_job(func,trigger =  'cron', month='1-3,7-9', hour="8-12,16-20" ,minute ='30')
scheduler.start()
发表评论 / Comment

用心评论~