APScheduler

1 簡介

APScheduler的全稱是Advanced Python Scheduler。它是一個輕量級的 Python 定時任務調度框架。APScheduler 支持三種調度任務：固定時間間隔，固定時間點（日期），Linux 下的 Crontab 命令。同時，它還支持異步執行、後臺執行調度任務。

2 安裝

使用 pip 包管理工具安裝 APScheduler 是最方便快捷的。

pip install APScheduler
# 若是出現因下載失敗致使安裝不上的狀況，建議使用代理 pip --proxy http://代理ip:端口 install APScheduler 

3 使用步驟

APScheduler 使用起來還算是比較簡單。運行一個調度任務只須要如下三部曲。

新建一個 schedulers (調度器) 。
添加一個調度任務(job stores)。
運行調度任務。

下面是執行每 2 秒報時的簡單示例代碼：

import datetime import time from apscheduler.schedulers.background import BackgroundScheduler def timedTask(): print(datetime.datetime.utcnow().strftime("%Y-%m-%d %H:%M:%S.%f")[:-3]) if __name__ == '__main__': # 建立後臺執行的 schedulers scheduler = BackgroundScheduler() # 添加調度任務 # 調度方法爲 timedTask，觸發器選擇 interval(間隔性)，間隔時長爲 2 秒 scheduler.add_job(timedTask, 'interval', seconds=2) # 啓動調度任務 scheduler.start() while True: print(time.time()) time.sleep(5) 

4 基礎組件

APScheduler 有四種組件，分別是：調度器(scheduler)，做業存儲(job store)，觸發器(trigger)，執行器(executor)。

• schedulers（調度器）
  它是任務調度器，屬於控制器角色。它配置做業存儲器和執行器能夠在調度器中完成，例如添加、修改和移除做業。

• triggers（觸發器）
  描述調度任務被觸發的條件。不過觸發器徹底是無狀態的。

• job stores（做業存儲器）
  任務持久化倉庫，默認保存任務在內存中，也可將任務保存都各類數據庫中，任務中的數據序列化後保存到持久化數據庫，從數據庫加載後又反序列化。

• executors（執行器）
  負責處理做業的運行，它們一般經過在做業中提交指定的可調用對象到一個線程或者進城池來進行。看成業完成時，執行器將會通知調度器。

4.1 schedulers（調度器）

我我的以爲 APScheduler 很是好用的緣由。它提供 7 種調度器，可以知足咱們各類場景的須要。例如：後臺執行某個操做，異步執行操做等。調度器分別是：

• BlockingScheduler : 調度器在當前進程的主線程中運行，也就是會阻塞當前線程。
• BackgroundScheduler : 調度器在後臺線程中運行，不會阻塞當前線程。
• AsyncIOScheduler : 結合 asyncio 模塊（一個異步框架）一塊兒使用。
• GeventScheduler : 程序中使用 gevent（高性能的Python併發框架）做爲IO模型，和 GeventExecutor 配合使用。
• TornadoScheduler : 程序中使用 Tornado（一個web框架）的IO模型，用 ioloop.add_timeout 完成定時喚醒。
• TwistedScheduler : 配合 TwistedExecutor，用 reactor.callLater 完成定時喚醒。
• QtScheduler : 你的應用是一個 Qt 應用，需使用QTimer完成定時喚醒。

4.2 triggers（觸發器）

APScheduler 有三種內建的 trigger:
1）date 觸發器
date 是最基本的一種調度，做業任務只會執行一次。它表示特定的時間點觸發。它的參數以下：

參數	說明
run_date (datetime 或 str)	做業的運行日期或時間
timezone (datetime.tzinfo 或 str)	指定時區

date 觸發器使用示例以下：

from datetime import datetime from datetime import date from apscheduler.schedulers.background import BackgroundScheduler def job_func(text): print(text) scheduler = BackgroundScheduler() # 在 2017-12-13 時刻運行一次 job_func 方法 scheduler .add_job(job_func, 'date', run_date=date(2017, 12, 13), args=['text']) # 在 2017-12-13 14:00:00 時刻運行一次 job_func 方法 scheduler .add_job(job_func, 'date', run_date=datetime(2017, 12, 13, 14, 0, 0), args=['text']) # 在 2017-12-13 14:00:01 時刻運行一次 job_func 方法 scheduler .add_job(job_func, 'date', run_date='2017-12-13 14:00:01', args=['text']) scheduler.start() 

2）interval 觸發器
固定時間間隔觸發。interval 間隔調度，參數以下：

參數	說明
weeks (int)	間隔幾周
days (int)	間隔幾天
hours (int)	間隔幾小時
minutes (int)	間隔幾分鐘
seconds (int)	間隔多少秒
start_date (datetime 或 str)	開始日期
end_date (datetime 或 str)	結束日期
timezone (datetime.tzinfo 或str)	時區

interval 觸發器使用示例以下：

