詳解Python利用APScheduler框架實現定時任務

Posted on by

背景

最近在做一些python工具的時候，常常會碰到定時器問題，總覺著使用threading.timer或者schedule模塊非常不優雅。所以這裡給自己做個記錄，也分享一個定時任務框架APScheduler。具體的架構原理就不細說瞭，用個例子說明一下怎麼簡易的使用。

樣例代碼

先上樣例代碼，如下：

#!/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
 
 
# 具體工作實現
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、調度器的選擇主要取決於編程環境以及 APScheduler 的用途。主要有以下幾種調度器：

apscheduler.schedulers.blocking.BlockingScheduler：當調度器是程序中唯一運行的東西時使用，阻塞式。

apscheduler.schedulers.background.BackgroundScheduler：當調度器需要後臺運行時使用。

apscheduler.schedulers.asyncio.AsyncIOScheduler：當程序使用 asyncio 框架時使用。

apscheduler.schedulers.gevent.GeventScheduler：當程序使用 gevent 框架時使用。

apscheduler.schedulers.tornado.TornadoScheduler：當構建 Tornado 程序時使用

apscheduler.schedulers.twisted.TwistedScheduler：當構建 Twisted 程序時使用

apscheduler.schedulers.qt.QtScheduler：當構建 Qt 程序時使用

個人覺著BackgroundScheduler已經很夠用瞭，在後臺啟動定時任務，也不會阻塞進程。

2、trigger後面跟隨的類似linux系統下cron寫法，樣例代碼中是每5秒執行一次。

3、這裡加瞭一個多進程通訊的隊列multiprocessing.Queue，主要是樣例代碼解決的場景是我實際工作中常碰到的，舉個栗子：多個進程間通訊，其中一個進程需要定時獲取另一個進程中的數據。可以參考樣例代碼。

執行結果

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定時任務框架，實現瞭Quartz的所有功能，使用起來十分方便。提供瞭基於日期、固定時間間隔以及crontab類型的任務，並且可以持久化任務。基於這些功能，我們可以很方便的實現一個Python定時任務系統。

它有以下三個特點：

  • 類似於 Liunx Cron 的調度程序(可選的開始/結束時間)
  • 基於時間間隔的執行調度(周期性調度，可選的開始/結束時間)
  • 一次性執行任務(在設定的日期/時間運行一次任務)

APScheduler有四種組成部分：

  • 觸發器(trigger) 包含調度邏輯，每一個作業有它自己的觸發器，用於決定接下來哪一個作業會運行。除瞭他們自己初始配置意外，觸發器完全是無狀態的。
  • 作業存儲(job store) 存儲被調度的作業，默認的作業存儲是簡單地把作業保存在內存中，其他的作業存儲是將作業保存在數據庫中。一個作業的數據講在保存在持久化作業存儲時被序列化，並在加載時被反序列化。調度器不能分享同一個作業存儲。
  • 執行器(executor) 處理作業的運行，他們通常通過在作業中提交制定的可調用對象到一個線程或者進城池來進行。當作業完成時，執行器將會通知調度器。
  • 調度器(scheduler) 是其他的組成部分。你通常在應用隻有一個調度器，應用的開發者通常不會直接處理作業存儲、調度器和觸發器，相反，調度器提供瞭處理這些的合適的接口。配置作業存儲和執行器可以在調度器中完成，例如添加、修改和移除作業。通過配置executor、jobstore、trigger，使用線程池(ThreadPoolExecutor默認值20)或進程池(ProcessPoolExecutor 默認值5)並且默認最多3個(max_instances)任務實例同時運行，實現對job的增刪改查等調度控制

示例代碼：

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()

到此這篇關於詳解Python利用APScheduler框架實現定時任務的文章就介紹到這瞭,更多相關Python定時任務內容請搜索WalkonNet以前的文章或繼續瀏覽下面的相關文章希望大傢以後多多支持WalkonNet！

推薦閱讀：