如何捕获asyncio.create_task创建的后台长运行任务异常？现有方案是否可靠？

假设从main()中通过task = asyncio.create_task(daemon_coro(signal))启动一个永久运行的后台守护进程，其中signal为asyncio.Event()对象。daemon_coro()会在出现异常或成功启动时设置signal。

目前通过结合done()与exception()方法来检测异常，代码如下：

tasks = set()
signal = asyncio.Event()

try:
    task = asyncio.create_task(daemon_coro(signal))
    tasks.add(task)

    # Wait for the signal. daemon_coro() executes signal.set()
    # in case of exceptions or in case the daemon was successfully
    # started.
    await signal.wait()

    if task.done() and task.exception() is not None:
        # There was an exception in daemon_coro(). Terminate the script.
        raise asyncio.CancelledError

    # If no exceptions were raised, then the daemon_coro()
    # keeps running in the background and other coros
    # are called here.

except asyncio.CancelledError:
    pass

# Clean-up
for task in tasks:
    task.cancel()

await asyncio.gather(*tasks, return_exceptions=True)

作为asyncio新手，不确定这种结合asyncio.Event()、done()和exception()来捕获asyncio.create_task()创建的永久运行任务异常的方案是否健壮、是否符合最佳实践？

你的方案具备基本可行性，但存在可优化点，也有不符合asyncio最佳实践的地方：

1. 当前方案的潜在问题 #

• 依赖daemon_coro()严格遵守“异常/成功都设置signal”的约定，一旦代码变更遗忘设置signal，主线程会永久阻塞在signal.wait()。
• 滥用asyncio.CancelledError作为终止信号：该异常原本用于任务取消场景，用它来传递异常终止逻辑会让代码语义模糊。
• 虽然task.done()配合task.exception()的检测逻辑在正常情况下能工作，但整体流程绕弯，不如直接针对任务状态做处理清晰。

2. 更健壮的替代方案 #

方案一：给任务绑定异常回调 #

直接给后台任务添加add_done_callback，任务抛出异常时自动触发处理逻辑，无需依赖Event传递信号：

import logging

tasks = set()
startup_success = asyncio.Event()

def handle_daemon_failure(task):
    try:
        # 调用result()会重新抛出任务内的异常
        task.result()
    except Exception:
        logging.exception("后台守护进程启动失败")
        # 触发主线程终止逻辑
        startup_success.set()  # 或者用单独的failure_event

try:
    task = asyncio.create_task(daemon_coro(startup_success))
    task.add_done_callback(handle_daemon_failure)
    tasks.add(task)

    # 等待启动成功信号
    await startup_success.wait()

    # 检查是否是异常触发的信号（可以用两个Event区分成功/失败）
    # 后续业务逻辑

except Exception as e:
    print(f"处理异常：{e}")

# 清理逻辑
for task in tasks:
    task.cancel()
await asyncio.gather(*tasks, return_exceptions=True)

方案二：同时等待任务与启动信号 #

用asyncio.wait同时监听任务完成（意味着异常）和启动成功信号，哪个先触发就处理哪个：

tasks = set()
startup_success = asyncio.Event()

try:
    task = asyncio.create_task(daemon_coro(startup_success))
    tasks.add(task)

    # 等待第一个完成的事件：任务异常退出 或 启动成功
    done, pending = await asyncio.wait(
        [task, startup_success.wait()],
        return_when=asyncio.FIRST_COMPLETED
    )

    if task in done:
        # 任务已完成，说明抛出了异常
        task.result()  # 重新抛出异常，或直接处理
    else:
        # 启动成功，执行后续逻辑
        pass

except Exception as e:
    print(f"守护进程启动失败：{e}")

# 清理逻辑
for task in tasks:
    task.cancel()
await asyncio.gather(*tasks, return_exceptions=True)

3. 总结 #

你的方案可以正常运行，但并非最佳实践。更推荐使用异常回调或同时等待任务与启动信号的方式，这类方案逻辑更清晰、健壮性更强，也更贴合asyncio的设计思路。

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