最新Python APScheduler 定時任務詳解

一、基本概念

APScheduler全稱Advanced Python Scheduler 作用為在指定的時間規則執行指定的作業。

• 指定時間規則的方式可以是間隔多久執行，可以是指定日期時間的執行，也可以類似Linux系統中Crontab中的方式執行任務。
• 指定的任務就是一個Python函數。

1.1、 觸發器：triggers

用於設定觸發任務的條件: 觸發器包含排程邏輯。每個任務都有自己的觸發器，用於確定何時應該執行作業。除了初始設定之外，觸發器完全是無狀態的

1.2、作業記憶體:job stores

用於存放任務，把任務存放在記憶體或資料庫中

• 預設情況下，任務存放在記憶體中。也可以設定存放在不同型別的資料庫中。如果任務存放在資料庫中，那麼任務的存取有一個序列化和反序列化的過程，同時修改和搜尋任務的功能也是由任務儲存器實現。
• 注意一個任務儲存器不要共用給多個排程器，否則會導致狀態混亂

1.3、執行器 executors

用於執行任務，可以設定執行模式為單執行緒或執行緒池：任務會被執行器放入執行緒池或程序池去執行，執行完畢後，執行器會通知排程器。

1.4、排程器 schedulers

把上方三個元件作為引數，通過建立排程器範例來執行:一個排程器由上方三個元件構成，一般來說，一個程式只要有一個排程器就可以了。開發者也不必直接操作任務儲存器、執行器以及觸發器，因為排程器提供了統一的介面，通過排程器就可以操作元件，比如任務的增刪改查。

二、排程器詳解

• BlockingScheduler : 阻塞式排程器：適用於只跑排程器的程式。
• BackgroundScheduler: 後臺排程器：適用於非阻塞的情況，排程器會在後臺獨立執行
• AsyncIOScheduler : AsyncIO排程器，適用於應用使用AsnycIO的情況。
• GeventScheduler : Gevent排程器，適用於應用通過Gevent的情況。
• TornadoScheduler: Tornado排程器，適用於構建Tornado應用。
• TwistedScheduler:Twisted排程器，適用於構建Twisted應用。
• QtScheduler: Qt排程器，適用於構建Qt應用。

2.1、APScheduler有三種內建的觸發器

• date:日期：觸發任務執行的具體日期
• interval: 間隔：觸發任務執行的時間間隔
• cron: 週期：觸發任務執行的週期

2.2、觸發器公共引數

• id:啟動任務的ID具有唯一性
• name： 設定啟動任務的名稱
• coalesce :當由於某種原因導致某個job積攢了好幾次沒有實際執行（比如說系統掛了5分鐘後恢復，有一個任務是每分鐘跑一次的，按道理說這5分鐘內本來是“計劃”執行5次的，但實際沒有執行），如果coalesce為True，下次這個job被submit給executor時，只會執行1次，也就是最後這次，如果為False，那麼會執行5次（不一定，因為還有其他條件，看後面misfire_grace_time的解釋）
• max_instance: 就是說同一個job同一時間最多有幾個範例再跑，比如一個耗時10分鐘的job，被指定每分鐘執行1次，如果我們max_instance值為5，那麼在第6~10分鐘上，新的執行範例不會被執行，因為已經有5個範例在跑了
• misfire_grace_time:設想和上述coalesce類似的場景，如果一個job本來14:00有一次執行，但是由於某種原因沒有被排程上，現在14:01了，這個14:00的執行範例被提交時，會檢查它預訂執行的時間和當下時間的差值（這裡是1分鐘），大於我們設定的30秒限制，那麼這個執行範例不會被執行。
• replace_existing: 如果排程的job在一個持久化的記憶體裡，當初始化應用程式時，必須要為job定義一個顯示的ID並使用replace_existing=True, 否則每次應用程式重啟時都會得到那個job的一個新副本

2.3、date內建觸發器

date 是最基本的一種排程，作業任務只會執行一次。它表示特定的時間點觸發。它的引數如下:

from datetime import datetime
from datetime import date
from apscheduler.schedulers.blocking import BlockingScheduler
def job(text):
    print(text)
scheduler = BlockingScheduler()
# 在 2019-8-30 執行一次 job 方法
scheduler.add_job(job, 'date', run_date=date(2022, 4, 9), args=['text1'], id="1", coalesce=True, max_instances=1)
# 在 2019-8-30 01:00:00 執行一次 job 方法
scheduler.add_job(job, 'date', run_date=datetime(2022, 4, 9, 17, 40, 58), args=['text2'], id="2", coalesce=True, max_instances=1)
# 在 2019-8-30 01:00:01 執行一次 job 方法
scheduler.add_job(job, 'date', run_date='2022-4-9 17:41:00', args=['text3'], id="3", coalesce=True, max_instances=1)
scheduler.start()

2.4、interval 週期觸發任務

@sched.scheduled_job(
    "interval", id=spider_job_name + "_bg_data", coalesce=True, max_instances=1, minutes=20
)
def tick_rzjg_detail_xq():
    """
    快速完成
    :return:
    """
    each = "rzjg_bg_data"
    cmd_str = f"cd {ROOT} && bash run_spider.sh {each} --loglevel=INFO"
    print(cmd_str)
    os.system(cmd_str)
def func():
    print("Press Ctrl+C to exit")
    # 直接觸發一次
    tick_rzjg_detail_xq()
    try:
        sched.start()
    except (KeyboardInterrupt, SystemExit):
        pass

if __name__ == "__main__":
    func()

2.5、cron 觸發器 在特定時間週期性地觸發，和Linux crontab格式相容。

它是功能最強大的觸發器

 import time
    from apscheduler.schedulers.blocking import BlockingScheduler

    def job(text):
        t = time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time()))
        print('{} --- {}'.format(text, t))

    scheduler = BlockingScheduler()
    # 在每天22點，每隔 1分鐘 執行一次 job 方法
    scheduler.add_job(job, 'cron', hour=22, minute='*/1', args=['job1'])
    # 在每天22和23點的25分，執行一次 job 方法
    scheduler.add_job(job, 'cron', hour='22-23', minute='25', args=['job2'])
    # 在每天 8 點，執行一次 job 方法
    scheduler.add_job(job, 'cron', hour='8', args=['job2'])
    # 在每天 8 點 20點，各執行一次 job 方法    設定最大執行範例數
    scheduler.add_job(job, 'cron', hour='8, 20', minute=30, max_instances=4)
    scheduler.start()

到此這篇關於Python APScheduler 定時任務詳解的文章就介紹到這了,更多相關Python APScheduler 定時任務內容請搜尋it145.com以前的文章或繼續瀏覽下面的相關文章希望大家以後多多支援it145.com！