快捷導(dǎo)航

詳解APScheduler如何設(shè)置任務(wù)不并發(fā)

更新時間：2022年07月19日 11:39:34 作者：Jayce~

本文主要介紹了APScheduler如何設(shè)置任務(wù)不并發(fā)，文中通過示例代碼介紹的非常詳細(xì)，對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價值，需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧

1.軟件環(huán)境

Windows10 教育版64位
Python 3.6.3
APScheduler 3.6.3

2.問題描述

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

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

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

APScheduler如何設(shè)置任務(wù)不并發(fā)（即第一個任務(wù)執(zhí)行完再執(zhí)行下一個）？

APScheduler在多個任務(wù)相同時間點同時被觸發(fā)時，會同時并發(fā)執(zhí)行多個任務(wù)，如使用下方的示例代碼：

'''
===========================================
  @author:  jayce
  @file:    apscheduler設(shè)置任務(wù)不并發(fā).py         
  @time:    2022/7/1/001   19:38 
===========================================
'''
from apscheduler.schedulers.blocking import BlockingScheduler
import time


def job_printer(text):
    '''
    死循環(huán)，用來模擬長時間執(zhí)行的任務(wù)
    :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秒執(zhí)行一次！'])
    schedule.add_job(job_printer, "cron", second='*/20', args=['每20秒執(zhí)行一次！'])
 
    schedule.print_jobs()
    schedule.start()

可以看到，函數(shù)job_printer是一個死循環(huán)，用來模擬長時間執(zhí)行的任務(wù)，我們使用add_job向APScheduler中添加2個job_printer，區(qū)別是2個任務(wù)的時間間隔為：每10秒執(zhí)行一次和每20秒執(zhí)行一次。
因為job_printer是一個死循環(huán)，相當(dāng)于job_printer一直沒有被執(zhí)行完，但其實APScheduler在任務(wù)沒有被執(zhí)行完的情況下，同時執(zhí)行多個不同的job_printer：

job text：每10秒執(zhí)行一次！
job text：每20秒執(zhí)行一次！
job text：每10秒執(zhí)行一次！
job text：每20秒執(zhí)行一次！
job text：每10秒執(zhí)行一次！
job text：每20秒執(zhí)行一次！
job text：每10秒執(zhí)行一次！
job text：每20秒執(zhí)行一次！
job text：每10秒執(zhí)行一次！
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交替被執(zhí)行，而其實10秒的job_printer其實根本沒有執(zhí)行完。這在CPU或者GPU等硬件設(shè)備能夠承擔(dān)負(fù)載的情況下，當(dāng)然是好事，但如果你的硬件不夠的話，發(fā)生OOM等資源不夠的情況，程序就被中斷了，導(dǎo)致你的模型訓(xùn)練或業(yè)務(wù)邏輯失??！
具體的：
我這邊是使用APScheduler和Tensorflow進(jìn)行在線學(xué)習(xí)（online learning）時，在不同的時間節(jié)點下會對模型使用不一樣的重訓(xùn)練方式，如有2個定時任務(wù)（A：每10秒執(zhí)行一次，B：每20秒執(zhí)行一次）和2種重訓(xùn)練方式（X和Y），當(dāng)你的顯存存在如下情況：

顯存很少只夠一個程序進(jìn)行訓(xùn)練，不能多個程序同時運行，否則會OOM；

那么只能引導(dǎo)程序依次執(zhí)行，而不能并發(fā)執(zhí)行，等當(dāng)同一時間內(nèi)X和Y同時被觸發(fā)時，只執(zhí)行其中1個，另外1個不執(zhí)行。

那這個時候又該怎么辦呢

3.解決方法

通過查閱官方文檔，發(fā)現(xiàn)可以通過設(shè)置執(zhí)行任務(wù)的線程數(shù)，來控制只有1個執(zhí)行器進(jìn)行任務(wù)的執(zhí)行，進(jìn)而達(dá)到執(zhí)行完任務(wù)X再執(zhí)行任務(wù)Y，具體如下：

'''
===========================================
  @author:  jayce
  @file:    apscheduler設(shè)置任務(wù)不并發(fā).py         
  @time:    2022/7/1/001   19:38 
===========================================
'''
from apscheduler.executors.pool import ThreadPoolExecutor


if __name__ == '__main__':
    # 為了防止全量和增量并發(fā)造成顯存溢出，進(jìn)而訓(xùn)練失敗，設(shè)置同一時間只能有一個任務(wù)運行
    schedule = BlockingScheduler(executors={'default': ThreadPoolExecutor(1)})

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

4.結(jié)果預(yù)覽

job text：每10秒執(zhí)行一次！
job text：每10秒執(zhí)行一次！
job text：每10秒執(zhí)行一次！
job text：每10秒執(zhí)行一次！
job text：每10秒執(zhí)行一次！
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秒執(zhí)行一次！
job text：每10秒執(zhí)行一次！
job text：每10秒執(zhí)行一次！
job text：每10秒執(zhí)行一次！
job text：每10秒執(zhí)行一次！
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)

即：

在這里插入圖片描述

可以看到，一直在執(zhí)行第1個被觸發(fā)的任務(wù)，相同時間被觸發(fā)的任務(wù)都被skipped了~~
當(dāng)然，如果你想要第1個任務(wù)執(zhí)行完時，執(zhí)行被跳過的任務(wù)，可以通過在add_job中設(shè)置misfire_grace_time實現(xiàn)！

FAQ

1.APScheduler如果某個任務(wù)掛掉了，整個定時任務(wù)程序會中斷嗎？還是下次時間繼續(xù)執(zhí)行該任務(wù)？

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

'''
===========================================
  @author:  jayce
  @file:    apscheduler設(shè)置任務(wù)不并發(fā).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():
    '''
    異常制造器，用來模擬任務(wù)執(zhí)行被中斷
    :return:
    '''
    return 1 / 0


def job_printer(text):
    '''
    死循環(huán)，用來模擬長時間執(zhí)行的任務(wù)
    :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秒執(zhí)行一次！'])
    schedule.add_job(exception_maker, "cron", second='*/5')

    schedule.print_jobs()
    schedule.start()

可以看到exception_maker已經(jīng)失敗多次，但是不影響其他任務(wù)和它自身的下次執(zhí)行：

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設(shè)置任務(wù)不并發(fā).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設(shè)置任務(wù)不并發(fā).py", line 14, in exception_maker
return 1 / 0
ZeroDivisionError: division by zero
job text：每10秒執(zhí)行一次！
job text：每10秒執(zhí)行一次！
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設(shè)置任務(wù)不并發(fā).py", line 14, in exception_maker
return 1 / 0
ZeroDivisionError: division by zero
job text：每10秒執(zhí)行一次！
job text：每10秒執(zhí)行一次！
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設(shè)置任務(wù)不并發(fā).py", line 14, in exception_maker
return 1 / 0
ZeroDivisionError: division by zero
job text：每10秒執(zhí)行一次！