官網(wǎng)地址

講解1

講解2

1 簡介

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

2 安裝

#使用 pip 包管理工具安裝 APScheduler 是最方便快捷的。pip install APScheduler # 如果出現(xiàn)因下載失敗導(dǎo)致安裝不上的情況，建議使用代理pip --proxy http://代理ip:端口 install APScheduler

3 使用步驟

APScheduler 使用起來還算是比較簡單。運行一個調(diào)度任務(wù)只需要以下三部曲。

新建一個 schedulers (調(diào)度器) 。
添加一個調(diào)度任務(wù)(job stores)。
運行調(diào)度任務(wù)。

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

import datetime import time from apscheduler.schedulers.background import BackgroundSchedulerdef timedTask():print(datetime.datetime.utcnow().strftime("%Y-%m-%d %H:%M:%S.%f")[:-3])if __name__ == '__main__':# 創(chuàng)建后臺執(zhí)行的 schedulersscheduler = BackgroundScheduler() # 添加調(diào)度任務(wù)# 調(diào)度方法為 timedTask，觸發(fā)器選擇 interval(間隔性)，間隔時長為 2 秒scheduler.add_job(timedTask, 'interval', seconds=2)# 啟動調(diào)度任務(wù)scheduler.start()while True:print(time.time())time.sleep(5)

4 基礎(chǔ)組件

APScheduler 有四種組件，分別是：調(diào)度器(scheduler)，作業(yè)存儲(job store)，觸發(fā)器(trigger)，執(zhí)行器(executor)。

• schedulers（調(diào)度器）
  它是任務(wù)調(diào)度器，屬于控制器角色。它配置作業(yè)存儲器和執(zhí)行器可以在調(diào)度器中完成，例如添加、修改和移除作業(yè)。

• triggers（觸發(fā)器）
  描述調(diào)度任務(wù)被觸發(fā)的條件。不過觸發(fā)器完全是無狀態(tài)的。

• job stores（作業(yè)存儲器）
  任務(wù)持久化倉庫，默認(rèn)保存任務(wù)在內(nèi)存中，也可將任務(wù)保存都各種數(shù)據(jù)庫中，任務(wù)中的數(shù)據(jù)序列化后保存到持久化數(shù)據(jù)庫，從數(shù)據(jù)庫加載后又反序列化。

• executors（執(zhí)行器）
  負(fù)責(zé)處理作業(yè)的運行，它們通常通過在作業(yè)中提交指定的可調(diào)用對象到一個線程或者進(jìn)城池來進(jìn)行。當(dāng)作業(yè)完成時，執(zhí)行器將會通知調(diào)度器。

• 4.1 schedulers（調(diào)度器）

  我個人覺得 APScheduler 非常好用的原因。它提供 7 種調(diào)度器，能夠滿足我們各種場景的需要。例如：后臺執(zhí)行某個操作，異步執(zhí)行操作等。調(diào)度器分別是：

  • BlockingScheduler : 調(diào)度器在當(dāng)前進(jìn)程的主線程中運行，也就是會阻塞當(dāng)前線程。
  • BackgroundScheduler : 調(diào)度器在后臺線程中運行，不會阻塞當(dāng)前線程。
  • AsyncIOScheduler : 結(jié)合 asyncio 模塊（一個異步框架）一起使用。
  • GeventScheduler : 程序中使用 gevent（高性能的Python并發(fā)框架）作為IO模型，和 GeventExecutor 配合使用。
  • TornadoScheduler : 程序中使用 Tornado（一個web框架）的IO模型，用 ioloop.add_timeout 完成定時喚醒。
  • TwistedScheduler : 配合 TwistedExecutor，用 reactor.callLater 完成定時喚醒。
  • QtScheduler : 你的應(yīng)用是一個 Qt 應(yīng)用，需使用QTimer完成定時喚醒。

  4.2 triggers（觸發(fā)器）

  APScheduler 有三種內(nèi)建的 trigger:
  1）date 觸發(fā)器
  date 是最基本的一種調(diào)度，作業(yè)任務(wù)只會執(zhí)行一次。它表示特定的時間點觸發(fā)。它的參數(shù)如下：

  參數(shù)說明

  run_date (datetime 或 str)	作業(yè)的運行日期或時間
  timezone (datetime.tzinfo 或 str)	指定時區(qū)

  #date 觸發(fā)器使用示例如下: from datetime import datetime from datetime import date from apscheduler.schedulers.background import BackgroundSchedulerdef 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, <span class="hljs-number">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 觸發(fā)器
  固定時間間隔觸發(fā)。interval 間隔調(diào)度，參數(shù)如下：

  參數(shù)說明

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

  interval 觸發(fā)器使用示例如下：

  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()# 每隔兩分鐘執(zhí)行一次 job_func 方法 scheduler .add_job(job_func, 'interval', minutes= 2)# 在 2017-12-13 14:00:01 ~ 2017-12-13 14:00:10 之間, 每隔兩分鐘執(zhí)行一次 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 觸發(fā)器
  在特定時間周期性地觸發(fā)，和Linux crontab格式兼容。它是功能最強大的觸發(fā)器。
  我們先了解 cron 參數(shù)：

  參數(shù)說明

  year (int 或 str)	年，4位數(shù)字
  month (int 或 str)	月 (范圍1-12)
  day (int 或 str)	日 (范圍1-31)
  week (int 或 str)	周 (范圍1-53)
  day_of_week (int 或 str)	周內(nèi)第幾天或者星期幾 (范圍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)	最晚結(jié)束時間(包含)
  timezone (datetime.tzinfo 或str)	指定時區(qū)

  這些參數(shù)是支持算數(shù)表達(dá)式，取值格式有如下：
  

  cron 觸發(fā)器使用示例如下：

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

  4.3 作業(yè)存儲(job store)

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

  這個兩種辦法的區(qū)別是：第一種方法返回一個 apscheduler.job.Job 的實例，可以用來改變或者移除 job。第二種方法只適用于應(yīng)用運行期間不會改變的 job。

  第二種添加任務(wù)方式的例子：

  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() 是根據(jù) job 的 id 來移除，所以要在 job 創(chuàng)建的時候指定一個 id。
  job.remove() 則是對 job 執(zhí)行 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() 方法能夠獲取當(dāng)前調(diào)度器中的所有 job 的列表

  修改 job
  如果你因計劃改變要對 job 進(jìn)行修改，可以使用Job.modify() 或者 modify_job()方法來修改 job 的屬性。但是值得注意的是，job 的 id 是無法被修改的。

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

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

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

  4.4 執(zhí)行器(executor)

  執(zhí)行器顧名思義是執(zhí)行調(diào)度任務(wù)的模塊。最常用的 executor 有兩種：ProcessPoolExecutor 和 ThreadPoolExecutor

  下面是顯式設(shè)置 job store(使用mongo存儲)和 executor 的代碼的示例。
  

  #注：本代碼來源于網(wǎng)絡(luò) 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, ProcessPoolExecutordef 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()

總結(jié)

以上是生活随笔為你收集整理的python 定时任务 Apscheduler的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯，歡迎將生活随笔推薦給好友。