詳解Python利用APScheduler框架實現(xiàn)定時任務(wù)
背景
最近在做一些python工具的時候,常常會碰到定時器問題,總覺著使用threading.timer或者schedule模塊非常不優(yōu)雅。所以這里給自己做個記錄,也分享一個定時任務(wù)框架APScheduler。具體的架構(gòu)原理就不細(xì)說了,用個例子說明一下怎么簡易的使用。
樣例代碼
先上樣例代碼,如下:
#!/user/bin/env python
# coding=utf-8
"""
@project : csdn
@author : 劍客阿良_ALiang
@file : apschedule_tool.py
@ide : PyCharm
@time : 2022-03-02 17:34:17
"""
from apscheduler.schedulers.background import BackgroundScheduler
from multiprocessing import Process, Queue
import time
import random
# 具體工作實現(xiàn)
def do_job(q: Queue):
while True:
if not q.empty():
_value = q.get(False)
print('{} poll -> {}'.format(time.strftime('%Y-%m-%d %H:%M:%S', time.localtime()), _value))
else:
break
def put_job(q: Queue):
while True:
_value = str(random.randint(1, 10))
q.put(_value)
print('{} put -> {}'.format(time.strftime('%Y-%m-%d %H:%M:%S', time.localtime()), _value))
time.sleep(1)
if __name__ == '__main__':
q = Queue()
scheduler = BackgroundScheduler()
# 每隔5秒運行一次
scheduler.add_job(do_job, trigger='cron', second='*/5', args=(q,))
scheduler.start()
Process(target=put_job, args=(q,)).start()
代碼詳解
1、調(diào)度器的選擇主要取決于編程環(huán)境以及 APScheduler 的用途。主要有以下幾種調(diào)度器:
apscheduler.schedulers.blocking.BlockingScheduler:當(dāng)調(diào)度器是程序中唯一運行的東西時使用,阻塞式。
apscheduler.schedulers.background.BackgroundScheduler:當(dāng)調(diào)度器需要后臺運行時使用。
apscheduler.schedulers.asyncio.AsyncIOScheduler:當(dāng)程序使用 asyncio 框架時使用。
apscheduler.schedulers.gevent.GeventScheduler:當(dāng)程序使用 gevent 框架時使用。
apscheduler.schedulers.tornado.TornadoScheduler:當(dāng)構(gòu)建 Tornado 程序時使用
apscheduler.schedulers.twisted.TwistedScheduler:當(dāng)構(gòu)建 Twisted 程序時使用
apscheduler.schedulers.qt.QtScheduler:當(dāng)構(gòu)建 Qt 程序時使用
個人覺著BackgroundScheduler已經(jīng)很夠用了,在后臺啟動定時任務(wù),也不會阻塞進(jìn)程。
2、trigger后面跟隨的類似linux系統(tǒng)下cron寫法,樣例代碼中是每5秒執(zhí)行一次。
3、這里加了一個多進(jìn)程通訊的隊列multiprocessing.Queue,主要是樣例代碼解決的場景是我實際工作中常碰到的,舉個栗子:多個進(jìn)程間通訊,其中一個進(jìn)程需要定時獲取另一個進(jìn)程中的數(shù)據(jù)??梢詤⒖紭永a。
執(zhí)行結(jié)果
2022-03-02 19:31:27 put -> 4
2022-03-02 19:31:28 put -> 10
2022-03-02 19:31:29 put -> 1
2022-03-02 19:31:30 poll -> 4
2022-03-02 19:31:30 poll -> 10
2022-03-02 19:31:30 poll -> 1
2022-03-02 19:31:30 put -> 2
2022-03-02 19:31:31 put -> 1
2022-03-02 19:31:32 put -> 6
2022-03-02 19:31:33 put -> 4
2022-03-02 19:31:34 put -> 8
2022-03-02 19:31:35 poll -> 2
2022-03-02 19:31:35 poll -> 1
2022-03-02 19:31:35 poll -> 6
2022-03-02 19:31:35 poll -> 4
2022-03-02 19:31:35 poll -> 8
2022-03-02 19:31:35 put -> 8
2022-03-02 19:31:36 put -> 10
2022-03-02 19:31:37 put -> 7
2022-03-02 19:31:38 put -> 2
2022-03-02 19:31:39 put -> 3
2022-03-02 19:31:40 poll -> 8
2022-03-02 19:31:40 poll -> 10
2022-03-02 19:31:40 poll -> 7
2022-03-02 19:31:40 poll -> 2
2022-03-02 19:31:40 poll -> 3
2022-03-02 19:31:40 put -> 5
Process finished with exit code -1
知識點補充
APScheduler(advanceded python scheduler)基于Quartz的一個Python定時任務(wù)框架,實現(xiàn)了Quartz的所有功能,使用起來十分方便。提供了基于日期、固定時間間隔以及crontab類型的任務(wù),并且可以持久化任務(wù)?;谶@些功能,我們可以很方便的實現(xiàn)一個Python定時任務(wù)系統(tǒng)。
它有以下三個特點:
- 類似于 Liunx Cron 的調(diào)度程序(可選的開始/結(jié)束時間)
- 基于時間間隔的執(zhí)行調(diào)度(周期性調(diào)度,可選的開始/結(jié)束時間)
- 一次性執(zhí)行任務(wù)(在設(shè)定的日期/時間運行一次任務(wù))
APScheduler有四種組成部分:
- 觸發(fā)器(trigger) 包含調(diào)度邏輯,每一個作業(yè)有它自己的觸發(fā)器,用于決定接下來哪一個作業(yè)會運行。除了他們自己初始配置意外,觸發(fā)器完全是無狀態(tài)的。
- 作業(yè)存儲(job store) 存儲被調(diào)度的作業(yè),默認(rèn)的作業(yè)存儲是簡單地把作業(yè)保存在內(nèi)存中,其他的作業(yè)存儲是將作業(yè)保存在數(shù)據(jù)庫中。一個作業(yè)的數(shù)據(jù)講在保存在持久化作業(yè)存儲時被序列化,并在加載時被反序列化。調(diào)度器不能分享同一個作業(yè)存儲。
- 執(zhí)行器(executor) 處理作業(yè)的運行,他們通常通過在作業(yè)中提交制定的可調(diào)用對象到一個線程或者進(jìn)城池來進(jìn)行。當(dāng)作業(yè)完成時,執(zhí)行器將會通知調(diào)度器。
- 調(diào)度器(scheduler) 是其他的組成部分。你通常在應(yīng)用只有一個調(diào)度器,應(yīng)用的開發(fā)者通常不會直接處理作業(yè)存儲、調(diào)度器和觸發(fā)器,相反,調(diào)度器提供了處理這些的合適的接口。配置作業(yè)存儲和執(zhí)行器可以在調(diào)度器中完成,例如添加、修改和移除作業(yè)。通過配置executor、jobstore、trigger,使用線程池(ThreadPoolExecutor默認(rèn)值20)或進(jìn)程池(ProcessPoolExecutor 默認(rèn)值5)并且默認(rèn)最多3個(max_instances)任務(wù)實例同時運行,實現(xiàn)對job的增刪改查等調(diào)度控制

