Python定時(shí)任務(wù)APScheduler的實(shí)例實(shí)例詳解
APScheduler 支持三種調(diào)度任務(wù):固定時(shí)間間隔,固定時(shí)間點(diǎn)(日期),Linux 下的 Crontab 命令。同時(shí),它還支持異步執(zhí)行、后臺(tái)執(zhí)行調(diào)度任務(wù)。
一、基本架構(gòu)
- 觸發(fā)器 triggers:設(shè)定觸發(fā)任務(wù)的條件
- 描述一個(gè)任務(wù)何時(shí)被觸發(fā),按日期或按時(shí)間間隔或按 cronjob 表達(dá)式三種方式觸發(fā)
- 任務(wù)存儲(chǔ)器 job stores:存放任務(wù),可以放內(nèi)存(默認(rèn))或數(shù)據(jù)庫(kù)
- 注:調(diào)度器之間不能共享任務(wù)存儲(chǔ)器
- 執(zhí)行器 executors:用于執(zhí)行任務(wù),可設(shè)定執(zhí)行模式
- 將指定的作業(yè)提交到線(xiàn)程池或者進(jìn)程池中運(yùn)行,任務(wù)完成通知調(diào)度器觸發(fā)相應(yīng)的事件。
- 調(diào)度器 schedulers:將上方三個(gè)組件作為參數(shù),創(chuàng)建調(diào)度器實(shí)例執(zhí)行。
協(xié)調(diào)三個(gè)組件的運(yùn)行。
二、調(diào)度器組件(schedulers)
- BlockingScheduler 阻塞式調(diào)度器
調(diào)度程序是進(jìn)程中唯一運(yùn)行的進(jìn)程,調(diào)用start函數(shù)會(huì)阻塞當(dāng)前線(xiàn)程,不能立即返回。
from apscheduler.schedulers.blocking import BlockingScheduler import time scheduler = BlockingScheduler() def job1(): print "%s: 執(zhí)行任務(wù)" % time.asctime() scheduler.add_job(job1, 'interval', seconds=3) scheduler.start()
- BackgroundScheduler 后臺(tái)調(diào)度器
當(dāng)前線(xiàn)程不會(huì)阻塞,調(diào)度器后臺(tái)執(zhí)行
from apscheduler.schedulers.background import BackgroundScheduler import time scheduler = BackgroundScheduler() def job1(): print "%s: 執(zhí)行任務(wù)" % time.asctime() scheduler.add_job(job1, 'interval', seconds=3) scheduler.start() time.sleep(10)
注:10秒執(zhí)行完后,程序結(jié)束。
- AsyncIOScheduler AsyncIO調(diào)度器
適用于使用了asyncio的情況
from apscheduler.schedulers.asyncio import AsyncIOScheduler import asyncio ... ... try: asyncio.get_event_loop().run_forever() except (KeyboardInterrupt, SystemExit): pass
- GeventScheduler Gevent調(diào)度器
使用了Gevent的情況
from apscheduler.schedulers.gevent import GeventScheduler ... ... g = scheduler.start() try: g.join() except (KeyboardInterrupt, SystemExit): pass
- TornadoScheduler Tornado調(diào)度器
適用于構(gòu)建Tornado應(yīng)用
- TwistedScheduler Twisted調(diào)度器
適用于構(gòu)建Twisted應(yīng)用
- QtScheduler Qt調(diào)度器
適用于構(gòu)建Qt應(yīng)用
三、觸發(fā)器組件(trigger)
date :只在某個(gè)時(shí)間點(diǎn)執(zhí)行一次,具體日期
run_date(datetime|str)
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 指定時(shí)區(qū)
interval :每隔一段時(shí)間允許一次,時(shí)間間隔
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 :任務(wù)的運(yùn)行周期
(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,它們的默認(rèn)值是 *
例如 day=1, minute=20 ,這就等于 year='*', month='*', day=1, week='*', day_of_week='*', hour='*',
minute=20, second=0 ,工作將在每個(gè)月的第一天以每小時(shí)20分鐘的時(shí)間執(zhí)行
表達(dá)式類(lèi)型
表達(dá)式 | 參數(shù)類(lèi)型 | 描述 |
---|---|---|
* | 所有 | 通配符。例: minutes=* 即每分鐘觸發(fā) |
*/a | 所有 | 可被a整除的通配符。 |
a-b | 所有 | 范圍a-b觸發(fā) |
a-b/c | 所有 | 范圍a-b,且可被c整除時(shí)觸發(fā) |
xth y | 日 | 第幾個(gè)星期幾觸發(fā)。x為第幾個(gè),y為星期幾 |
last x | 日 | 一個(gè)月中,最后個(gè)星期幾觸發(fā) |
last | 日 | 一個(gè)月最后一天觸發(fā) |
x,y,z | 所有 | 組合表達(dá)式,可以組合確定值或上方的表達(dá)式 |
注:當(dāng)設(shè)置的時(shí)間間隔小于,任務(wù)的執(zhí)行時(shí)間,線(xiàn)程會(huì)阻塞住,等待執(zhí)行完了才能執(zhí)行下一個(gè)任務(wù),可以設(shè)置 max_instance 指定一個(gè)任務(wù)同一時(shí)刻有多少個(gè)實(shí)例在運(yùn)行,默認(rèn)為1
四、配置調(diào)度器
線(xiàn)程池執(zhí)行器默認(rèn)為10,內(nèi)存任務(wù)存儲(chǔ)器為 memoryjobstore ,如果想自己配置的話(huà)可以執(zhí)行以下操作
需求:
- 兩個(gè)任務(wù)儲(chǔ)存器分別搭配兩個(gè)執(zhí)行器;同時(shí),還要修改任務(wù)的默認(rèn)參數(shù);最后還要改時(shí)區(qū)
- 名稱(chēng)為“mongo”的 MongoDBJobStore
- 名稱(chēng)為“default”的 SQLAlchemyJobStore
- 名稱(chēng)為“ThreadPoolExecutor ”的 ThreadPoolExecutor ,最大線(xiàn)程20個(gè)
- 名稱(chēng)“processpool”的 ProcessPoolExecutor ,最大進(jìn)程5個(gè)
- UTC時(shí)間作為調(diào)度器的時(shí)區(qū)
- 默認(rèn)為新任務(wù)關(guān)閉 合并模式 ()
- 設(shè)置新任務(wù)的默認(rèn)最大實(shí)例數(shù)為3
方法一:
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)
方法二:
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', })
方法三:
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() # ..這里可以添加任務(wù) scheduler.configure(jobstores=jobstores, executors=executors, job_defaults=job_defaults, timezone=utc)
五、啟動(dòng)調(diào)度器
除了 BlockingScheduler 外,其他非阻塞的調(diào)度器都會(huì)立即返回,運(yùn)行之后的代碼。
BlockingScheduler 需要將運(yùn)行的代碼放在start()之前
1.添加任務(wù)
1.調(diào)用add_job() #可以傳參max_instance,同一任務(wù)的運(yùn)行實(shí)例個(gè)數(shù)
當(dāng)有任務(wù)中途中斷,后面恢復(fù)后,有N個(gè)任務(wù)沒(méi)有執(zhí)行
coalesce:true ,恢復(fù)的任務(wù)會(huì)執(zhí)行一次
coalesce:false,恢復(fù)后的任務(wù)會(huì)執(zhí)行N次配合misfire_grace_time使用
misfire_grace_time設(shè)置時(shí)間差值,由于某些原因沒(méi)有運(yùn)行,再次提交時(shí),大于設(shè)置的時(shí)間,實(shí)例不會(huì)運(yùn)行。
2.裝飾器scheduled_job()
立即運(yùn)行可以不設(shè)置trigger參數(shù)
2.移除任務(wù)
# 根據(jù)任務(wù)實(shí)例刪除 job = scheduler.add_job(myfunc, 'interval', minutes=2) job.remove() # 根據(jù)任務(wù)id刪除 scheduler.add_job(myfunc, 'interval', minutes=2, id='my_job_id') scheduler.remove_job('my_job_id')
3.暫停任務(wù)
job = scheduler.add_job(myfunc, 'interval', minutes=2) # 根據(jù)任務(wù)實(shí)例 job.pause() #暫停 job.resume() #繼續(xù) # 根據(jù)任務(wù)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.調(diào)度器操作
scheduler.start() #開(kāi)啟 scheduler.shotdown(wait=True|False) #關(guān)閉 False 無(wú)論任務(wù)是否執(zhí)行,強(qiáng)制關(guān)閉
異常捕獲
# 可以添加apscheduler日志至DEBUG級(jí)別,這樣就能捕獲異常信息 import logging logging.basicConfig() logging.getLogger('apscheduler').setLevel(logging.DEBUG)
總結(jié)
以上所述是小編給大家介紹的Python定時(shí)任務(wù)APScheduler的實(shí)例實(shí)例詳解,希望對(duì)大家有所幫助,如果大家有任何疑問(wèn)請(qǐng)給我留言,小編會(huì)及時(shí)回復(fù)大家的。在此也非常感謝大家對(duì)腳本之家網(wǎng)站的支持!
如果你覺(jué)得本文對(duì)你有幫助,歡迎轉(zhuǎn)載,煩請(qǐng)注明出處,謝謝!
相關(guān)文章
教你使用Python獲取QQ音樂(lè)某個(gè)歌手的歌單
這篇文章主要介紹了Python獲取QQ音樂(lè)某個(gè)歌手的歌單,從qq音樂(lè)中獲取某個(gè)你喜歡的歌手的清單,涉及到的庫(kù)有requests、json,本文結(jié)合示例代碼給大家介紹的非常詳細(xì),需要的朋友可以參考下2022-04-04Python實(shí)現(xiàn)批量獲取當(dāng)前文件夾下的文件名
這篇文章主要為大家詳細(xì)介紹了如何利用Python實(shí)現(xiàn)批量獲取當(dāng)前文件夾下的文件名,文中的示例代碼講解詳細(xì),感興趣的小伙伴可以跟隨小編一起學(xué)習(xí)一下2024-02-02利用插件和python實(shí)現(xiàn)Excel轉(zhuǎn)json的兩種辦法
轉(zhuǎn)換Excel表格到JSON格式有很多方法,下面這篇文章主要給大家介紹了關(guān)于利用插件和python實(shí)現(xiàn)Excel轉(zhuǎn)json的兩種辦法,文中通過(guò)代碼介紹的非常詳細(xì),需要的朋友可以參考下2023-11-11弄懂這56個(gè)Python使用技巧(輕松掌握Python高效開(kāi)發(fā))
這篇文章主要介紹了弄懂這56個(gè)Python使用技巧(輕松掌握Python高效開(kāi)發(fā)),小編覺(jué)得挺不錯(cuò)的,現(xiàn)在分享給大家,也給大家做個(gè)參考。一起跟隨小編過(guò)來(lái)看看吧2019-09-09用Python編寫(xiě)一個(gè)國(guó)際象棋AI程序
在這篇文章中我會(huì)介紹這個(gè)AI如何工作,每一個(gè)部分做什么,它為什么能那樣工作起來(lái)。你可以直接通讀本文,或者去下載代碼,邊讀邊看代碼。雖然去看看其他文件中有什么AI依賴(lài)的類(lèi)也可能有幫助,但是AI部分全都在A(yíng)I.py文件中2014-11-11Python中字典的基礎(chǔ)知識(shí)歸納小結(jié)
這篇文章主要介紹了Python中字典的基礎(chǔ)知識(shí)歸納小結(jié),都是Python入門(mén)學(xué)習(xí)中的基本知識(shí),值得反復(fù)鞏固:)需要的朋友可以參考下2015-08-08python scp 批量同步文件的實(shí)現(xiàn)方法
今天小編就為大家分享一篇python scp 批量同步文件的實(shí)現(xiàn)方法,具有很好的參考價(jià)值,希望對(duì)大家有所幫助。一起跟隨小編過(guò)來(lái)看看吧2019-01-01