import datetime from apscheduler.schedulers.background import BackgroundScheduler def job_func(text): print(datetime.datetime.utcnow().strftime("%Y-%m-%d %H:%M:%S.%f")[:-3]) scheduler = BackgroundScheduler() # 每隔兩分鐘執行一次 job_func 方法 scheduler .add_job(job_func, 'interval', minutes=2) # 在 2017-12-13 14:00:01 ~ 2017-12-13 14:00:10 之間, 每隔兩分鐘執行一次 job_func 方法 scheduler .add_job(job_func, 'interval', minutes=2, start_date='2017-12-13 14:00:01' , end_date='2017-12-13 14:00:10') scheduler.start() 

3）cron 觸發器
在特定時間週期性地觸發，和Linux crontab格式兼容。它是功能最強大的觸發器。
咱們先了解 cron 參數：

參數	說明
year (int 或 str)	年，4位數字
month (int 或 str)	月 (範圍1-12)
day (int 或 str)	日 (範圍1-31
week (int 或 str)	周 (範圍1-53)
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)	指定時區

這些參數是支持算數表達式，取值格式有以下：

 

image.png

cron 觸發器使用示例以下：

import datetime from apscheduler.schedulers.background import BackgroundScheduler def job_func(text): print("當前時間：", datetime.datetime.utcnow().strftime("%Y-%m-%d %H:%M:%S.%f")[:-3]) scheduler = BackgroundScheduler() # 在每一年 1-三、7-9 月份中的每一個星期1、二中的 00:00, 01:00, 02:00 和 03:00 執行 job_func 任務 scheduler .add_job(job_func, 'cron', month='1-3,7-9',day='0, tue', hour='0-3') scheduler.start() 

4.3 做業存儲(job store)

該組件是對調度任務的管理。
1）添加 job
有兩種添加方法，其中一種上述代碼用到的 add_job()， 另外一種則是scheduled_job()修飾器來修飾函數。

這個兩種辦法的區別是：第一種方法返回一個 apscheduler.job.Job 的實例，能夠用來改變或者移除 job。第二種方法只適用於應用運行期間不會改變的 job。

第二種添加任務方式的例子：

import datetime from apscheduler.schedulers.background import BackgroundScheduler @scheduler.scheduled_job(job_func, 'interval', minutes=2) def job_func(text): print(datetime.datetime.utcnow().strftime("%Y-%m-%d %H:%M:%S.%f")[:-3]) scheduler = BackgroundScheduler() scheduler.start() 

2）移除 job
移除 job 也有兩種方法：remove_job() 和 job.remove()。
remove_job() 是根據 job 的 id 來移除，因此要在 job 建立的時候指定一個 id。
job.remove() 則是對 job 執行 remove 方法便可

scheduler.add_job(job_func, 'interval', minutes=2, id='job_one') scheduler.remove_job(job_one) job = add_job(job_func, 'interval', minutes=2, id='job_one') job.remvoe() 

3）獲取 job 列表
經過 scheduler.get_jobs() 方法可以獲取當前調度器中的全部 job 的列表

修改 job
若是你因計劃改變要對 job 進行修改，可使用Job.modify() 或者 modify_job()方法來修改 job 的屬性。可是值得注意的是，job 的 id 是沒法被修改的。

scheduler.add_job(job_func, 'interval', minutes=2, id='job_one') scheduler.start() # 將觸發時間間隔修改爲 5分鐘 scheduler.modify_job('job_one', minutes=5) job = scheduler.add_job(job_func, 'interval', minutes=2) # 將觸發時間間隔修改爲 5分鐘 job.modify(minutes=5) 

5）關閉 job
默認狀況下調度器會等待全部正在運行的做業完成後，關閉全部的調度器和做業存儲。若是你不想等待，能夠將 wait 選項設置爲 False。

scheduler.shutdown()
scheduler.shutdown(wait=false) 

4.4 執行器(executor)

執行器顧名思義是執行調度任務的模塊。最經常使用的 executor 有兩種：ProcessPoolExecutor 和 ThreadPoolExecutor

下面是顯式設置 job store(使用mongo存儲)和 executor 的代碼的示例。
注：本代碼來源於網絡

from pymongo import MongoClient from apscheduler.schedulers.blocking import BlockingScheduler from apscheduler.jobstores.mongodb import MongoDBJobStore from apscheduler.jobstores.memory import MemoryJobStore from apscheduler.executors.pool import ThreadPoolExecutor, ProcessPoolExecutor def my_job(): print 'hello world' host = '127.0.0.1' port = 27017 client = MongoClient(host, port) jobstores = { 'mongo': MongoDBJobStore(collection='job', database='test', client=client), 'default': MemoryJobStore() } executors = { 'default': ThreadPoolExecutor(10), 'processpool': ProcessPoolExecutor(3) } job_defaults = { 'coalesce': False, 'max_instances': 3 } scheduler = BlockingScheduler(jobstores=jobstores, executors=executors, job_defaults=job_defaults) scheduler.add_job(my_job, 'interval', seconds=5) try: scheduler.start() except SystemExit: client.close()

做者：_周小董連接：https://www.jianshu.com/p/4d531c50ff9b來源：簡書