背景

最近在做一些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)文章希望大家以后多多支持腳本之家！