Python定時任務APScheduler的實例實例詳解
更新時間：2019年07月22日 08:53:06 作者：mangM
APScheduler 支持三種調度任務：固定時間間隔，固定時間點（日期），Linux 下的 Crontab 命令。這篇文章主要介紹了Python定時任務APScheduler的使用,需要的朋友可以參考下
APScheduler 支持三種調度任務：固定時間間隔，固定時間點（日期），Linux 下的 Crontab 命令。同時，它還支持異步執(zhí)行、后臺執(zhí)行調度任務。
一、基本架構
- 觸發(fā)器 triggers：設定觸發(fā)任務的條件
- 描述一個任務何時被觸發(fā)，按日期或按時間間隔或按 cronjob 表達式三種方式觸發(fā)
- 任務存儲器 job stores：存放任務，可以放內存(默認)或數(shù)據(jù)庫
- 注：調度器之間不能共享任務存儲器
- 執(zhí)行器 executors：用于執(zhí)行任務，可設定執(zhí)行模式
- 將指定的作業(yè)提交到線程池或者進程池中運行，任務完成通知調度器觸發(fā)相應的事件。
- 調度器 schedulers：將上方三個組件作為參數(shù)，創(chuàng)建調度器實例執(zhí)行。
協(xié)調三個組件的運行。
二、調度器組件(schedulers)
- BlockingScheduler 阻塞式調度器
調度程序是進程中唯一運行的進程，調用start函數(shù)會阻塞當前線程，不能立即返回。
from apscheduler.schedulers.blocking import BlockingScheduler import time scheduler = BlockingScheduler() def job1(): print "%s: 執(zhí)行任務" % time.asctime() scheduler.add_job(job1, 'interval', seconds=3) scheduler.start()
- BackgroundScheduler 后臺調度器
當前線程不會阻塞，調度器后臺執(zhí)行
from apscheduler.schedulers.background import BackgroundScheduler import time scheduler = BackgroundScheduler() def job1(): print "%s: 執(zhí)行任務" % time.asctime() scheduler.add_job(job1, 'interval', seconds=3) scheduler.start() time.sleep(10)
注：10秒執(zhí)行完后，程序結束。
- AsyncIOScheduler AsyncIO調度器
適用于使用了asyncio的情況
from apscheduler.schedulers.asyncio import AsyncIOScheduler import asyncio ... ... try: asyncio.get_event_loop().run_forever() except (KeyboardInterrupt, SystemExit): pass
- GeventScheduler Gevent調度器
使用了Gevent的情況
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應用
三、觸發(fā)器組件(trigger)
date ：只在某個時間點執(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 指定時區(qū)
interval ：每隔一段時間允許一次，時間間隔
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 ：任務的運行周期
(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分鐘的時間執(zhí)行
表達式類型
| 表達式 | 參數(shù)類型 | 描述 |
|---|---|---|
| * | 所有 | 通配符。例： minutes=* 即每分鐘觸發(fā) |
| */a | 所有 | 可被a整除的通配符。 |
| a-b | 所有 | 范圍a-b觸發(fā) |
| a-b/c | 所有 | 范圍a-b，且可被c整除時觸發(fā) |
| xth y | 日 | 第幾個星期幾觸發(fā)。x為第幾個，y為星期幾 |
| last x | 日 | 一個月中，最后個星期幾觸發(fā) |
| last | 日 | 一個月最后一天觸發(fā) |
| x,y,z | 所有 | 組合表達式，可以組合確定值或上方的表達式 |
注：當設置的時間間隔小于，任務的執(zhí)行時間，線程會阻塞住，等待執(zhí)行完了才能執(zhí)行下一個任務，可以設置 max_instance 指定一個任務同一時刻有多少個實例在運行，默認為1
四、配置調度器
線程池執(zhí)行器默認為10，內存任務存儲器為 memoryjobstore ,如果想自己配置的話可以執(zhí)行以下操作
需求：
- 兩個任務儲存器分別搭配兩個執(zhí)行器；同時，還要修改任務的默認參數(shù)；最后還要改時區(qū)
- 名稱為“mongo”的 MongoDBJobStore
- 名稱為“default”的 SQLAlchemyJobStore
- 名稱為“ThreadPoolExecutor ”的 ThreadPoolExecutor ，最大線程20個
- 名稱“processpool”的 ProcessPoolExecutor ，最大進程5個
- UTC時間作為調度器的時區(qū)
- 默認為新任務關閉 合并模式 （）
- 設置新任務的默認最大實例數(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()
# ..這里可以添加任務
scheduler.configure(jobstores=jobstores, executors=executors, job_defaults=job_defaults, timezone=utc)
五、啟動調度器
除了 BlockingScheduler 外，其他非阻塞的調度器都會立即返回，運行之后的代碼。
BlockingScheduler 需要將運行的代碼放在start()之前
1.添加任務
1.調用add_job() #可以傳參max_instance,同一任務的運行實例個數(shù)
當有任務中途中斷，后面恢復后，有N個任務沒有執(zhí)行
coalesce：true ，恢復的任務會執(zhí)行一次
coalesce：false，恢復后的任務會執(zhí)行N次配合misfire_grace_time使用
misfire_grace_time設置時間差值，由于某些原因沒有運行，再次提交時，大于設置的時間，實例不會運行。
2.裝飾器scheduled_job()
立即運行可以不設置trigger參數(shù)
2.移除任務
# 根據(jù)任務實例刪除
job = scheduler.add_job(myfunc, 'interval', minutes=2)
job.remove()
# 根據(jù)任務id刪除
scheduler.add_job(myfunc, 'interval', minutes=2, id='my_job_id')
scheduler.remove_job('my_job_id')
3.暫停任務
job = scheduler.add_job(myfunc, 'interval', minutes=2)
# 根據(jù)任務實例
job.pause() #暫停
job.resume() #繼續(xù)
# 根據(jù)任務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.調度器操作
scheduler.start() #開啟 scheduler.shotdown(wait=True|False) #關閉 False 無論任務是否執(zhí)行，強制關閉
異常捕獲
# 可以添加apscheduler日志至DEBUG級別，這樣就能捕獲異常信息
import logging
logging.basicConfig()
logging.getLogger('apscheduler').setLevel(logging.DEBUG)
總結
以上所述是小編給大家介紹的Python定時任務APScheduler的實例實例詳解,希望對大家有所幫助，如果大家有任何疑問請給我留言，小編會及時回復大家的。在此也非常感謝大家對腳本之家網(wǎng)站的支持！
如果你覺得本文對你有幫助，歡迎轉載，煩請注明出處，謝謝！
相關文章
基于Python的網(wǎng)頁自動化工具DrissionPage的使用詳解
DrissionPage 是一個基于 python 的網(wǎng)頁自動化工具,它既能控制瀏覽器,也能收發(fā)數(shù)據(jù)包,還能把兩者合而為一,下面就跟隨小編一起來學習一下它的具體使用吧2024-01-01
這篇文章主要介紹了pycharm 如何縮進和SQL亂碼及SQL包含變量，具有很好的參考價值，希望對大家有所幫助。如有錯誤或未考慮完全的地方，望不吝賜教2022-07-07
Python數(shù)據(jù)持久化存儲實現(xiàn)方法分析
這篇文章主要介紹了Python數(shù)據(jù)持久化存儲實現(xiàn)方法,結合實例形式分析了Python基于pymongo及mysql模塊的數(shù)據(jù)持久化存儲操作相關實現(xiàn)技巧,需要的朋友可以參考下2019-12-12
今天小編就為大家分享一篇使用pandas把某一列的字符值轉換為數(shù)字的實例，具有很好的參考價值，希望對大家有所幫助。一起跟隨小編過來看看吧2019-01-01
在django中使用post方法時,需要增加csrftoken的例子
這篇文章主要介紹了在django中使用post方法時,需要增加csrftoken的例子，具有很好的參考價值，希望對大家有所幫助。一起跟隨小編過來看看吧2020-03-03
這篇文章主要介紹了pytorch 實現(xiàn)查看數(shù)據(jù)類型和大小的操作，具有很好的參考價值，希望對大家有所幫助。如有錯誤或未考慮完全的地方，望不吝賜教2021-05-05
這篇文章主要介紹了python 代碼運行時間獲取方式，文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值，需要的朋友們下面隨著小編來一起學習學習吧2020-09-09
這篇文章主要介紹了OpenCV dlib實現(xiàn)人臉采集，本文給大家介紹的非常詳細，對大家的學習或工作具有一定的參考借鑒價值，需要的朋友可以參考下2021-09-09
這篇文章主要為大家介紹了爬蟲訓練前端基礎Bootstrap5排版表格圖像，有需要的朋友可以借鑒參考下，希望能夠有所幫助，祝大家多多進步，早日升職加薪2023-02-02
這篇文章主要介紹了Python有參函數(shù)使用代碼實例,文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值,需要的朋友可以參考下2020-01-01