詳解APScheduler如何設置任務不並發

Posted on by

1.軟件環境

Windows10 教育版64位
Python 3.6.3
APScheduler 3.6.3

2.問題描述

Python中定時任務的解決方案，總體來說有四種，分別是：crontab、 scheduler、 Celery、APScheduler，其中：

  • crontab是 Linux 的一個定時任務管理工具，在Windows上面有替代品pycron，但Windows不像 Linux那樣有很多強大的命令程序，pycron使用起來有局限性，定制性不好；
  • Scheduler太過於簡單、復雜一點的定時任務做起來太困難，特別是以月份以上時間單位的定時任務；
  • Celery依賴的軟件比較多，比較耗資源；
  • APScheduler（Advanced Python Scheduler） 基於 Quartz，可以跨平臺而且配置方便，提供瞭date、interval、cron3種不同的觸發器，與Linux上原生的 crontab 格式兼容，可以設置任何高度復雜的定時任務，靈活的要死。

在此不介紹APScheduler的基本特性，有需要的可以直接去看APScheduler官方文檔，我們直接切到主題：

APScheduler如何設置任務不並發（即第一個任務執行完再執行下一個）？

APScheduler在多個任務相同時間點同時被觸發時，會同時並發執行多個任務，如使用下方的示例代碼：

'''
===========================================
  @author:  jayce
  @file:    apscheduler設置任務不並發.py         
  @time:    2022/7/1/001   19:38 
===========================================
'''
from apscheduler.schedulers.blocking import BlockingScheduler
import time


def job_printer(text):
    '''
    死循環，用來模擬長時間執行的任務
    :param text: 
    :return: 
    '''
    while True:
        time.sleep(2)
        print("job text：{}".format(text))


if __name__ == '__main__':
    schedule = BlockingScheduler()

    schedule.add_job(job_printer, "cron", second='*/10', args=['每10秒執行一次！'])
    schedule.add_job(job_printer, "cron", second='*/20', args=['每20秒執行一次！'])
 
    schedule.print_jobs()
    schedule.start()

可以看到，函數job_printer是一個死循環，用來模擬長時間執行的任務，我們使用add_job向APScheduler中添加2個job_printer，區別是2個任務的時間間隔為：每10秒執行一次和每20秒執行一次。
因為job_printer是一個死循環，相當於job_printer一直沒有被執行完，但其實APScheduler在任務沒有被執行完的情況下，同時執行多個不同的job_printer：

job text：每10秒執行一次！
job text：每20秒執行一次！
job text：每10秒執行一次！
job text：每20秒執行一次！
job text：每10秒執行一次！
job text：每20秒執行一次！
job text：每10秒執行一次！
job text：每20秒執行一次！
job text：每10秒執行一次！
Execution of job "job_printer (trigger: cron[second='*/10'], next run at: 2022-07-01 20:47:50 CST)" skipped: maximum number of running instances reached (1)

即：

在這裡插入圖片描述

可以看到10秒的job_printer和20秒的job_printer交替被執行，而其實10秒的job_printer其實根本沒有執行完。這在CPU或者GPU等硬件設備能夠承擔負載的情況下，當然是好事，但如果你的硬件不夠的話，發生OOM等資源不夠的情況，程序就被中斷瞭，導致你的模型訓練或業務邏輯失敗！
具體的：
我這邊是使用APScheduler和Tensorflow進行在線學習（online learning）時，在不同的時間節點下會對模型使用不一樣的重訓練方式，如有2個定時任務（A：每10秒執行一次，B：每20秒執行一次）和2種重訓練方式（X和Y），當你的顯存存在如下情況：

顯存很少隻夠一個程序進行訓練，不能多個程序同時運行，否則會OOM；

那麼隻能引導程序依次執行，而不能並發執行，等當同一時間內X和Y同時被觸發時，隻執行其中1個，另外1個不執行。

那這個時候又該怎麼辦呢

3.解決方法

通過查閱官方文檔，發現可以通過設置執行任務的線程數，來控制隻有1個執行器進行任務的執行，進而達到執行完任務X再執行任務Y，具體如下：

'''
===========================================
  @author:  jayce
  @file:    apscheduler設置任務不並發.py         
  @time:    2022/7/1/001   19:38 
===========================================
'''
from apscheduler.executors.pool import ThreadPoolExecutor


if __name__ == '__main__':
    # 為瞭防止全量和增量並發造成顯存溢出，進而訓練失敗，設置同一時間隻能有一個任務運行
    schedule = BlockingScheduler(executors={'default': ThreadPoolExecutor(1)})

通過向BlockingScheduler設定最大的ThreadPoolExecutor=1，即可達到我們想要的效果!

4.結果預覽

