如何将自定义异常抛出到正在运行的任务中答案

【问题标题】：How to throw a custom exception into a running task如何将自定义异常抛出到正在运行的任务中
【发布时间】：2019-12-21 12:26:12
【问题描述】：

我试图弄清楚是否有可能将自定义异常抛出到正在运行的 asyncio 任务中，类似于 Task.cancel(self) 所实现的，它计划在底层协程中引发 CancelledError。

我遇到了Task.get_coro().throw(exc)，但调用它似乎就像打开一大罐蠕虫一样，因为我们可能会让任务处于糟糕的状态。特别是考虑到task is throwing CancelledError into its coroutine 时发生的所有机制。

考虑以下示例：

import asyncio

class Reset(Exception):
    pass

async def infinite():
    while True:
        try:
            print('work')
            await asyncio.sleep(1)
            print('more work')
        except Reset:
            print('reset')
            continue
        except asyncio.CancelledError:
            print('cancel')
            break

async def main():
    infinite_task = asyncio.create_task(infinite())
    await asyncio.sleep(0)  # Allow infinite_task to enter its work loop.
    infinite_task.get_coro().throw(Reset())
    await infinite_task

asyncio.run(main())

## OUTPUT ##
# "work"
# "reset"
# "work"
# hangs forever ... bad :(

我尝试做的事情是否可行？感觉好像我不应该像这样操纵底层协程。有什么解决方法吗？

【问题讨论】：

标签： python python-3.x python-asyncio

【解决方案1】：

无法将自定义异常抛出到正在运行的任务中。您不应该与.throw 混为一谈 - 这是一个实现细节，更改它可能会破坏某些东西。

如果您想将信息（关于重置）传递给任务，请通过参数进行。以下是它的实现方式：

import asyncio
from contextlib import suppress


async def infinite(need_reset):
    try:
        while True:
            inner_task = asyncio.create_task(inner_job())

            await asyncio.wait(
                [
                    need_reset.wait(),
                    inner_task
                ], 
                return_when=asyncio.FIRST_COMPLETED
            )

            if need_reset.is_set():
                print('reset')
                await cancel(inner_task)
                need_reset.clear()
    except asyncio.CancelledError:
        print('cancel')
        raise  # you should never suppress, see:
               # https://stackoverflow.com/a/33578893/1113207


async def inner_job():
    print('work')
    await asyncio.sleep(1)
    print('more work')


async def cancel(task):
    # more info: https://stackoverflow.com/a/43810272/1113207
    task.cancel()
    with suppress(asyncio.CancelledError):
        await task


async def main():
    need_reset = asyncio.Event()
    infinite_task = asyncio.create_task(infinite(need_reset))

    await asyncio.sleep(1.5)
    need_reset.set()

    await asyncio.sleep(1.5)
    await cancel(infinite_task)


asyncio.run(main())

输出：

work
more work
work
reset
work
more work
work
cancel

【讨论】：