Python APScheduler 如何捕获回调方法中抛出的异常

问题根因 #

你当前的写法无法捕获异常的核心原因是：BackgroundScheduler的任务执行在独立的后台线程中，sched.start() 调用后会立即返回，你包裹start()的try...except 只能捕获调度器启动阶段的异常，完全触达不到后续任务线程内抛出的错误。

解决方案1：在回调函数内部直接捕获异常 #

这是最直接的方案，适合只需处理单个任务异常的场景，直接在任务逻辑外层套异常捕获即可：

def expiry_callback():
    try:
        raise ValueError
        print("inside job")
    except ValueError as e:
        print(f'ran into an issue!! {e}')

解决方案2：使用APScheduler事件监听器全局捕获异常 #

如果有大量任务需要统一处理异常，不需要给每个任务单独加捕获逻辑，可以用APScheduler自带的事件监听机制，监听任务执行失败的事件统一处理：

import time
from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.events import EVENT_JOB_ERROR, EVENT_JOB_EXECUTED


def expiry_callback():
    raise ValueError
    print("inside job")

# 定义全局异常监听器
def job_exception_listener(event):
    if event.exception:
        # 可获取异常对象、任务ID、堆栈信息等完整上下文
        print(f'任务 {event.job_id} 运行出错!! 异常信息: {event.exception}')


sched = BackgroundScheduler(daemon=True)
# 注册监听器，监听任务执行完成、执行错误事件
sched.add_listener(job_exception_listener, EVENT_JOB_EXECUTED | EVENT_JOB_ERROR)
sched.add_job(expiry_callback,'interval',seconds=1)

sched.start()

try:
    while True:
        time.sleep(5)
except (KeyboardInterrupt, SystemExit):
    sched.shutdown()

内容的提问来源于stack exchange，提问作者Its Andrew