job text：每10秒執行一次！
job text：每10秒執行一次！
job text：每10秒執行一次！
job text：每10秒執行一次！
job text：每10秒執行一次！
Execution of job "job_printer (trigger: cron[second='*/10'], next run at: 2022-07-01 21:17:50 CST)" skipped: maximum number of running instances reached (1)
job text：每10秒執行一次！
job text：每10秒執行一次！
job text：每10秒執行一次！
job text：每10秒執行一次！
job text：每10秒執行一次！
Execution of job "job_printer (trigger: cron[second='*/10'], next run at: 2022-07-01 21:18:00 CST)" skipped: maximum number of running instances reached (1)
Execution of job "job_printer (trigger: cron[second='*/20'], next run at: 2022-07-01 21:18:00 CST)" skipped: maximum number of running instances reached (1)

即：

在這裡插入圖片描述

可以看到，一直在執行第1個被觸發的任務，相同時間被觸發的任務都被skipped瞭~~
當然，如果你想要第1個任務執行完時，執行被跳過的任務，可以通過在add_job中設置misfire_grace_time實現！

FAQ

1.APScheduler如果某個任務掛掉瞭，整個定時任務程序會中斷嗎？還是下次時間繼續執行該任務？

答案是：程序不會中斷，到下次執行任務的時間點，還會重新執行。
具體的，使用如下測試代碼：

'''
===========================================
  @author:  jayce
  @file:    apscheduler設置任務不並發.py         
  @time:    2022/7/1/001   19:38 
===========================================
'''
from apscheduler.schedulers.blocking import BlockingScheduler
from apscheduler.executors.pool import ThreadPoolExecutor
import time


def exception_maker():
    '''
    異常制造器，用來模擬任務執行被中斷
    :return:
    '''
    return 1 / 0


def job_printer(text):
    '''
    死循環，用來模擬長時間執行的任務
    :param text:
    :return:
    '''
    while True:
        time.sleep(2)
        print("job text：{}".format(text))


if __name__ == '__main__':
    schedule = BlockingScheduler()

    schedule.add_job(job_printer, "cron", second='*/10', args=['每10秒執行一次！'])
    schedule.add_job(exception_maker, "cron", second='*/5')

    schedule.print_jobs()
    schedule.start()

可以看到exception_maker已經失敗多次，但是不影響其他任務和它自身的下次執行：

Job "exception_maker (trigger: cron[second='*/5'], next run at: 2022-07-01 19:53:30 CST)" raised an exception
Traceback (most recent call last):
  File "C:\Users\Jayce\Anaconda3\envs\tf2.3\lib\site-packages\apscheduler\executors\base.py", line 125, in run_job
    retval = job.func(*job.args, **job.kwargs)
  File "E:/Code/Python/demo代碼/apscheduler設置任務不並發.py", line 14, in exception_maker
    return 1 / 0
ZeroDivisionError: division by zero
Job "exception_maker (trigger: cron[second='*/5'], next run at: 2022-07-01 19:53:35 CST)" raised an exception
Traceback (most recent call last):
  File "C:\Users\Jayce\Anaconda3\envs\tf2.3\lib\site-packages\apscheduler\executors\base.py", line 125, in run_job
    retval = job.func(*job.args, **job.kwargs)
  File "E:/Code/Python/demo代碼/apscheduler設置任務不並發.py", line 14, in exception_maker
    return 1 / 0
ZeroDivisionError: division by zero
job text：每10秒執行一次！
job text：每10秒執行一次！
Job "exception_maker (trigger: cron[second='*/5'], next run at: 2022-07-01 19:53:40 CST)" raised an exception
Traceback (most recent call last):
  File "C:\Users\Jayce\Anaconda3\envs\tf2.3\lib\site-packages\apscheduler\executors\base.py", line 125, in run_job
    retval = job.func(*job.args, **job.kwargs)
  File "E:/Code/Python/demo代碼/apscheduler設置任務不並發.py", line 14, in exception_maker
    return 1 / 0
ZeroDivisionError: division by zero
job text：每10秒執行一次！
job text：每10秒執行一次！
Execution of job "job_printer (trigger: cron[second='*/10'], next run at: 2022-07-01 19:53:40 CST)" skipped: maximum number of running instances reached (1)
Job "exception_maker (trigger: cron[second='*/5'], next run at: 2022-07-01 19:53:45 CST)" raised an exception
Traceback (most recent call last):
  File "C:\Users\Jayce\Anaconda3\envs\tf2.3\lib\site-packages\apscheduler\executors\base.py", line 125, in run_job
    retval = job.func(*job.args, **job.kwargs)
  File "E:/Code/Python/demo代碼/apscheduler設置任務不並發.py", line 14, in exception_maker
    return 1 / 0
ZeroDivisionError: division by zero
job text：每10秒執行一次！

即：

在這裡插入圖片描述

到此這篇關於詳解APScheduler如何設置任務不並發的文章就介紹到這瞭,更多相關APScheduler 任務不並發內容請搜索WalkonNet以前的文章或繼續瀏覽下面的相關文章希望大傢以後多多支持WalkonNet！

推薦閱讀：