APScheduler scheduler.shutdown(wait=True)未等待后台任务完成问题

我创建了两个BackgroundScheduler实例，启动调度器任务后调用scheduler.shutdown(wait=True)，但该方法并未等待任务执行完成。我通过编写while循环调用scheduler.get_jobs()检查待处理任务的方式临时解决了此问题，但根据APScheduler文档描述，shutdown(wait=True)参数的作用就是等待任务完成，我是否误解了该参数的功能？

相关代码 #

import time
import os
import random

from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.executors.pool import ThreadPoolExecutor
from datetime import datetime,timezone,timedelta
from zoneinfo import ZoneInfo

some_data = None # task1() updates it and task2() reads it

def time_now():
    return datetime.now(timezone.utc).astimezone(ZoneInfo('localtime'))
    
def time_now_str():
    return time_now().strftime('%y-%m-%d %H:%M:%S.%f')

def random_int():
    return random.randint(10000000000, 99999999999)

def task1(taskid):
    global some_data
    os.system('/usr/bin/dd if=/dev/zero of=/dev/null bs=10 count=10000 2> /dev/null') ## dummy task
    some_data =random_int()
    print(f'## Task1 ID: {taskid}, Generated some random data = {some_data}, '+time_now_str())
    return some_data

def task2(taskid):
    global some_data
    if some_data is not None:
        curr_data = some_data
        print(f'-- Task2 ID: {taskid}, STARTED task2 for data = {curr_data}: '+time_now_str())
        os.system('/usr/bin/dd if=/dev/zero of=/dev/null bs=10 count=5000000 2> /dev/null') ## dummy task
        print(f'-- Task2 ID: {taskid}, DONE task2 for data = {curr_data}: '+time_now_str())
    else:
        print(f'-- Task2 ID: {taskid}, WARNING, no data yet, looks like task1() has not run yet, '+time_now_str())

        
## main()
 
task1_scheduler = BackgroundScheduler(executors={'default': ThreadPoolExecutor(1)}, job_defaults={'misfire_grace_time': 500, 'coalesce': False})
task2_scheduler = BackgroundScheduler(executors={'default': ThreadPoolExecutor(1)}, job_defaults={'misfire_grace_time': 500, 'coalesce': False})
        
start_time = time_now()

print('Start Time: '+start_time.strftime('%y-%m-%d %H:%M:%S.%f'))

task1_tm = start_time
for idx, interval in enumerate( [1, 8, 14], start=1):
    task1_id = 'task1_id_'+str(idx)
    task1_tm += timedelta(seconds=interval)
    task1_scheduler.add_job(task1,'date',  run_date=task1_tm,  args=[task1_id], id=task1_id, max_instances=1)

task2_tm=start_time
for idx, interval in enumerate([0, 4,19,28], start=1):
    task2_id =  'task2_id_'+str(idx)
    task2_tm += timedelta(seconds=interval)
    task2_scheduler.add_job(task2,'date',  run_date=task2_tm, args=[task2_id], id =task2_id, max_instances=1)

task1_scheduler.start()
task2_scheduler.start()

# I should not need this
while len(task1_scheduler.get_jobs()) > 0 or len(task2_scheduler.get_jobs()) > 0:
    time.sleep(1)

task1_scheduler.shutdown(wait=True)
task2_scheduler.shutdown(wait=True)
    
print(f'Tasks completed: {time_now_str()}\n')

运行环境 #

• OS: Linux Mint 21.2
• APScheduler: v3.10.4
• Python: v3.10.12

解答 #

你没有误解shutdown(wait=True)的功能，问题出在调用时机和调度器的逻辑特性上：

1. get_jobs()的统计范围：scheduler.get_jobs()仅返回尚未被调度执行的任务，已经进入执行池的运行中任务不会被统计。你的while循环退出时，可能仍有任务在执行，但调度器已无待调度任务，此时调用shutdown(wait=True)其实无需等待；但如果没有while循环，启动调度器后立刻调用shutdown，调度器可能还没来得及把所有任务加入执行队列，wait=True只能等待当前正在执行的任务，无法等待未被调度的任务。

2. BackgroundScheduler的异步特性：BackgroundScheduler.start()是异步启动的，调度线程在后台运行。如果启动后立刻调用shutdown，调度器可能还未完成所有任务的调度工作，wait=True只能覆盖已启动的任务，无法覆盖还未进入调度流程的任务。

3. 优化方案：

   • 无需用while循环轮询get_jobs()，可以等待最后一个任务的预计执行结束时间后，再调用shutdown(wait=True)，确保所有任务都已进入执行流程。
   • 若业务允许，可改用BlockingScheduler，但多调度器场景下需将其放在独立线程中运行。
   • 利用调度器的事件机制，监听任务完成事件，待所有任务触发完成事件后再执行shutdown。

总结：shutdown(wait=True)确实会等待当前正在执行的任务完成，但不会等待尚未被调度的任务。你的核心问题是调用shutdown时，部分任务还未进入调度流程，因此需要先确保所有任务都已被调度，再执行shutdown。

内容的提问来源于stack exchange，提问作者Rajiv Kumar