Python高效定時(shí)任務(wù)處理APScheduler庫深入學(xué)習(xí)
介紹
APScheduler是Python中一個(gè)強(qiáng)大的第三方庫，用于在后臺(tái)執(zhí)行定時(shí)任務(wù)。它允許我們根據(jù)設(shè)定的時(shí)間間隔、日期規(guī)則或特定時(shí)間來執(zhí)行任務(wù)，適用于定時(shí)執(zhí)行腳本、定時(shí)發(fā)送郵件、定時(shí)處理數(shù)據(jù)等場(chǎng)景。APScheduler的功能使得在Python中實(shí)現(xiàn)定時(shí)任務(wù)變得非常簡單和高效。本文將從入門到精通地介紹APScheduler庫的使用方法，帶你掌握在Python中實(shí)現(xiàn)定時(shí)任務(wù)的技巧。
- 安裝和導(dǎo)入
- 創(chuàng)建定時(shí)任務(wù)
- 定時(shí)任務(wù)觸發(fā)器
- 任務(wù)存儲(chǔ)
- 并發(fā)執(zhí)行
- 阻塞和非阻塞
- 錯(cuò)誤處理
- 立即執(zhí)行任務(wù)
- 調(diào)度器持久化
- 任務(wù)監(jiān)聽器
- 移除定時(shí)任務(wù)
1. 安裝和導(dǎo)入
首先，我們需要安裝APScheduler庫?？梢允褂胮ip命令進(jìn)行安裝：
pip install apscheduler
安裝完成后，我們可以在Python代碼中導(dǎo)入APScheduler：
from apscheduler.schedulers.background import BackgroundScheduler
2. 創(chuàng)建定時(shí)任務(wù)
APScheduler提供了BackgroundScheduler和BlockingScheduler兩種類型的調(diào)度器，用于創(chuàng)建定時(shí)任務(wù)。BackgroundScheduler在后臺(tái)運(yùn)行，不會(huì)阻塞主線程；而BlockingScheduler會(huì)阻塞主線程直到所有任務(wù)完成。
from apscheduler.schedulers.background import BackgroundScheduler
import time
# 創(chuàng)建后臺(tái)調(diào)度器
scheduler = BackgroundScheduler()
# 定義任務(wù)函數(shù)
def job():
print("定時(shí)任務(wù)執(zhí)行：", time.strftime("%Y-%m-%d %H:%M:%S"))
# 添加定時(shí)任務(wù)，每隔5秒執(zhí)行一次
scheduler.add_job(job, 'interval', seconds=5)
# 啟動(dòng)調(diào)度器
scheduler.start()
# 主線程等待一段時(shí)間后結(jié)束
time.sleep(20)
# 關(guān)閉調(diào)度器
scheduler.shutdown()
print("主線程結(jié)束")在上述代碼中，我們首先創(chuàng)建了一個(gè)后臺(tái)調(diào)度器scheduler，然后定義了一個(gè)名為job的任務(wù)函數(shù)，在其中打印當(dāng)前時(shí)間。使用scheduler.add_job()添加了一個(gè)定時(shí)任務(wù)，設(shè)置為每隔5秒執(zhí)行一次。然后，我們啟動(dòng)了調(diào)度器scheduler，讓定時(shí)任務(wù)在后臺(tái)執(zhí)行。主線程等待20秒后結(jié)束，并調(diào)用scheduler.shutdown()關(guān)閉調(diào)度器。
3. 定時(shí)任務(wù)觸發(fā)器
APScheduler提供了多種觸發(fā)器類型，用于設(shè)置定時(shí)任務(wù)的觸發(fā)條件。
interval觸發(fā)器： 按照設(shè)定的時(shí)間間隔來觸發(fā)任務(wù)。
from apscheduler.schedulers.background import BackgroundScheduler
import time
# 創(chuàng)建后臺(tái)調(diào)度器
scheduler = BackgroundScheduler()
# 定義任務(wù)函數(shù)
def job():
print("定時(shí)任務(wù)執(zhí)行：", time.strftime("%Y-%m-%d %H:%M:%S"))
# 添加定時(shí)任務(wù)，每隔5秒執(zhí)行一次
scheduler.add_job(job, 'interval', seconds=5)
# 啟動(dòng)調(diào)度器
scheduler.start()
# 主線程等待一段時(shí)間后結(jié)束
time.sleep(20)
# 關(guān)閉調(diào)度器
scheduler.shutdown()
print("主線程結(jié)束")在上述代碼中，我們使用'interval'觸發(fā)器，設(shè)置任務(wù)每隔5秒執(zhí)行一次。
cron觸發(fā)器： 使用類似于Linux中cron表達(dá)式的規(guī)則來觸發(fā)任務(wù)，可以精確到秒。
from apscheduler.schedulers.background import BackgroundScheduler
import time
# 創(chuàng)建后臺(tái)調(diào)度器
scheduler = BackgroundScheduler()
# 定義任務(wù)函數(shù)
def job():
print("定時(shí)任務(wù)執(zhí)行：", time.strftime("%Y-%m-%d %H:%M:%S"))
# 添加定時(shí)任務(wù)，每天的13點(diǎn)30分觸發(fā)任務(wù)
scheduler.add_job(job, 'cron', hour=13, minute=30)
# 啟動(dòng)調(diào)度器
scheduler.start()
# 主線程等待一段時(shí)間后結(jié)束
time.sleep(60)
# 關(guān)閉調(diào)度器
scheduler.shutdown()
print("主線程結(jié)束")在上述代碼中，我們使用'cron'觸發(fā)器，設(shè)置任務(wù)每天的13點(diǎn)30分觸發(fā)。
date觸發(fā)器： 在指定的時(shí)間點(diǎn)觸發(fā)任務(wù)。
from apscheduler.schedulers.background import BackgroundScheduler
import time
# 創(chuàng)建后臺(tái)調(diào)度器
scheduler = BackgroundScheduler()
# 定義任務(wù)函數(shù)
def job():
print("定時(shí)任務(wù)執(zhí)行：", time.strftime("%Y-%m-%d %H:%M:%S"))
# 添加定時(shí)任務(wù)，設(shè)置任務(wù)在2023年7月31日10點(diǎn)30分觸發(fā)
scheduler.add_job(job, 'date', run_date='2023-07-31 10:30:00')
# 啟動(dòng)調(diào)度器
scheduler.start()
# 主線程等待一段時(shí)間后結(jié)束
time.sleep(60)
# 關(guān)閉調(diào)度器
scheduler.shutdown()
print("主線程結(jié)束")在上述代碼中，我們使用'date'觸發(fā)器，設(shè)置任務(wù)在2023年7月31日10點(diǎn)30分觸發(fā)。
4. 任務(wù)存儲(chǔ)
APScheduler支持將任務(wù)存儲(chǔ)在不同的后端存儲(chǔ)中，如內(nèi)存、數(shù)據(jù)庫等。默認(rèn)情況下，任務(wù)是存儲(chǔ)在內(nèi)存中的。
from apscheduler.schedulers.background import BackgroundScheduler
import time
# 創(chuàng)建后臺(tái)調(diào)度器
scheduler = BackgroundScheduler()
# 定義任務(wù)函數(shù)
def job():
print("定時(shí)任務(wù)執(zhí)行：", time.strftime("%Y-%m-%d %H:%M:%S"))
# 添加定時(shí)任務(wù)，每隔5秒執(zhí)行一次
scheduler.add_job(job, 'interval', seconds=5)
# 啟動(dòng)調(diào)度器
scheduler.start()
# 主線程等待一段時(shí)間后結(jié)束
time.sleep(20)
# 關(guān)閉調(diào)度器
scheduler.shutdown()
print("主線程結(jié)束")在上述代碼中，我們使用默認(rèn)的內(nèi)存存儲(chǔ)來存儲(chǔ)任務(wù)。
如果需要將任務(wù)存儲(chǔ)在數(shù)據(jù)庫中，可以使用jobstores參數(shù)來設(shè)置。
from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore
import time
# 創(chuàng)建后臺(tái)調(diào)度器
scheduler = BackgroundScheduler()
# 創(chuàng)建數(shù)據(jù)庫存儲(chǔ)
jobstores = {
'default': SQLAlchemyJobStore(url='sqlite:///jobs.sqlite')
}
# 定義任務(wù)函數(shù)
def job():
print("定時(shí)任務(wù)執(zhí)行：", time.strftime("%Y-%m-%d %H:%M:%S"))
# 添加定時(shí)任務(wù)，每隔5秒執(zhí)行一次
scheduler.add_job(job, 'interval', seconds=5)
# 啟動(dòng)調(diào)度器
scheduler.start()
# 主線程等待一段時(shí)間后結(jié)束
time.sleep(20)
# 關(guān)閉調(diào)度器
scheduler.shutdown()
print("主線程結(jié)束")在上述代碼中，我們使用了SQLAlchemyJobStore來將任務(wù)存儲(chǔ)在SQLite數(shù)據(jù)庫中。
5. 并發(fā)執(zhí)行
默認(rèn)情況下，APScheduler會(huì)將任務(wù)串行執(zhí)行，也就是說一個(gè)任務(wù)結(jié)束后才會(huì)執(zhí)行下一個(gè)任務(wù)。如果希望并發(fā)執(zhí)行多個(gè)任務(wù)，可以使用max_instances參數(shù)來設(shè)置。
from apscheduler.schedulers.background import BackgroundScheduler
import time
# 創(chuàng)建后臺(tái)調(diào)度器
scheduler = BackgroundScheduler()
# 定義任務(wù)函數(shù)
def job(index):
print(f"定時(shí)任務(wù){(diào)index}執(zhí)行：", time.strftime("%Y-%m-%d %H:%M:%S"))
# 添加定時(shí)任務(wù)，每隔5秒執(zhí)行一次，最多并發(fā)3個(gè)任務(wù)
scheduler.add_job(job, 'interval', seconds=5, args=[1], max_instances=3)
scheduler.add_job(job, 'interval', seconds=5, args=[2], max_instances=3)
scheduler.add_job(job, 'interval', seconds=5, args=[3], max_instances=3)
# 啟動(dòng)調(diào)度器
scheduler.start()
# 主線程等待一段時(shí)間后結(jié)束
time.sleep(20)
# 關(guān)閉調(diào)度器
scheduler.shutdown()
print("主線程結(jié)束")在上述代碼中，我們使用了args參數(shù)傳遞參數(shù)給任務(wù)函數(shù)，并使用max_instances參數(shù)設(shè)置最多并發(fā)3個(gè)任務(wù)。
6. 阻塞和非阻塞
APScheduler提供了阻塞和非阻塞兩種調(diào)度器類型。
阻塞調(diào)度器： 在調(diào)度器啟動(dòng)后，會(huì)阻塞主線程直到所有任務(wù)完成。
from apscheduler.schedulers.blocking import BlockingScheduler
import time
# 創(chuàng)建阻塞調(diào)度器
scheduler = BlockingScheduler()
# 定義任務(wù)函數(shù)
def job():
print("定時(shí)任務(wù)執(zhí)行：", time.strftime("%Y-%m-%d %H:%M:%S"))
# 添加定時(shí)任務(wù)，每隔5秒執(zhí)行一次
scheduler.add_job(job, 'interval', seconds=5)
# 啟動(dòng)調(diào)度器
scheduler.start()
print("主線程結(jié)束")非阻塞調(diào)度器： 在調(diào)度器啟動(dòng)后，不會(huì)阻塞主線程。
from apscheduler.schedulers.background import BackgroundScheduler
import time
# 創(chuàng)建后臺(tái)調(diào)度器
scheduler = BackgroundScheduler()
# 定義任務(wù)函數(shù)
def job():
print("定時(shí)任務(wù)執(zhí)行：", time.strftime("%Y-%m-%d %H:%M:%S"))
# 添加定時(shí)任務(wù)，每隔5秒執(zhí)行一次
scheduler.add_job(job, 'interval', seconds=5)
# 啟動(dòng)調(diào)度器
scheduler.start()
# 主線程等待一段時(shí)間后結(jié)束
time.sleep(20)
# 關(guān)閉調(diào)度器
scheduler.shutdown()
print("主線程結(jié)束")在上述代碼中，我們分別使用BlockingScheduler和BackgroundScheduler創(chuàng)建了阻塞和非阻塞調(diào)度器。
7. 錯(cuò)誤處理
在任務(wù)執(zhí)行過程中，可能會(huì)出現(xiàn)異常。APScheduler提供了異常處理機(jī)制，我們可以通過try...except...捕獲任務(wù)函數(shù)中的異常，并進(jìn)行相應(yīng)的處理。
from apscheduler.schedulers.background import BackgroundScheduler
import time
# 創(chuàng)建后臺(tái)調(diào)度器
scheduler = BackgroundScheduler()
# 定義任務(wù)函數(shù)
def job():
try:
print("定時(shí)任務(wù)執(zhí)行：", time.strftime("%Y-%m-%d %H:%M:%S"))
# 拋出一個(gè)異常
raise ValueError("任務(wù)出現(xiàn)異常")
except Exception as e:
print("任務(wù)執(zhí)行過程中發(fā)生異常：", str(e))
# 添加定時(shí)任務(wù)，每隔5秒執(zhí)行一次
scheduler.add_job(job, 'interval', seconds=5)
# 啟動(dòng)調(diào)度器
scheduler.start()
# 主線程等待一段時(shí)間后結(jié)束
time.sleep(20)
# 關(guān)閉調(diào)度器
scheduler.shutdown()
print("主線程結(jié)束")在上述代碼中，我們?cè)谌蝿?wù)函數(shù)中拋出了一個(gè)ValueError異常，并通過try...except...捕獲并輸出了異常信息。
8. 立即執(zhí)行任務(wù)
有時(shí)候我們可能需要立即執(zhí)行一個(gè)任務(wù)，而不是等到下次觸發(fā)時(shí)間。APScheduler提供了run_job方法來立即執(zhí)行任務(wù)。
from apscheduler.schedulers.background import BackgroundScheduler
import time
# 創(chuàng)建后臺(tái)調(diào)度器
scheduler = BackgroundScheduler()
# 定義任務(wù)函數(shù)
def job():
print("定時(shí)任務(wù)執(zhí)行：", time.strftime("%Y-%m-%d %H:%M:%S"))
# 添加定時(shí)任務(wù)，每隔5秒執(zhí)行一次
scheduler.add_job(job, 'interval', seconds=5)
# 啟動(dòng)調(diào)度器
scheduler.start()
# 立即執(zhí)行任務(wù)
scheduler.run_job(job)
# 主線程等待一段時(shí)間后結(jié)束
time.sleep(20)
# 關(guān)閉調(diào)度器
scheduler.shutdown()
print("主線程結(jié)束")在上述代碼中，我們使用scheduler.run_job(job)方法立即執(zhí)行了任務(wù)。
9. 調(diào)度器持久化
在實(shí)際應(yīng)用中，我們可能需要將調(diào)度器的配置保存到文件中，以便在下次啟動(dòng)時(shí)恢復(fù)。
from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore
import time
# 創(chuàng)建數(shù)據(jù)庫存儲(chǔ)
jobstores = {
'default': SQLAlchemyJobStore(url='sqlite:///jobs.sqlite')
}
# 創(chuàng)建后臺(tái)調(diào)度器，并指定jobstores參數(shù)
scheduler = BackgroundScheduler(jobstores=jobstores)
# 定義任務(wù)函數(shù)
def job():
print("定時(shí)任務(wù)執(zhí)行：", time.strftime("%Y-%m-%d %H:%M:%S"))
# 添加定時(shí)任務(wù)，每隔5秒執(zhí)行一次
scheduler.add_job(job, 'interval', seconds=5)
# 啟動(dòng)調(diào)度器
scheduler.start()
# 主線程等待一段時(shí)間后結(jié)束
time.sleep(20)
# 關(guān)閉調(diào)度器
scheduler.shutdown()
print("主線程結(jié)束")在上述代碼中，我們創(chuàng)建了一個(gè)數(shù)據(jù)庫存儲(chǔ)jobstores，并在創(chuàng)建后臺(tái)調(diào)度器時(shí)指定了jobstores參數(shù)。這樣，在調(diào)度器運(yùn)行過程中，任務(wù)的配置將會(huì)被持久化到數(shù)據(jù)庫中。
10. 任務(wù)監(jiān)聽器
APScheduler提供了任務(wù)監(jiān)聽器，用于監(jiān)聽任務(wù)的狀態(tài)變化。我們可以通過add_listener方法添加監(jiān)聽器，并在任務(wù)狀態(tài)發(fā)生變化時(shí)進(jìn)行相應(yīng)的處理。
from apscheduler.schedulers.background import BackgroundScheduler
import time
# 創(chuàng)建后臺(tái)調(diào)度器
scheduler = BackgroundScheduler()
# 定義任務(wù)函數(shù)
def job():
print("定時(shí)任務(wù)執(zhí)行：", time.strftime("%Y-%m-%d %H:%M:%S"))
# 添加定時(shí)任務(wù)，每隔5秒執(zhí)行一次
scheduler.add_job(job, 'interval', seconds=5)
# 定義任務(wù)監(jiān)聽器
def my_listener(event):
if event.exception:
print("任務(wù)執(zhí)行過程中發(fā)生異常：", str(event.exception))
else:
print("任務(wù)執(zhí)行成功")
# 添加任務(wù)監(jiān)聽器
scheduler.add_listener(my_listener, mask='all')
# 啟動(dòng)調(diào)度器
scheduler.start()
# 主線程等待一段時(shí)間后結(jié)束
time.sleep(20)
# 關(guān)閉調(diào)度器
scheduler.shutdown()
print("主線程結(jié)束")在上述代碼中，我們創(chuàng)建了一個(gè)任務(wù)監(jiān)聽器my_listener，并在任務(wù)執(zhí)行過程中通過if...else...判斷是否出現(xiàn)異常。然后通過scheduler.add_listener(my_listener, mask='all')方法添加了監(jiān)聽器。
11. 移除定時(shí)任務(wù)
如果我們希望在調(diào)度器運(yùn)行過程中移除某個(gè)定時(shí)任務(wù)，可以使用scheduler.remove_job(job_id)方法。
from apscheduler.schedulers.background import BackgroundScheduler
import time
# 創(chuàng)建后臺(tái)調(diào)度器
scheduler = BackgroundScheduler()
# 定義任務(wù)函數(shù)
def job():
print("定時(shí)任務(wù)執(zhí)行：", time.strftime("%Y-%m-%d %H:%M:%S"))
# 添加定時(shí)任務(wù)，每隔5秒執(zhí)行一次，并獲取任務(wù)ID
job_id = scheduler.add_job(job, 'interval', seconds=5).id
# 啟動(dòng)調(diào)度器
scheduler.start()
# 主線程等待一段時(shí)間后移除定時(shí)任務(wù)
time.sleep(10)
scheduler.remove_job(job_id)
# 主線程等待一段時(shí)間后結(jié)束
time.sleep(10)
# 關(guān)閉調(diào)度器
scheduler.shutdown()
print("主線程結(jié)束")在上述代碼中，我們通過scheduler.add_job(job, 'interval', seconds=5).id獲取了定時(shí)任務(wù)的ID，并使用scheduler.remove_job(job_id)移除了定時(shí)任務(wù)。
總結(jié)
通過本文的介紹，我們學(xué)習(xí)了APScheduler庫的基本用法，包括創(chuàng)建定時(shí)任務(wù)、定時(shí)任務(wù)觸發(fā)器、任務(wù)存儲(chǔ)、并發(fā)執(zhí)行、阻塞和非阻塞調(diào)度器、錯(cuò)誤處理、立即執(zhí)行任務(wù)、調(diào)度器持久化、任務(wù)監(jiān)聽器和移除定時(shí)任務(wù)等。APScheduler為Python開發(fā)者提供了一個(gè)強(qiáng)大的定時(shí)任務(wù)調(diào)度框架，使得在Python中實(shí)現(xiàn)定時(shí)任務(wù)變得非常簡單和高效。掌握APScheduler的使用將為我們的項(xiàng)目和程序帶來很大的便利。
以上就是Python高效定時(shí)任務(wù)處理APScheduler庫深入學(xué)習(xí)的詳細(xì)內(nèi)容，更多關(guān)于Python APScheduler定時(shí)任務(wù)的資料請(qǐng)關(guān)注腳本之家其它相關(guān)文章！
- python自動(dòng)定時(shí)任務(wù)schedule庫的使用方法
- Python apscheduler實(shí)現(xiàn)定時(shí)任務(wù)的方法詳解
- Python?Apschedule定時(shí)任務(wù)框架的用法詳解
- Python第三方模塊apscheduler安裝和基本使用
- python定時(shí)任務(wù)schedule庫用法詳細(xì)講解
- Python flask框架定時(shí)任務(wù)apscheduler應(yīng)用介紹
- Python中schedule模塊關(guān)于定時(shí)任務(wù)使用方法
- Python定時(shí)任務(wù)框架APScheduler安裝使用詳解
- 最新Python?APScheduler?定時(shí)任務(wù)詳解
- Python中schedule擴(kuò)展的具體使用
相關(guān)文章
Python實(shí)現(xiàn)平行坐標(biāo)圖的兩種方法小結(jié)
今天小編就為大家分享一篇Python實(shí)現(xiàn)平行坐標(biāo)圖的兩種方法小結(jié)，具有很好的參考價(jià)值，希望對(duì)大家有所幫助。一起跟隨小編過來看看吧2019-07-07
Python操作數(shù)據(jù)庫之?dāng)?shù)據(jù)庫編程接口
這篇文章主要介紹了Python操作數(shù)據(jù)庫之?dāng)?shù)據(jù)庫編程接口，文章圍繞主題展開詳細(xì)的內(nèi)容介紹，具有一定的參考價(jià)值，感興趣的小伙伴可以參考一下2022-06-06
Python CSS選擇器爬取京東網(wǎng)商品信息過程解析
這篇文章主要介紹了Python CSS選擇器爬取京東網(wǎng)商品信息過程解析,文中通過示例代碼介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值,需要的朋友可以參考下2020-06-06
今天教大家如何利用python進(jìn)行數(shù)值分析,文中有非常詳細(xì)的代碼示例,對(duì)正在學(xué)習(xí)python的小伙伴們很有幫助,需要的朋友可以參考下2021-06-06
一篇文章教你用Python實(shí)現(xiàn)一鍵文件重命名
這篇文章主要介紹了如何用python一鍵文件重命名，文中示例代碼介紹的非常詳細(xì)，具有一定的參考價(jià)值，感興趣的小伙伴們可以參考一下2021-09-09
解決python中畫圖時(shí)x,y軸名稱出現(xiàn)中文亂碼的問題
今天小編就為大家分享一篇解決python中畫圖時(shí)x,y軸名稱出現(xiàn)中文亂碼的問題，具有很好的參考價(jià)值，希望對(duì)大家有所幫助。一起跟隨小編過來看看吧2019-01-01
本篇文章主要介紹了Python中Django 后臺(tái)自定義表單控件，其實(shí) django 已經(jīng)為我們提供了一些可用的表單控件，比如：多選框、單選按鈕等，有興趣的開業(yè)了解一下。2017-03-03
簡單幾步教你學(xué)會(huì)Python接口自動(dòng)化測(cè)試
這篇文章主要介紹了簡單幾步教你學(xué)會(huì)Python接口自動(dòng)化測(cè)試,本文從一個(gè)簡單的登錄接口測(cè)試入手，一步步調(diào)整優(yōu)化接口調(diào)用姿勢(shì),期望讀者可以通過本文對(duì)接口自動(dòng)化測(cè)試有一個(gè)大致的了解,需要的朋友可以參考下2023-08-08
這篇文章主要介紹了Django session登陸并獲取值的實(shí)例，具有很好的參考價(jià)值，希望對(duì)大家有所幫助。一起跟隨小編過來看看吧2021-03-03
Python簡單實(shí)現(xiàn)網(wǎng)頁內(nèi)容抓取功能示例
這篇文章主要介紹了Python簡單實(shí)現(xiàn)網(wǎng)頁內(nèi)容抓取功能,結(jié)合實(shí)例形式分析了Python基于urllib模塊的網(wǎng)頁請(qǐng)求、內(nèi)容讀取等相關(guān)操作技巧,需要的朋友可以參考下2018-06-06