asyncio - 从任务中重新引发异常

Question

我正在使用 asyncio 进行一些 TCP 通信。我有一个Receive() 函数，它在无限循环中执行read()。这使用asyncio.create_task(Receive()) 作为后台任务运行。

现在，如果连接被对等方关闭，则会引发我在Receive() 函数中捕获的异常（或可能是任何其他异常）。但是，我想重新引发该异常，以便外部代码可以决定要做什么（例如重新连接）。

由于在任务中引发了异常，我不知道如何检索它。

我试图创建一个例子来说明我的意思：

import asyncio

async def divide(x):
    try:
        return 1/x
    except Exception as e:
        print("Divide inner exception: ", e)
        raise   # Re-raise so main() can handle it

async def someFn():
    asyncio.create_task(divide(0))  # Exception is never retrieved
    # await divide(0) # This will raise two exceptions - the original in divide() and in main()

async def main():
    try:
        await someFn()
        # Do other things while someFn() runs
    except Exception as e:
        print("main exception: ", e)

asyncio.run(main())

如何获取main()中的任务异常？

Answer 1

如何获取main()中的任务异常？

您可以利用引发异常的任务完成这一事实，而不是让任务在后台运行，而是实际等待其完成。这要求创建任务的代码不保留create_task() 返回的任务对象，并将其返回给调用者或将其存储到一组共享的任务中。 （trio 之类的库甚至不允许盲目地生成后台任务，而是要求每个任务都附有一个 nursery，该nursery 定义了谁处理其异常。

例如：

async def someFn():
    # set up a background task, but also return it to the caller
    t = asyncio.create_task(divide(0))
    return t

async def other_things():
    await asyncio.sleep(1)

async def main():
    try:
        task = await someFn()
        # await both the background task and other things, immediately
        # propagating an exception in either
        await asyncio.gather(task, other_things())
    except Exception as e:
        print("main exception: ", e)

Answer 2

另一种选择是使用add_done_callback，这样可以灵活地不必事先知道在代码中处理任务的位置（使用asyncio.gather() 选项时需要这样做）。

import asyncio

async def divide(x):
    try:
        return 1/x
    except Exception as e:
        print("Divide inner exception: ", e)
        raise   # Re-raise so callback can handle it 
                # (or don't handle exception here and allow callback to manage)

async def someFn():
    task = asyncio.create_task(divide(0))
    task.add_done_callback(task_cb)

def task_cb(task):
    try:
        task.result()
    except asyncio.CancelledError:
        pass  # Task cancellation should not be logged as an error.
    except Exception as e:
        print('Exception raised by task: ', e)

async def main():
    try:
        await someFn()
        # Do other things while someFn() runs
    except Exception as e:
        # Handle the exception in some way
        print("main exception: ", e)

asyncio.run(main())