Python进阶(9)定时运行程序APScheduler

1. 前言

• 官网&＃xff0c;源码&＃xff0c;官方examples
• 安装pip install apscheduler
• 组件介绍
  • triggers&＃xff1a;什么时候触发任务。
  • job stores&＃xff1a;默认情况下将任务保存在内存中。还没用到需要序列化到数据库的情况。
  • executors&＃xff1a;执行Job&＃xff0c;主要取决于用了什么框架。默认使用线程池&＃xff0c;常用的还有进程池。
  • schedulers&＃xff1a;用于管理以上三个组件。

1. Scheduler

1.1. 基本概念

• 作用&＃xff1a;用于管理 triggers&＃xff0c;job stores&＃xff0c;executors。
• 如何选择&＃xff1a;主要就是根据运行的环境来选择。
  • BlockingScheduler&＃xff1a;程序只有调度器运行时选择。
  • BackgroundScheduler&＃xff1a;没有使用其他框架&＃xff0c;且希望调度器在应用后台运行。

1.2. 基本API

• 所有的Scheduler都继承自 BaseScheduler。
• 初始化函数 def __init__(self, gconfig&＃61;{}, **options):
  • gconfig 可以看做是一个全局初始化参数。
  • **options 可以看做是自定义初始化参数。
  • 上面两者所包含的参数是完全相同的&＃xff0c;且自定义初始化参数会覆盖全局初始化参数。
  • 包含的参数有&＃xff1a;
    • logger&＃xff1a;str/logging.Logger&＃xff0c;用于输出日志信息。
    • timezone&＃xff1a;str/datetime.tzinfo&＃xff0c;指定时区。
    • jobstore_retry_interval&＃xff1a;int/float&＃xff0c;当job store抛出异常时&＃xff0c;最短的重试时间间隔。
    • job_defaults&＃xff1a;dict&＃xff0c;新建jobs的默认参数
    • jobstores&＃xff1a;dict&＃xff0c;job store alias -> job store instance or configuration dict
    • executors&＃xff1a;dict&＃xff0c; executor alias -> executor instance or configuration dict
• configure(gconfig&＃61;{}, prefix&＃61;&＃39;apscheduler.&＃39;, **options)&＃xff1a;在开始运行前设置参数。
  • 参数类型与初始化时相同。
• start(paused&＃61;False)&＃xff1a;运行scheduler。
  • 如果paused&＃61;True则需要等到运行 resume() 方法后才会真正运行scheduler。
• shutdown(wait&＃61;True)&＃xff1a;关闭scheduler
  • 同时关闭了 executors和job stores。
  • 不会停止正在运行的jobs。
  • wait&＃61;True时会等待所有jobs运行完成。
• pause()&＃xff1a;暂停&＃xff0c;与resume()配合使用。
• resume()&＃xff1a;重新开始&＃xff0c;与pause()配合使用。
• add/remove_executor
• add/remove_jobstore
• add/remove_listener
• job相关API
  • add_job()&＃xff1a;参数太多
    • func
    • trigger&＃xff1a;str或trigger对象&＃xff0c;初始化参数通过 **trigger_args 来获得。
    • args&＃xff1a;func 参数
    • kwargs&＃xff1a;func 参数
    • id
    • name
    • misfire_grace_timec
    • coalesce
    • max_instances
    • next_run_time
    • jobstore&＃xff1a;str&＃xff0c;通过 alias 来选择。
    • executor&＃xff1a;str&＃xff0c;通过 alias 来选择。
    • replace_existing
    • **trigger_args&＃xff1a;用于 trigger
  • modify_job(job_id, jobstore&＃61;None, **changes)
    • 改变某个job的参数。
  • reschedule_job(job_id, jobstore&＃61;None, trigger&＃61;None, **trigger_args)
    • 改变某个job的trigger&＃xff0c;并修改下次运行时间。.
  • pause/resume_job&＃xff1a;暂停/继续某个job。
  • get_jobs(jobstore&＃61;None, pending&＃61;None)
  • get_job(job_id, jobstore&＃61;None)
  • remove_job(job_id, jobstore&＃61;None)
  • remove_all_jobs(jobstore&＃61;None)
  • print_jobs(jobstore&＃61;None, out&＃61;sys.stdout)

1.3. 实例

