前言

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

一、APScheduler介紹

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

APScheduler提供了多種不同的調(diào)度器，方便開(kāi)發(fā)者根據(jù)自己的實(shí)際需要進(jìn)行使用；同時(shí)也提供了不同的存儲(chǔ)機(jī)制，可以方便與Redis，數(shù)據(jù)庫(kù)等第三方的外部持久化機(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í)非常簡(jiǎn)單，基于pip直接安裝即可：　

pip install APScheduler 

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

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

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

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

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

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

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

　　處理作業(yè)的運(yùn)行，他們通常通過(guò)在作業(yè)中提交制定的可調(diào)用對(duì)象到一個(gè)線程或者進(jìn)城池來(lái)進(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è)。

這里簡(jiǎn)單列一下常用的若干調(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ì)算模型來(lái)動(dòng)態(tài)選擇調(diào)度器。

三、APScheduler的job管理

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

3.1 Job的新增

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

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

代碼示例：

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

基于修飾器scheduled_job來(lái)動(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來(lái)動(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() 來(lái)完成，它會(huì)返回所有的job實(shí)例。或者使用 print_jobs() 來(lái)輸出所有格式化的作業(yè)列表。

3.5. 修改作業(yè) job

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

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

另外我們也可以通過(guò)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的代碼示例

這里使用裝飾器來(lái)展示一個(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") 

代碼說(shuō)明：

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

另外一個(gè)調(diào)度器是模仿cron來(lái)執(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è)異常問(wèn)題的思考

在執(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ái)分析，是沒(méi)有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) 

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

六、總結(jié)

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

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