AsyncIOScheduler CronTrigger在run_until_complete下失效问题求助

使用AsyncIOScheduler和CronTrigger调度异步函数时遇到异常：

• 调用asyncio.get_event_loop().run_forever()时，dummyJob能按Cron规则正常触发执行；
• 改用asyncio.get_event_loop().run_until_complete(pastTime(endTime))后，dummyJob完全不执行，调整misfire_grace_time也无效果。

怀疑是pastTime函数阻塞了协程调度，需确认该现象是否为预期情况，同时需要实现按条件停止事件循环后切换到其他定时任务的功能。

原代码如下：

import asyncio
from datetime import timedelta,datetime
from apscheduler.schedulers.asyncio import AsyncIOScheduler
from apscheduler.triggers.cron import CronTrigger
from time import sleep

async def pastTime(endTime):
    while(datetime.now()<=endTime):
        print(datetime.now());
        print(endTime);
        sleep(1);
        
async def dummyJob():# The real one needs to call into asyncIO
    print("scheduled job running at");
    print(datetime.now());
    
def entranceFcn():
    trigger = CronTrigger(day_of_week='mon-fri', hour= '9-23', minute="*");
    sched = AsyncIOScheduler();
    sched.add_job(dummyJob, trigger, misfire_grace_time=None);
    sched.start();
    TODAY_16_00_TIME = datetime.now().replace(hour=22,minute=20,second=0,microsecond=0);
    endTime = TODAY_16_00_TIME + timedelta(seconds = 60);
    try:
       #asyncio.get_event_loop().run_until_complete(pastTime(endTime));
       asyncio.get_event_loop().run_forever();#   => this works!
    except KeyboardInterrupt:
       sched.shutdown()

entranceFcn();

这是完全符合预期的现象，核心问题有两点：

1. 同步阻塞调用占用事件循环：pastTime是异步函数，但内部使用了time.sleep(1)——这是同步阻塞方法，会直接占满事件循环的执行时间，导致AsyncIOScheduler的调度任务根本没有机会被触发。
2. run_until_complete的执行逻辑限制：该方法会持续阻塞，直到传入的协程执行完毕。在阻塞期间，事件循环无法处理任何其他回调（包括调度器的定时任务），因此dummyJob完全无法运行。

步骤1：修复pastTime的阻塞问题 #

将同步的time.sleep(1)替换为异步的asyncio.sleep(1)，让事件循环在等待时可以处理其他任务。

步骤2：实现按条件停循环+任务切换 #

通过控制事件循环的启停，在满足停止条件后初始化新的调度任务并重启循环。

修改后的完整代码：

import asyncio
from datetime import timedelta, datetime
from apscheduler.schedulers.asyncio import AsyncIOScheduler
from apscheduler.triggers.cron import CronTrigger

async def pastTime(endTime, loop):
    while datetime.now() <= endTime:
        print(f"当前时间: {datetime.now()}")
        print(f"结束时间: {endTime}")
        await asyncio.sleep(1)  # 异步sleep，释放事件循环
    # 条件满足，停止当前事件循环
    loop.stop()

async def dummyJob():
    print(f"调度任务执行: {datetime.now()}")

async def newScheduledJob():
    print(f"新调度任务执行: {datetime.now()}")

def entranceFcn():
    loop = asyncio.get_event_loop()
    
    # 初始化第一个调度器
    trigger = CronTrigger(day_of_week='mon-fri', hour='9-23', minute="*")
    sched = AsyncIOScheduler(event_loop=loop)
    sched.add_job(dummyJob, trigger, misfire_grace_time=None)
    sched.start()

    # 设置结束时间
    TODAY_TARGET_TIME = datetime.now().replace(hour=22, minute=20, second=0, microsecond=0)
    endTime = TODAY_TARGET_TIME + timedelta(seconds=60)

    try:
        # 将pastTime加入事件循环，同时用run_forever维持循环运行
        loop.create_task(pastTime(endTime, loop))
        loop.run_forever()
    except KeyboardInterrupt:
        sched.shutdown()
    finally:
        # 切换到新的定时任务
        print("===== 切换到新定时任务 =====")
        new_sched = AsyncIOScheduler(event_loop=loop)
        # 示例：每分钟执行一次新任务
        new_sched.add_job(newScheduledJob, 'interval', minutes=1)
        new_sched.start()
        # 重新启动事件循环
        loop.run_forever()

entranceFcn()

关键修改说明 #

• 用await asyncio.sleep(1)替代time.sleep(1)，确保事件循环在等待期间可以处理调度器任务；
• 使用loop.create_task()将pastTime加入事件循环，配合loop.run_forever()实现多任务并发，调度器和pastTime可同时执行；
• 当pastTime满足停止条件时调用loop.stop()终止循环，在finally块中初始化新调度器并重启循环，完成任务切换。

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