You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

如何让Python APScheduler触发类实例中的函数?

基于BackgroundScheduler的自定义任务调度类

我最近写了一个无特殊框架依赖的类,核心是把BackgroundScheduler作为类属性来管理定时任务,基础代码结构如下:

from apscheduler.schedulers.background import BackgroundScheduler

class MyClass(object):
    def __init__(self, ...):
        # 这里可以写其他初始化逻辑
        self.Scheduler = BackgroundScheduler()
        # 可选:在这里直接启动调度器,或者后续通过单独方法启动
        # self.Scheduler.start()

这个类里我定义了两个核心方法,用来完成任务调度配置和任务触发处理的闭环:

一、根据传入参数配置并添加调度任务

这个方法负责接收外部传入的调度规则,把任务和处理函数绑定后,添加到调度器中:

def _schedule_Events(self, *args, **kwargs):
    try:
        # 复制传入的调度配置,避免修改原参数
        Schedule_Def = kwargs.copy()
        # 将任务处理函数绑定到调度配置
        Schedule_Def['func'] = self._handle_scheduled_Event
        # 将配置好的任务添加到调度器
        job = self.Scheduler.add_job(**Schedule_Def)
        # 可选:返回任务实例,方便后续做修改、删除等操作
        return job
    except Exception as e:
        # 捕获并处理任务添加失败的异常
        print(f"添加调度任务失败: {str(e)}")
        raise

二、任务触发时的处理逻辑

这是定时任务触发后实际执行业务逻辑的方法,调度器会在指定时间调用它:

def _handle_scheduled_Event(self, *args, **kwargs):
    try:
        # 这里写任务触发后的具体业务逻辑
        print("定时任务触发,开始处理...")
        # 示例:获取传入的任务参数并处理
        task_info = kwargs.get('task_info', '默认任务')
        print(f"正在处理任务: {task_info}")
        # 这里可以添加具体的业务操作,比如数据处理、接口调用等
        print("任务处理完成")
    except Exception as e:
        # 捕获并处理任务执行过程中的异常
        print(f"任务处理失败: {str(e)}")
        raise

一些实用提示

  • 记得在合适的时机启动调度器,比如在__init__末尾调用self.Scheduler.start(),或者提供一个单独的start_scheduler方法
  • 程序退出前要调用self.Scheduler.shutdown(),保证任务优雅停止,避免残留进程
  • 传入的Schedule_Def参数要符合APScheduler的规则,比如指定trigger(可选date/interval/cron)、对应的时间参数(比如cron_hour=8表示每天8点执行)

内容的提问来源于stack exchange,提问作者Blindfreddy

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.05.25 07:25:02