前言

大家應(yīng)該都知道在編程語言中，定時(shí)任務(wù)是常用的一種調(diào)度形式，在Python中也涌現(xiàn)了非常多的調(diào)度模塊，本文將簡要介紹APScheduler的基本使用方法。

一、APScheduler介紹

APScheduler是基于Quartz的一個(gè)python定時(shí)任務(wù)框架，實(shí)現(xiàn)了Quartz的所有功能，使用起來十分方便。提供了基于日期、固定時(shí)間間隔以及crontab類型的任務(wù)，并且可以持久化任務(wù)。

APScheduler提供了多種不同的調(diào)度器，方便開發(fā)者根據(jù)自己的實(shí)際需要進(jìn)行使用；同時(shí)也提供了不同的存儲(chǔ)機(jī)制，可以方便與Redis，數(shù)據(jù)庫等第三方的外部持久化機(jī)制進(jìn)行協(xié)同工作，總之功能非常強(qiáng)大和易用。

在Python的世界中，另外一個(gè)齊名的調(diào)度模塊是Celery，功能也非常的強(qiáng)大，號(hào)稱分布式的調(diào)度器，感興趣的讀者可以自行進(jìn)行研究。

官網(wǎng)文檔地址：http://apscheduler.readthedocs.io/en/latest/

安裝包位置: https://pypi.python.org/pypi/APScheduler/

在系統(tǒng)中，如何進(jìn)行安裝呢？其實(shí)非常簡單，基于pip直接安裝即可：　

pip install APScheduler 

二、APScheduler的主要的調(diào)度類

在APScheduler中有以下幾個(gè)非常重要的概念，需要大家理解：   

1、觸發(fā)器(trigger)

　　包含調(diào)度邏輯，每一個(gè)作業(yè)有它自己的觸發(fā)器，用于決定接下來哪一個(gè)作業(yè)會(huì)運(yùn)行，根據(jù)trigger中定義的時(shí)間點(diǎn)，頻率，時(shí)間區(qū)間等等參數(shù)設(shè)置。除了他們自己初始配置以外，觸發(fā)器完全是無狀態(tài)的。

2、作業(yè)存儲(chǔ)(job store)

　　存儲(chǔ)被調(diào)度的作業(yè)，默認(rèn)的作業(yè)存儲(chǔ)是簡單地把作業(yè)保存在內(nèi)存中，其他的作業(yè)存儲(chǔ)是將作業(yè)保存在數(shù)據(jù)庫中。一個(gè)作業(yè)的數(shù)據(jù)講在保存在持久化作業(yè)存儲(chǔ)時(shí)被序列化，并在加載時(shí)被反序列化。調(diào)度器不能分享同一個(gè)作業(yè)存儲(chǔ)。job store支持主流的存儲(chǔ)機(jī)制：redis, mongodb, 關(guān)系型數(shù)據(jù)庫,　內(nèi)存等等

3、執(zhí)行器(executor)

　　處理作業(yè)的運(yùn)行，他們通常通過在作業(yè)中提交制定的可調(diào)用對(duì)象到一個(gè)線程或者進(jìn)城池來進(jìn)行。當(dāng)作業(yè)完成時(shí)，執(zhí)行器將會(huì)通知調(diào)度器?；诔鼗牟僮?，可以針對(duì)不同類型的作業(yè)任務(wù)，更為高效地使用cpu的計(jì)算資源。
調(diào)度器(scheduler)

　　通常在應(yīng)用只有一個(gè)調(diào)度器，調(diào)度器提供了處理這些的合適的接口。配置作業(yè)存儲(chǔ)和執(zhí)行器可以在調(diào)度器中完成，例如添加、修改和移除作業(yè)。

這里簡單列一下常用的若干調(diào)度器：

