如何让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
相关产品推荐
相关产品推荐

