如何为无法正常取消的asyncio任务设置取消超时？

问题代码 #

import asyncio
from async_timeout import timeout


async def sleep_coroutine(sleep=10):
    while True:
        print("Coroutine running")
        try:
            await asyncio.sleep(10)
        except Exception:
            print("Ignoring exception")


async def main():
    task = asyncio.get_event_loop().create_task(sleep_coroutine())
    await asyncio.sleep(2)

    print("Cancelling")
    task.cancel()

    with timeout(3) as cancel_timeout:
        await task

    if cancel_timeout.expired:
        print("Timed out !")

    print("Finally done !")


if __name__ == "__main__":
    asyncio.get_event_loop().run_until_complete(main())

问题分析 #

这段代码的核心问题在于：sleep_coroutine使用宽泛的except Exception捕获了所有异常，包括asyncio.CancelledError并直接忽略。调用task.cancel()后，任务不会终止，await task会陷入无限阻塞。而async_timeout依赖任务取消机制，因此无法解决该问题。

解决方案 #

由于无法修改目标协程，我们无法通过asyncio的原生取消机制终止它，但可以放弃等待该任务避免主线程阻塞；如果需要彻底终止任务，只能借助进程级别的控制手段。

方案1：放弃等待任务（避免阻塞） #

修改main函数，使用asyncio.wait_for设置等待超时，超时后直接跳过等待逻辑：

import asyncio


async def sleep_coroutine(sleep=10):
    while True:
        print("Coroutine running")
        try:
            await asyncio.sleep(10)
        except Exception:
            print("Ignoring exception")


async def main():
    task = asyncio.get_event_loop().create_task(sleep_coroutine())
    await asyncio.sleep(2)

    print("Cancelling")
    task.cancel()

    # 设置3秒等待超时，超时后放弃等待
    try:
        await asyncio.wait_for(task, timeout=3)
    except asyncio.TimeoutError:
        print("Timed out !")
        # 注意：此时任务仍会在后台运行，无法通过asyncio强制终止

    print("Finally done !")


if __name__ == "__main__":
    asyncio.get_event_loop().run_until_complete(main())

方案2：进程级强制终止（彻底结束任务） #

如果必须彻底终止不响应取消的协程，可将其放到单独的子进程中，超时后杀死子进程：

import asyncio
from multiprocessing import Process


def run_coroutine():
    async def sleep_coroutine():
        while True:
            print("Coroutine running")
            try:
                await asyncio.sleep(10)
            except Exception:
                print("Ignoring exception")

    asyncio.get_event_loop().run_until_complete(sleep_coroutine())


async def main():
    proc = Process(target=run_coroutine)
    proc.start()
    await asyncio.sleep(2)

    print("Cancelling")
    proc.terminate()

    # 等待进程终止，设置3秒超时
    try:
        await asyncio.wait_for(asyncio.to_thread(proc.join), timeout=3)
    except asyncio.TimeoutError:
        print("Timed out !")
        proc.kill()

    print("Finally done !")


if __name__ == "__main__":
    asyncio.get_event_loop().run_until_complete(main())

关键说明 #

• asyncio的协程取消依赖CancelledError的传播，若协程捕获该异常并不重新抛出，asyncio无法强制终止它。
• 方案1只是避免主线程阻塞，任务仍会在后台占用资源；方案2通过进程隔离实现彻底终止，但会引入进程间通信的开销。

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