示例代碼:
from apscheduler.schedulers.blocking import BlockingScheduler
from datetime import datetime
# 輸出時間
def job():
print(datetime.now().strftime("%Y-%m-%d %H:%M:%S"))
# BlockingScheduler
sched = BlockingScheduler()
sched.add_job(my_job, 'interval', seconds=5, id='my_job_id')
sched.start()
到此這篇關(guān)于詳解Python利用APScheduler框架實現(xiàn)定時任務(wù)的文章就介紹到這了,更多相關(guān)Python定時任務(wù)內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家!
- python定時任務(wù)apscheduler的詳細(xì)使用教程
- python 基于Apscheduler實現(xiàn)定時任務(wù)
- Python定時任務(wù)框架APScheduler原理及常用代碼
- Python定時任務(wù)APScheduler安裝及使用解析
- Python定時任務(wù)APScheduler原理及實例解析
- Python使用APScheduler實現(xiàn)定時任務(wù)過程解析
- Python定時任務(wù)工具之APScheduler使用方式
- Python定時任務(wù)APScheduler的實例實例詳解
- 詳解Python使用apscheduler定時執(zhí)行任務(wù)
相關(guān)文章
Python(Django)項目與Apache的管理交互的方法
這篇文章主要介紹了Python(Django)項目與Apache的管理交互的方法,小編覺得挺不錯的,現(xiàn)在分享給大家,也給大家做個參考。一起跟隨小編過來看看吧2018-05-05
Python任務(wù)調(diào)度模塊APScheduler使用
這篇文章主要介紹了Python任務(wù)調(diào)度模塊APScheduler使用,文中通過示例代碼介紹的非常詳細(xì),對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價值,需要的朋友可以參考下2020-04-04
用TensorFlow實現(xiàn)lasso回歸和嶺回歸算法的示例
本篇文章主要介紹了用TensorFlow實現(xiàn)lasso回歸和嶺回歸算法的示例,小編覺得挺不錯的,現(xiàn)在分享給大家,也給大家做個參考。一起跟隨小編過來看看吧2018-05-05
Python的for和break循環(huán)結(jié)構(gòu)中使用else語句的技巧
平時我們把在if結(jié)構(gòu)中使用else語句當(dāng)作理所當(dāng)然,然而,Python強大的語法糖可以讓else語句在for和while循環(huán)中使用!下面我們就通過例子來看一下Python的for和break循環(huán)結(jié)構(gòu)中使用else語句的技巧2016-05-05
python3+selenium4實現(xiàn)切換窗口與iframe的方法
在自動化測試過程中,有時后會遇到元素定位方式?jīng)]有問題,但是依舊拋出無法找到元素的異常的問題,有可能是由于當(dāng)前焦點不在指定頁面或iframe導(dǎo)致的,本文就來說明 一下2021-05-05
Python計算元素在列表中出現(xiàn)的次數(shù)實例
本文介紹如何在Python中定義一個列表,并使用count()方法計算某個元素在列表中出現(xiàn)的次數(shù),示例中展示了具體的操作步驟和輸出結(jié)果2024-11-11

