Python：如何在不退出主协程的情况下启停调度协程任务

核心是利用asyncio.create_task()创建后台任务，通过协作式取消机制（task.cancel()）终止子协程，并确保任务能彻底停止后回到主协程。

关键注意点 #

• asyncio的取消是协作式的：子协程必须包含可让出控制权的等待点（比如await asyncio.sleep()），否则cancel()无法中断它。
• 调用task.cancel()后，必须await task等待任务完成取消流程，确保子协程的清理逻辑执行完毕。
• 主协程可通过用户输入、事件信号等方式触发取消操作。

基础实现代码 #

import asyncio

# 永久运行的子协程
async def scheduled_task():
    try:
        while True:
            print("子任务正在运行...")
            # 必须保留等待点，让事件循环有机会处理取消信号
            await asyncio.sleep(1)
    except asyncio.CancelledError:
        print("子任务已被取消，执行清理操作...")
        # 这里可以添加资源释放、状态保存等清理逻辑
        raise  # 重新抛出异常，标记任务为完成状态

# 主协程
async def main_coroutine():
    print("主协程启动")
    
    # 1. 调度子任务，主协程挂起等待取消信号
    task = asyncio.create_task(scheduled_task())
    print("子任务已调度，主协程进入等待状态")
    
    # 模拟用户触发取消（实际场景可替换为API调用、事件信号等）
    await asyncio.get_event_loop().run_in_executor(None, input, "按下回车键取消子任务...")
    
    # 4. 取消子任务并等待其结束
    task.cancel()
    try:
        await task
    except asyncio.CancelledError:
        pass  # 子协程已处理过该异常，无需重复处理
    
    # 回到主协程继续执行
    print("回到主协程，恢复运行")
    while True:
        print("主协程持续运行中...")
        await asyncio.sleep(2)

if __name__ == "__main__":
    try:
        asyncio.run(main_coroutine())
    except KeyboardInterrupt:
        print("程序退出")

代码解释 #

1. 子协程scheduled_task()：

   • 用try-except捕获CancelledError，在取消时执行清理逻辑，最后重新抛出异常确保任务被标记为完成。
   • await asyncio.sleep(1)是关键的等待点，让事件循环能插入取消信号的处理。

2. 主协程main_coroutine()：

   • asyncio.create_task()将子协程转为后台任务，不阻塞主协程的后续逻辑。
   • run_in_executor调用同步的input()，避免阻塞事件循环（asyncio不允许在协程中直接调用同步阻塞函数）。
   • await task确保子任务完成取消流程后，主协程才继续执行。

扩展：用事件信号优雅取消 #

如果需要通过代码逻辑触发取消（而非用户输入），可以使用asyncio.Event()实现更灵活的控制：

import asyncio

async def scheduled_task(stop_event):
    try:
        while not stop_event.is_set():
            print("子任务正在运行...")
            await asyncio.sleep(1)
    except asyncio.CancelledError:
        print("子任务被强制取消")
        raise

async def main_coroutine():
    print("主协程启动")
    stop_event = asyncio.Event()
    
    task = asyncio.create_task(scheduled_task(stop_event))
    print("子任务已调度")
    
    # 模拟3秒后自动触发取消
    await asyncio.sleep(3)
    stop_event.set()  # 设置事件，子协程自行退出
    # 也可以直接用task.cancel()强制取消
    # task.cancel()
    
    await task
    print("回到主协程")
    
    while True:
        print("主协程持续运行中...")
        await asyncio.sleep(2)

if __name__ == "__main__":
    asyncio.run(main_coroutine())

这种方式让子协程主动检测退出信号，适合需要优雅停止的场景，避免依赖异常处理。

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