APScheduler监听器未触发错过任务事件,求正确实现方案
解决APScheduler EVENT_JOB_MISSED监听器不触发的问题
你的代码无法触发EVENT_JOB_MISSED的核心原因是:任务执行时间(2秒)超过了间隔时间(1秒),但默认情况下APScheduler会因为max_instances=1(同一任务最多同时运行1个实例)而跳过后续任务,此时触发的是EVENT_JOB_MAX_INSTANCES事件,而非EVENT_JOB_MISSED。
EVENT_JOB_MISSED是当任务的原定执行时间已过,且超过了misfire_grace_time(错过执行的宽容时间),调度器无法再执行该任务时才会触发。要触发这个事件,需要让调度器确实错过任务的执行窗口,而非仅仅因为实例数限制跳过。
下面是修正后的代码:
from apscheduler.schedulers.background import BackgroundScheduler from apscheduler.events import EVENT_JOB_MISSED, EVENT_JOB_MAX_INSTANCES import time def listener(ev): if ev.code == EVENT_JOB_MISSED: print(f"MISSED JOB: 任务ID {ev.job_id},原定执行时间 {ev.scheduled_run_time}") elif ev.code == EVENT_JOB_MAX_INSTANCES: print(f"任务实例数超限: 任务ID {ev.job_id},当前实例数已达上限") def dataread(): print(f"任务执行中: {time.strftime('%H:%M:%S')}") time.sleep(2) # 配置调度器,设置错过执行的宽容时间为0.5秒 scheduler = BackgroundScheduler(misfire_grace_time=0.5) # 同时监听两种事件,便于调试区分情况 scheduler.add_listener(listener, EVENT_JOB_MISSED | EVENT_JOB_MAX_INSTANCES) # 添加任务,关闭自动跳过错过的任务,确保调度器处理错过的任务 scheduler.add_job( dataread, trigger="interval", seconds=1, skip_misfired_events=False ) scheduler.start() # 暂停调度器2秒,制造任务错过执行的场景 print("暂停调度器2秒,模拟任务错过...") scheduler.pause() time.sleep(2) scheduler.resume() try: # 保持主线程存活 while True: time.sleep(1) except (KeyboardInterrupt, SystemExit): scheduler.shutdown()
关键修改点说明:
- 区分事件类型:同时监听
EVENT_JOB_MISSED和EVENT_JOB_MAX_INSTANCES,明确不同场景下的事件触发情况。 - 配置
misfire_grace_time:设置宽容时间为0.5秒,当任务错过执行时间超过这个值时,会触发EVENT_JOB_MISSED。 - 关闭
skip_misfired_events:设为False,确保调度器不会自动跳过错过的任务,而是检查是否超过宽容时间后触发对应事件。 - 制造错过场景:通过暂停调度器2秒,让期间的任务无法被调度,恢复后调度器会检测到这些任务已错过并触发
EVENT_JOB_MISSED。 - 简化主线程存活逻辑:用
while True循环替代asyncio事件循环,更适合BackgroundScheduler的使用场景。
内容的提问来源于stack exchange,提问作者mikeardy
相关产品推荐
相关产品推荐