• BlockingScheduler：僅可用在當(dāng)前你的進(jìn)程之內(nèi)，與當(dāng)前的進(jìn)行共享計(jì)算資源
• BackgroundScheduler:　在后臺(tái)運(yùn)行調(diào)度，不影響當(dāng)前的系統(tǒng)計(jì)算運(yùn)行
• AsyncIOScheduler:　如果當(dāng)前系統(tǒng)中使用了async module，則需要使用異步的調(diào)度器
• GeventScheduler:　如果使用了gevent，則需要使用該調(diào)度
• TornadoScheduler:　如果使用了Tornado, 則使用當(dāng)前的調(diào)度器
• TwistedScheduler:Twister應(yīng)用的調(diào)度器
• QtScheduler:　Qt的調(diào)度器
  

由此可知，在APscheduler的調(diào)度器中，是與底層的實(shí)現(xiàn)機(jī)制緊密相關(guān)的，需要依據(jù)當(dāng)前的計(jì)算模型來動(dòng)態(tài)選擇調(diào)度器。

三、APScheduler的job管理

Job是APScheduler中的核心，其承接目前需要執(zhí)行的工作和任務(wù)，其可以在系統(tǒng)運(yùn)行過程中動(dòng)態(tài)地進(jìn)行增加／修改／刪除／查詢等操作。

3.1 Job的新增

共有兩種方式進(jìn)行新增job的操作：

基于add_job來動(dòng)態(tài)增加

代碼示例：

sched.add_job(job_function, 'cron', day_of_week='mon-fri', hour='0-9', minute="*", second="*/4") 

基于修飾器scheduled_job來動(dòng)態(tài)裝飾job的實(shí)際函數(shù)

代碼示例：

@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!") 

3.2 移除作業(yè)

job = scheduler.add_job(myfunc, 'interval', minutes=2) 
job.remove() 
Same, using an explicit job ID: 
 
scheduler.add_job(myfunc, 'interval', minutes=2, id='my_job_id') 
scheduler.remove_job('my_job_id') 

基于job id來動(dòng)態(tài)移除特定的job.

3.3 暫停和恢復(fù)作業(yè)

暫停作業(yè):

  – apscheduler.job.Job.pause()

  – apscheduler.schedulers.base.BaseScheduler.pause_job()

恢復(fù)作業(yè):

  – apscheduler.job.Job.resume()

  – apscheduler.schedulers.base.BaseScheduler.resume_job()

3.4. 獲得job列表

獲得調(diào)度作業(yè)的列表，可以使用 get_jobs() 來完成，它會(huì)返回所有的job實(shí)例。或者使用 print_jobs() 來輸出所有格式化的作業(yè)列表。

3.5. 修改作業(yè) job

可以通過apscheduler.job.Job.modify() or　modify_job()來動(dòng)態(tài)修改job的屬性信息，除了job id無法修改之外，都是可以修改的。

job.modify(max_instances=6, name='Alternate name') 

另外我們也可以通過apscheduler.job.Job.reschedule() or reschedule_job()動(dòng)態(tài)重新設(shè)置trigger，示例如下：

scheduler.reschedule_job('my_job_id', trigger='cron', minute='*/5') 

 3.6. 關(guān)閉調(diào)度器

默認(rèn)情況下調(diào)度器會(huì)等待所有正在運(yùn)行的作業(yè)完成后，關(guān)閉所有的調(diào)度器和作業(yè)存儲(chǔ)。如果你不想等待，可以將wait選項(xiàng)設(shè)置為False。

scheduler.shutdown() 
scheduler.shutdown(wait=False) 

四、 APScheduler的代碼示例

這里使用裝飾器來展示一個(gè)調(diào)度的使用：

from apscheduler.schedulers.blocking import BlockingScheduler 
 
sched = BlockingScheduler() 
 
@sched.scheduled_job('interval', seconds=3) 
def timed_job(): 
 print('This job is run every three minutes.') 
 
