基于Python的 APScheduler 的简单使用
2019-05-29 本文已影响0人
best_su
一. 安装APScheduler
pip install apscheduler
二. 基本概念
APScheduler的四大组件:
- 触发器triggers: 用于设定触发任务的条件
- 任务存储器job stores: 用于存放任务,吧任务存放在内存或数据库中
- 执行器 executors: 用于执行任务,可以设定执行模式为单线程或者线程池
- 调度器 schedulers:把上方的三个组件作为参数, 通过创建调度器实例来运行
三.触发器组件
- data 日期: 触发任务运行的具体日期
- interval 间隔:触发任务运行的时间间隔
- cron 周期: 触发任务运行的周期
四.调度器组件
-
BlockingScheduler 阻塞式调度器：适用于只跑调度器的程序。
-
BackgroundScheduler 后台调度器：适用于非阻塞的情况，调度器会在后台独立运行。
-
AsyncIOScheduler AsyncIO调度器，适用于应用使用AsnycIO的情况。
-
GeventScheduler Gevent调度器，适用于应用通过Gevent的情况。
-
TornadoScheduler Tornado调度器，适用于构建Tornado应用。
-
TwistedScheduler Twisted调度器，适用于构建Twisted应用。
-
QtScheduler Qt调度器，适用于构建Qt应用。
五. 代码
date: 指定时间点出发任务
# 使用年月日 定时
from datetime import *
from apscheduler.schedulers.blocking import BlockingScheduler
sched = BlockingScheduler()
def temp_func(text):
print(text)
# 在2019年11月6日执行
sched.add_job(temp_func, 'date', run_date=date(2019, 11, 6), args=['text'])
sched.start()
# 使用年月日 时分秒 定时
from datetime import *
from apscheduler.schedulers.blocking import BlockingScheduler
sched = BlockingScheduler()
def temp_func(text):
print(text)
# 在2019年11月6日14时6分40秒执行
sched.add_job(temp_func, 'date', run_date=datetime(2019, 4, 9, 14 , 6, 40), args=['text'])
#也可以这样
#sched.add_job(temp_func, 'date', run_date='2009-11-06 16:30:05', args=['text'])
sched.start()
#如果不指定时间 就会立即执行
from datetime import *
from apscheduler.schedulers.blocking import BlockingScheduler
sched = BlockingScheduler()
def temp_func(text):
print(text)
sched.add_job(temp_func, args=['text'])
interval周期触发任务
from datetime import datetime
from apscheduler.schedulers.blocking import BlockingScheduler
def temp_function():
print("Hello World")
sched = BlockingScheduler()
# 每1小时触发 一次
sched.add_job(temp_function, 'interval', hours=1)
#可以设置周期的开始时间和结束时间
sched.add_job(job_function, 'interval', hours=2, start_date='2019-10-1 09:30:00', end_date='2020-06-1 11:00:00')
sched.start()
cron 强大的类crontab表达式
# 注意参数顺序
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
)
month和day_of_week参数分别接受的是英语缩写jan– dec 和 mon – sun.
当省略时间参数时，在显式指定参数之前的参数会被设定为，之后的参数会被设定为最小值，week 和day_of_week的最小值为。比如，设定day=1, minute=20等同于设定year='', month='', day=1, week='', day_of_week='', hour='*', minute=20, second=0，即每个月的第一天，且当分钟到达20时就触发
表达式类型
| 序号 | 表达式 | 参数 | 类型 |
|---|---|---|---|
| 1 | * | 所有 | 通配符。例：minutes=*即每分钟触发 |
| 2 | */a | 所有 | 可被a整除的通配符。 |
| 3 | a-b | 所有 | 范围a-b触发。 |
| 4 | a-b/c | 所有 | 范围a-b，且可被c整除时触发。 |
| 5 | xth y | 日 | 第几个星期几触发。x为第几个，y为星期几。 |
| 6 | last x | 日 | 一个月中，最后个星期几触发。 |
| 7 | last | 日 | 一个月最后一天触发。 |
| 8 | x,y,z | 所有 | 组合表达式，可以组合确定值或上方的表达式。 |
from apscheduler.schedulers.blocking import BlockingScheduler
def temp_function():
print "Hello World"
sched = BlockingScheduler()
# 任务会在6月、7月、8月、11月和12月的第三个周五，00:00、01:00、02:00和03:00触发
sched.add_job(temp_function, 'cron', month='6-8,11-12', day='3rd fri', hour='0-3')
sched.start()
- start_date 和 end_date 可以用来适用时间范围
# 在2014-05-30 00:00:00前，每周一到每周五 5:30运行
sched.add_job(job_function, 'cron', day_of_week='mon-fri', hour=5, minute=30, end_date='2014-05-30')
通过装饰器来实现
from apscheduler.scheduler import BlockingScheduler
@sched.scheduled_temp('interval', id='temp_id', hours=2)
def temp_function():
print("Hello World")