from datetime import datetime
import time
import osfrom apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.schedulers.blocking import BlockingSchedulerdef tick():print(&＃39;Tick! The time is: %s&＃39; % datetime.now())# background sheculer 实例
scheduler &＃61; BackgroundScheduler()
scheduler.add_job(tick, &＃39;interval&＃39;, seconds&＃61;3)
scheduler.start()
print(&＃39;Press Ctrl&＃43;{0} to exit&＃39;.format(&＃39;Break&＃39; if os.name &＃61;&＃61; &＃39;nt&＃39; else &＃39;C&＃39;))
try:# This is here to simulate application activity (which keeps the main thread alive).while True:time.sleep(2)
except (KeyboardInterrupt, SystemExit):# Not strictly necessary if daemonic mode is enabled but should be done if possiblescheduler.shutdown()# blocking sheduler 实例
scheduler &＃61; BlockingScheduler()
scheduler.add_job(tick, &＃39;interval&＃39;, seconds&＃61;3)
print(&＃39;Press Ctrl&＃43;{0} to exit&＃39;.format(&＃39;Break&＃39; if os.name &＃61;&＃61; &＃39;nt&＃39; else &＃39;C&＃39;))try:scheduler.start()
except (KeyboardInterrupt, SystemExit):pass


2. trigger

2.1. 基本概念

• 基本作用&＃xff1a;决定job运行的时间点/时间间隔。
• 分类
  • cron&＃xff1a;功能最强大&＃xff0c;可以指定时间、时间间隔、时间范围等。参考
  • date&＃xff1a;指定某个时间点出发&＃xff0c;只执行一次。
  • interval&＃xff1a;指定时间间隔出发任务&＃xff0c;可指定时间范围。

2.2. apscheduler.triggers.cron.CronTrigger

• 参数很多&＃xff0c;功能强大。
• 对我来说&＃xff0c;因为在linux中也常用&＃xff0c;所以我用起来也最顺手。
• 主要参数&＃xff1a;
  • year&＃xff1a;int/str&＃xff0c;4位
  • month&＃xff1a;int/str&＃xff0c;[1, 12]
  • day&＃xff1a;int/str&＃xff0c;[1, 31]
  • week&＃xff1a;int/str&＃xff0c;[1, 53]
  • day_of_week&＃xff1a;int/str&＃xff0c;[0, 6]&＃xff0c;从周一开始&＃xff0c;也可以用英文的钱三个小写字母。
  • hour&＃xff1a;int/str&＃xff0c;[0, 23]
  • minute&＃xff1a;int/str&＃xff0c;[0, 59]
  • second&＃xff1a;int/str&＃xff0c;[0, 59]
  • start_date&＃xff1a;datetime/str
  • end_date&＃xff1a;datetime/str
  • timezone&＃xff1a;datetime.tzinfo|str
  • jitter&＃xff1a;int/None&＃xff0c;运行时间随机偏移[-jitter, jitter]秒。
• 对于没有指定的参数&＃xff0c;根据输入参数分别设置为*或最小值&＃xff1a;
  • 大于指定元素中最小的&＃xff0c;设置为 *
  • 小于指定元素中最小的&＃xff0c;设置为其最小值。
  • 举例&＃xff1a;输入 day&＃61;1, minute&＃61;20&＃xff0c;则设置second&＃61;0&＃xff0c;其他元素为*。
• 其他常用的一些语法

2.3. 其他 trigger

• apscheduler.triggers.interval.IntervalTrigger
  • 通过设置间隔时间来确定下次运行时间。
  • 主要参数&＃xff1a;
    • weeks&＃xff1a;int
    • days&＃xff1a;int
    • hours&＃xff1a;int
    • minutes&＃xff1a;int
    • seconds&＃xff1a;int
    • start_date&＃xff1a;datetime/str
    • end_date&＃xff1a;datetime/str
    • timezone&＃xff1a;datetime.tzinfo|str
    • jitter&＃xff1a;int/None&＃xff0c;运行时间随机偏移[-jitter, jitter]秒。
• apscheduler.triggers.date.DateTrigger(run_date&＃61;None, timezone&＃61;None)
  • 只在给定的时间运行一次
  • run_date&＃xff1a;datetime|str&＃xff0c;指定运行时间。
  • timezone&＃xff1a;datetime.tzinfo|str
• 集合操作&＃xff1a;
  • apscheduler.triggers.combining.AndTrigger(triggers, jitter&＃61;None)
  • apscheduler.triggers.combining.OrTrigger(triggers, jitter&＃61;None)
  • 主要就是用于合并多个 triiger。

2.5. 举例&＃xff1a;add_job 中trigger的应用

• 回顾一下 scheduler.add_job 方法&＃xff0c;与trigger有关的参数有&＃xff1a;
  • trigger&＃61;None&＃xff1a;
    • 可以设置为字符串 date/interval/cron。
    • 也可以设置为上面提到过的各种 trigger 对象。
  • **trigger_args&＃xff1a;即 add_job 方法中所有没有定义的命名参数都会作为trigger对象的参数导入。