@sched.scheduled_job('cron', day_of_week='mon-fri', hour='0-9', minute='30-59', second='*/3') 
def scheduled_job(): 
 print('This job is run every weekday at 5pm.') 
 
print('before the start funciton') 
sched.start() 
print("let us figure out the situation") 

代碼說明：

在這段代碼中，使用了當(dāng)前進(jìn)程中共享計(jì)算資源的BlockingScheduler，共使用了２個(gè)調(diào)度器，其中一個(gè)是間隔３秒的執(zhí)行。

另外一個(gè)調(diào)度器是模仿cron來執(zhí)行的，在周一到周五其間，每天的０點(diǎn)到９點(diǎn)直接，在３０分到５９分之間執(zhí)行，執(zhí)行頻次為３秒。

基于正常代碼的示例如下：

from apscheduler.schedulers.background import BackgroundScheduler 
from apscheduler.schedulers.blocking import BlockingScheduler 
 
import datetime 
import time 
import logging 
 
def job_function(): 
 print "Hello World" + " " + str(datetime.datetime.now()) 
 
if __name__ == '__main__': 
 log = logging.getLogger('apscheduler.executors.default') 
 log.setLevel(logging.INFO) # DEBUG 
 
 fmt = logging.Formatter('%(levelname)s:%(name)s:%(message)s') 
 h = logging.StreamHandler() 
 h.setFormatter(fmt) 
 log.addHandler(h) 
 
 print('start to do it') 
 
 sched = BlockingScheduler() 
 
 # Schedules job_function to be run on the third Friday 
 # of June, July, August, November and December at 00:00, 01:00, 02:00 and 03:00 
 sched.add_job(job_function, 'cron', day_of_week='mon-fri', hour='0-9', minute="*", second="*/4") 
 
 sched.start() 

五、某個(gè)異常問題的思考

在執(zhí)行以下代碼之時(shí)候，定時(shí)任務(wù)一直未能正常生效：

from apscheduler.schedulers.background import BackgroundScheduler 
from apscheduler.schedulers.blocking import BlockingScheduler 
 
import datetime 
import time 
 
def job_function(): 
 print "Hello World" + " " + str(datetime.datetime.now()) 
 
if __name__ == '__main__': 
 print('start to do it') 
 sched = BlockingScheduler() 
 sched.add_job(job_function, 'cron', day_of_week='mon-fri', hour='0-9', minute="*", second="*/4") 
 sched.start() 

代碼報(bào)錯(cuò)的錯(cuò)誤信息為：

No handlers could be found for logger “apscheduler.scheduler” 

從字面意思來分析，是沒有l(wèi)ogging模塊的logger存在，故需要添加上去即可。

新增對(duì)應(yīng)的logging信息即可：

import logging 

log = logging.getLogger('apscheduler.executors.default') 
 log.setLevel(logging.INFO) # DEBUG 
 
 fmt = logging.Formatter('%(levelname)s:%(name)s:%(message)s') 
 h = logging.StreamHandler() 
 h.setFormatter(fmt) 
 log.addHandler(h) 

后來筆者重新做了一次執(zhí)行，即使移除掉logging的內(nèi)容，依然可以正常執(zhí)行，故可以推測為需要?jiǎng)討B(tài)引入一次依賴包logging即可。

六、總結(jié)

APScheduler是一個(gè)非常強(qiáng)大易用的類庫，為了我們簡單快捷的解決問題提供了很多的工具，并且提供了很多靈活的擴(kuò)展點(diǎn)，只要你添加若干的web頁面，就可以創(chuàng)建一個(gè)強(qiáng)大的任務(wù)調(diào)度系統(tǒng)，不是嗎？

好了，以上就是這篇文章的全部內(nèi)容了，希望本文的內(nèi)容對(duì)大家的學(xué)習(xí)或者工作能帶來一定的幫助，如果有疑問大家可以留言交流，謝謝大家對(duì)腳本之家的支持。

最新評(píng)論