日本免费高清视频-国产福利视频导航-黄色在线播放国产-天天操天天操天天操天天操|www.shdianci.com

學無先后，達者為師

網站首頁 編程語言 正文

Python?APScheduler?定時任務詳解_python

作者：愛吃辣椒的鍋包肉 ? 更新時間： 2022-07-25 編程語言

一、基本概念

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 是最基本的一種調度，作業任務只會執行一次。它表示特定的時間點觸發。它的參數如下:

參數 說明
run_date (datetime 或 str) 作業的運行日期或時間
timezone (datetime.tzinfo 或 str) 指定時區 
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 周期觸發任務

參數 說明
weeks (int) 間隔幾周
days (int) 間隔幾天
hours (int) 間隔幾小時
minutes (int) 間隔幾分鐘
seconds (int) 間隔多少秒
start_date (datetime 或 str) 開始日期
end_date (datetime 或 str) 結束日期
timezone (datetime.tzinfo 或str) 時區 
@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格式兼容。

它是功能最強大的觸發器

參數 說明
year (int 或 str) 年，4位數字
month (int 或 str) 月 (范圍1-12)
day (int 或 str) 日 (范圍1-31)
week (int 或 str) 周 (范圍1-53)
day_of_week (int 或 str) 周內第幾天或者星期幾 (范圍0-6 或者 mon,tue,wed,thu,fri,sat,sun)
hour (int 或 str) 時 (范圍0-23)
minute (int 或 str) 分 (范圍0-59)
second (int 或 str) 秒 (范圍0-59)
start_date (datetime 或 str) 最早開始日期(包含)
end_date (datetime 或 str) 最晚結束時間(包含)
timezone (datetime.tzinfo 或str) 指定時區
表達式 參數類型 描述
* 所有 通配符。例：minutes=*即每分鐘觸發
*/a 所有 可被a整除的通配符
a-b 所有 范圍a-b觸發
a-b/c 所有 范圍a-b，且可被c整除時觸發
xth y 日 第幾個星期幾觸發。x為第幾個，y為星期幾
last x 日 一個月中，最后個星期幾觸發
last 日 一個月最后一天觸發
x,y,z 所有 組合表達式，可以組合確定值或上方的表達式 
 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()

原文鏈接：https://blog.csdn.net/weixin_44301439/article/details/124062178

欄目分類
最近更新