• 举例&＃xff1a;

# date
# 方法一
sched.add_job(my_job, &＃39;date&＃39;, run_date&＃61;datetime(2009, 11, 6, 16, 30, 5), args&＃61;[&＃39;text&＃39;])
# 方法二
sched.add_job(my_job, &＃39;date&＃39;, run_date&＃61;&＃39;2009-11-06 16:30:05&＃39;, args&＃61;[&＃39;text&＃39;])
# 方法三&＃xff08;马上执行&＃xff09;
sched.add_job(my_job, args&＃61;[&＃39;text&＃39;])# interval
# 方法一
sched.add_job(job_function, &＃39;interval&＃39;, hours&＃61;2, start_date&＃61;&＃39;2010-10-10 09:30:00&＃39;, end_date&＃61;&＃39;2014-06-15 11:00:00&＃39;)
# 方法二
&＃64;sched.scheduled_job(&＃39;interval&＃39;, id&＃61;&＃39;my_job_id&＃39;, hours&＃61;2)
def job_function():print("Hello World")# cron
# 方法一
sched.add_job(job_function, &＃39;cron&＃39;, day_of_week&＃61;&＃39;mon-fri&＃39;, hour&＃61;5, minute&＃61;30, end_date&＃61;&＃39;2014-05-30&＃39;)
# 方法二
&＃64;sched.scheduled_job(&＃39;cron&＃39;, id&＃61;&＃39;my_job_id&＃39;, day&＃61;&＃39;last sun&＃39;)
def some_decorated_task():print("I am printed at 00:00:00 on the last Sunday of every month!")
# 方法三
sched.add_job(job_function, CronTrigger.from_crontab(&＃39;0 0 1-15 may-aug *&＃39;))# AndTrigger
trigger &＃61; AndTrigger([IntervalTrigger(hours&＃61;2),CronTrigger(day_of_week&＃61;&＃39;sat,sun&＃39;)])
scheduler.add_job(job_function, trigger)# OrTrigger
trigger &＃61; OrTrigger([CronTrigger(day_of_week&＃61;&＃39;mon&＃39;, hour&＃61;2),CronTrigger(day_of_week&＃61;&＃39;tue&＃39;, hour&＃61;15)])
scheduler.add_job(job_function, trigger)


3. executors & job store

3.1. Executors

• 用于指定通过什么后端来执行任务。
• 常用的有&＃xff1a;线程池与进程池。
  • 默认使用线程池。
• 如何使用&＃xff1a;
  • 第一步&＃xff1a;在定义 scheduler 时输入 executors 参数。
    • 该参数是一个字典&＃xff0c;key为executor的别名&＃xff0c;value为executors的对象。
  • 第二步&＃xff1a;在 add_job 的时候指定对应executor的别名。
• 其他&＃xff1a;
  • 默认情况下&＃xff0c;用线程池已经够了。
  • 计算密集型任务要使用进程池。

3.2. Job Stores

• 作用&＃xff1a;将job保存到哪里。
  • 可以保存到各类数据库中。因为没用到&＃xff0c;所以没细看。
  • 默认保存到内存中。
• 如何使用&＃xff1a;
  • 第一步&＃xff1a;在定义 scheduler 时输入 jobstores 参数。
    • 该参数是一个字典&＃xff0c;key为executor的别名&＃xff0c;value为executors的对象。
  • 第二步&＃xff1a;在 add_job 的时候指定对应jobstore的别名。

3.3. 举例

from pytz import utcfrom apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore
from apscheduler.executors.pool import ThreadPoolExecutor, ProcessPoolExecutorjobstores &＃61; {&＃39;mongo&＃39;: {&＃39;type&＃39;: &＃39;mongodb&＃39;},&＃39;default&＃39;: SQLAlchemyJobStore(url&＃61;&＃39;sqlite:///jobs.sqlite&＃39;)
}executors &＃61; {&＃39;default&＃39;: {&＃39;type&＃39;: &＃39;threadpool&＃39;, &＃39;max_workers&＃39;: 20},&＃39;processpool&＃39;: ProcessPoolExecutor(max_workers&＃61;5)
}
# executors &＃61; {
# &＃39;default&＃39;: ThreadPoolExecutor(20),
# &＃39;processpool&＃39;: ProcessPoolExecutor(5)
# }job_defaults &＃61; {&＃39;coalesce&＃39;: False,&＃39;max_instances&＃39;: 3
}
scheduler &＃61; BackgroundScheduler()# .. do something else here, maybe add jobs etc.scheduler.configure(jobstores&＃61;jobstores, executors&＃61;executors, job_defaults&＃61;job_defaults, timezone&＃61;utc)