Python定時任務APScheduler的例項例項詳解
APScheduler 支援三種排程任務:固定時間間隔,固定時間點(日期),Linux 下的 Crontab 命令。同時,它還支援非同步執行、後臺執行排程任務。
一、基本架構
- 觸發器 triggers:設定觸發任務的條件
- 描述一個任務何時被觸發,按日期或按時間間隔或按 cronjob 表示式三種方式觸發
- 任務儲存器 job stores:存放任務,可以放記憶體(預設)或資料庫
- 注:排程器之間不能共享任務儲存器
- 執行器 executors:用於執行任務,可設定執行模式
- 將指定的作業提交到執行緒池或者程序池中執行,任務完成通知排程器觸發相應的事件。
- 排程器 schedulers:將上方三個元件作為引數,建立排程器例項執行。
協調三個元件的執行。
二、排程器元件(schedulers)
- BlockingScheduler 阻塞式排程器
排程程式是程序中唯一執行的程序,呼叫start函式會阻塞當前執行緒,不能立即返回。
1 2 3 4 5 6 7 |
from apscheduler.schedulers.blocking import BlockingScheduler
import time
scheduler = BlockingScheduler()
def job1():
print "%s: 執行任務" % time.asctime()
scheduler.add_job(job1, 'interval' , seconds = 3 )
scheduler.start()
|
- BackgroundScheduler 後臺排程器
當前執行緒不會阻塞,排程器後臺執行
1 2 3 4 5 6 7 8 |
from apscheduler.schedulers.background import BackgroundScheduler
import time
scheduler = BackgroundScheduler()
def job1():
print "%s: 執行任務" % time.asctime()
scheduler.add_job(job1, 'interval' , seconds = 3 )
scheduler.start()
time.sleep( 10 )
|
注:10秒執行完後,程式結束。
- AsyncIOScheduler AsyncIO排程器
適用於使用了asyncio的情況
1 2 3 4 5 6 7 8 |
from apscheduler.schedulers.asyncio import AsyncIOScheduler
import asyncio
...
...
try :
asyncio.get_event_loop().run_forever()
except (KeyboardInterrupt, SystemExit):
pass
|
- GeventScheduler Gevent排程器
使用了Gevent的情況
1 2 3 4 5 6 7 8 |
from apscheduler.schedulers.gevent import GeventScheduler
...
...
g = scheduler.start()
try :
g.join()
except (KeyboardInterrupt, SystemExit):
pass
|
- TornadoScheduler Tornado排程器
適用於構建Tornado應用
- TwistedScheduler Twisted排程器
適用於構建Twisted應用
- QtScheduler Qt排程器
適用於構建Qt應用
三、觸發器元件(trigger)
date :只在某個時間點執行一次,具體日期
run_date(datetime|str)
1 2 |
scheduler.add_job(my_job, 'date' , run_date = datetime( 2019 , 7 , 12 , 15 , 30 , 5 ), args = [])
scheduler.add_job(my_job, 'date' , run_date = "2019-07-12" , args = [])
|
timezone 指定時區
interval :每隔一段時間允許一次,時間間隔
1 2 3 |
weeks = 0 | days = 0 | hours = 0 | minutes = 0 | seconds = 0 , start_date = None , end_date = None , timezone = None
scheduler.add_job(my_job, 'interval' , hours = 2 )
scheduler.add_job(my_job, 'interval' , hours = 2 , start_date = '2017-9-8 21:30:00' , end_date = ' 2018 - 06 - 15 21 : 30 : 00 )
|
cron :任務的執行週期
1 |
(year = None , month = None , day = None , week = None , day_of_week = None , hour = None , minute = None , second = None , start_date = None , end_date = None , timezone = None )
|
除了week和 day_of_week,它們的預設值是*
例如day=1, minute=20 ,這就等於year='*', month='*', day=1, week='*', day_of_week='*', hour='*',
minute=20, second=0 ,工作將在每個月的第一天以每小時20分鐘的時間執行
表示式型別
表示式 | 引數型別 | 描述 |
---|---|---|
* | 所有 | 萬用字元。例: minutes=*即每分鐘觸發 |
*/a | 所有 | 可被a整除的萬用字元。 |
a-b | 所有 | 範圍a-b觸發 |
a-b/c | 所有 | 範圍a-b,且可被c整除時觸發 |
xth y | 日 | 第幾個星期幾觸發。x為第幾個,y為星期幾 |
last x | 日 | 一個月中,最後個星期幾觸發 |
last | 日 | 一個月最後一天觸發 |
x,y,z | 所有 | 組合表示式,可以組合確定值或上方的表示式 |
注:當設定的時間間隔小於,任務的執行時間,執行緒會阻塞住,等待執行完了才能執行下一個任務,可以設定max_instance 指定一個任務同一時刻有多少個例項在執行,預設為1
四、配置排程器
執行緒池執行器預設為10,記憶體任務儲存器為memoryjobstore ,如果想自己配置的話可以執行以下操作
需求:
- 兩個任務儲存器分別搭配兩個執行器;同時,還要修改任務的預設引數;最後還要改時區
- 名稱為“mongo”的MongoDBJobStore
- 名稱為“default”的SQLAlchemyJobStore
- 名稱為“ThreadPoolExecutor ”的ThreadPoolExecutor ,最大執行緒20個
- 名稱“processpool”的ProcessPoolExecutor ,最大程序5個
- UTC時間作為排程器的時區
- 預設為新任務關閉合並模式 ()
- 設定新任務的預設最大例項數為3
方法一:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 |
from pytz import utc
from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.jobstores.mongodb import MongoDBJobStore
from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore
from apscheduler.executors.pool import ThreadPoolExecutor, ProcessPoolExecutor
jobstores = {
'mongo' : MongoDBJobStore(),
'default' : SQLAlchemyJobStore(url = 'sqlite:///jobs.sqlite' )
}
executors = {
'default' : ThreadPoolExecutor( 20 ),
'processpool' : ProcessPoolExecutor( 5 )
}
job_defaults = {
'coalesce' : False ,
'max_instances' : 3
}
scheduler = BackgroundScheduler(jobstores = jobstores, executors = executors, job_defaults = job_defaults, timezone = utc)
|
方法二:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 |
from apscheduler.schedulers.background import BackgroundScheduler
# The "apscheduler." prefix is hard coded
scheduler = BackgroundScheduler({
'apscheduler.jobstores.mongo' : {
'type' : 'mongodb'
},
'apscheduler.jobstores.default' : {
'type' : 'sqlalchemy' ,
'url' : 'sqlite:///jobs.sqlite'
},
'apscheduler.executors.default' : {
'class' : 'apscheduler.executors.pool:ThreadPoolExecutor' ,
'max_workers' : '20'
},
'apscheduler.executors.processpool' : {
'type' : 'processpool' ,
'max_workers' : '5'
},
'apscheduler.job_defaults.coalesce' : 'false' ,
'apscheduler.job_defaults.max_instances' : '3' ,
'apscheduler.timezone' : 'UTC' ,
})
|
方法三:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 |
from pytz import utc
from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore
from apscheduler.executors.pool import ProcessPoolExecutor
jobstores = {
'mongo' : { 'type' : 'mongodb' },
'default' : SQLAlchemyJobStore(url = 'sqlite:///jobs.sqlite' )
}
executors = {
'default' : { 'type' : 'threadpool' , 'max_workers' : 20 },
'processpool' : ProcessPoolExecutor(max_workers = 5 )
}
job_defaults = {
'coalesce' : False ,
'max_instances' : 3
}
scheduler = BackgroundScheduler()
# ..這裡可以新增任務
scheduler.configure(jobstores = jobstores, executors = executors, job_defaults = job_defaults, timezone = utc)
|
五、啟動排程器
除了BlockingScheduler 外,其他非阻塞的排程器都會立即返回,執行之後的程式碼。
BlockingScheduler 需要將執行的程式碼放在start()之前
1.新增任務
1.呼叫add_job() #可以傳參max_instance,同一任務的執行例項個數
當有任務中途中斷,後面恢復後,有N個任務沒有執行
coalesce:true ,恢復的任務會執行一次
coalesce:false,恢復後的任務會執行N次配合misfire_grace_time使用
misfire_grace_time設定時間差值,由於某些原因沒有執行,再次提交時,大於設定的時間,例項不會執行。
2.裝飾器scheduled_job()
立即執行可以不設定trigger引數
2.移除任務
1 2 3 4 5 6 |
# 根據任務例項刪除
job = scheduler.add_job(myfunc, 'interval' , minutes = 2 )
job.remove()
# 根據任務id刪除
scheduler.add_job(myfunc, 'interval' , minutes = 2 , id = 'my_job_id' )
scheduler.remove_job( 'my_job_id' )
|
3.暫停任務
1 2 3 4 5 6 7 8 |
job = scheduler.add_job(myfunc, 'interval' , minutes = 2 )
# 根據任務例項
job.pause() #暫停
job.resume() #繼續
# 根據任務id暫停
scheduler.add_job(myfunc, 'interval' , minutes = 2 , id = 'my_job_id' )
scheduler.pause_job( 'my_job_id' )
scheduler.resume_job( 'my_job_id' )
|
4.排程器操作
1 2 |
scheduler.start() #開啟
scheduler.shotdown(wait = True | False ) #關閉 False 無論任務是否執行,強制關閉
|
異常捕獲
1 2 3 4 |
# 可以新增apscheduler日誌至DEBUG級別,這樣就能捕獲異常資訊
import logging
logging.basicConfig()
logging.getLogger( 'apscheduler' ).setLevel(logging.DEBUG)
|
總結
以上所述是小編給大家介紹的Python定時任務APScheduler的例項例項詳解,希望對大家有所幫助,如果大家有任何疑問請給我留言,小編會及時回覆大家的。在此也非常感謝大家對指令碼之家網站的支援!
如果你覺得本文對你有幫助,歡迎轉載,煩請註明出處,謝謝!