如何通过Ctrl+C立即终止apscheduler.BlockingScheduler进程？

背景 #

Python 3.8，APScheduler 3.10.1

使用BlockingScheduler运行多个定时任务，希望通过Ctrl+C即时终止调度器进程，但当前代码未达到预期效果。

调度器基类代码 #

import traceback
from typing import Any, Callable
from apscheduler.schedulers.background import BlockingScheduler
from apscheduler.events import EVENT_JOB_EXECUTED, EVENT_JOB_ERROR
from utils.tool.decorator import sched_job
from utils.tool.logger import log

logger = log(__file__, "utils", warning_only=False)


class APSchedulerBase:
    def __init__(self, scheduler: Any = None, add_listener: bool = True):
        if scheduler is None:
            self.sched = BlockingScheduler(timezone="Asia/Shanghai")
            if add_listener:
                self.sched.add_listener(self.ap_listener, EVENT_JOB_ERROR | EVENT_JOB_EXECUTED)
        else:
            self.sched = scheduler

    def ap_listener(self, event):
        if event.exception == KeyboardInterrupt or event.exception == SystemExit:
            self.on_kill()
        if event.exception:
            logger.error(str(traceback.format_exc()))
        else:
            pass

    def on_kill(self):
        logger.warning("Scheduler Exit.")
        self.sched.shutdown(wait=False)

    def register_task(
            self, func: Callable, trigger: str,
            **kwargs
    ):
        @sched_job(
            self.sched,
            logger,
            trigger=trigger,
            **kwargs
        )
        def wrap():
            func()
        wrap()

    def start(self):
        try:
            logger.warning("Scheduler Start.")
            self.sched.start()
        except (KeyboardInterrupt, SystemExit):
            self.on_kill()

示例代码 #

Ctrl+C仅在f()即将执行时生效：

import time
def f():
   print(1)
   time.sleep(10)
   print(2)
s = APSchedulerBase()
s.register_task(f, trigger='cron', hour=12, minute=30)
s.start()

当前问题 #

仅在任务即将执行时按下Ctrl+C才能终止进程，若进程处于休眠/等待下一次任务触发状态时，Ctrl+C无响应，需等到下一次任务触发才会生效。如何实现Ctrl+C即时终止？

问题根源是BlockingScheduler内部的等待逻辑会进入深度休眠，导致信号无法被及时捕获。以下两种方法可以解决：

方法1：启用守护线程配置 #

修改Scheduler初始化时的daemon参数，让调度线程成为守护线程，确保信号能被主线程及时处理：

def __init__(self, scheduler: Any = None, add_listener: bool = True):
    if scheduler is None:
        # 添加daemon=True参数
        self.sched = BlockingScheduler(timezone="Asia/Shanghai", daemon=True)
        if add_listener:
            self.sched.add_listener(self.ap_listener, EVENT_JOB_ERROR | EVENT_JOB_EXECUTED)
    else:
        self.sched = scheduler

方法2：手动注册信号处理函数 #

直接为SIGINT（Ctrl+C）和SIGTERM信号注册处理函数，绕过APScheduler内部的等待逻辑，确保信号被即时响应：

首先导入signal模块，然后修改基类：

import traceback
import signal  # 新增导入
from typing import Any, Callable
from apscheduler.schedulers.background import BlockingScheduler
from apscheduler.events import EVENT_JOB_EXECUTED, EVENT_JOB_ERROR
from utils.tool.decorator import sched_job
from utils.tool.logger import log

logger = log(__file__, "utils", warning_only=False)


class APSchedulerBase:
    def __init__(self, scheduler: Any = None, add_listener: bool = True):
        if scheduler is None:
            self.sched = BlockingScheduler(timezone="Asia/Shanghai")
            if add_listener:
                self.sched.add_listener(self.ap_listener, EVENT_JOB_ERROR | EVENT_JOB_EXECUTED)
        else:
            self.sched = scheduler
        # 注册信号处理函数
        signal.signal(signal.SIGINT, self.handle_signal)
        signal.signal(signal.SIGTERM, self.handle_signal)

    def handle_signal(self, signum, frame):
        logger.warning(f"收到信号 {signum}，正在关闭调度器")
        self.on_kill()
        exit(0)

    # 其余方法保持不变...

关键说明 #

• 方法2的可靠性更高，因为它直接接管了信号处理流程，无论调度器处于等待任务还是执行任务的状态，都能立即响应Ctrl+C。
• 原代码中的try-except块仅在start()方法调用初期生效，一旦进入Scheduler内部的无限循环，该异常捕获逻辑就无法再捕获信号，因此必须通过信号注册的方式处理。

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