如何从asyncio.gather创建的子任务中获取对应父任务

asyncio本身没有内置父/子任务的关联映射，你可以通过以下两种稳定方式获取执行asyncio.gather的父任务：

方案1：使用contextvars传递父任务（优先推荐） #

这是Python官方支持的标准实现，不依赖asyncio内部逻辑，兼容性最好：

1. 定义全局上下文变量存储父任务实例
2. 包装asyncio.gather方法，在调用时把当前任务（即父任务）存入上下文
3. 子协程内直接从上下文变量读取父任务即可

示例代码：

import asyncio
import contextvars

# 定义上下文变量，用于存储执行gather的父任务
parent_task_ctx = contextvars.ContextVar("parent_task")

def gather_with_parent(*coros, return_exceptions=False):
    # 获取当前调用gather的任务，就是目标父任务
    current_parent = asyncio.current_task()
    parent_task_ctx.set(current_parent)
    return asyncio.gather(*coros, return_exceptions=return_exceptions)

async def child_coro():
    # 直接从上下文读取父任务
    parent_task = parent_task_ctx.get()
    print(f"当前协程对应的父任务为：{parent_task.get_name()}")
    await asyncio.sleep(0.5)

async def main():
    await gather_with_parent(
        child_coro(),
        child_coro()
    )

asyncio.run(main())

该方案天然支持嵌套gather场景：上下文变量会自动在子协程中继承，不需要额外修改就能获取最顶层或者任意层级的父任务。

方案2：猴子补丁全局asyncio.gather #

如果不想修改现有代码的gather调用逻辑，可以给全局gather打补丁，给每个子任务绑定父任务属性：

import asyncio
from functools import wraps

# 保留原生gather实现
original_gather = asyncio.gather

@wraps(original_gather)
def patched_gather(*coros, return_exceptions=False):
    parent_task = asyncio.current_task()
    # 包装每个传入的协程，绑定父任务属性
    async def wrap_coro(coro):
        return await coro
    wrapped_coros = []
    for c in coros:
        wrapped = wrap_coro(c)
        setattr(wrapped, "_parent_task", parent_task)
        wrapped_coros.append(wrapped)
    return original_gather(*wrapped_coros, return_exceptions=return_exceptions)

# 替换全局gather实现
asyncio.gather = patched_gather

async def child_coro():
    # 从当前任务的协程属性读取父任务
    current_task = asyncio.current_task()
    parent_task = getattr(current_task.get_coro(), "_parent_task", None)
    print(f"父任务为：{parent_task}")
    await asyncio.sleep(0.5)

async def main():
    await asyncio.gather(
        child_coro(),
        child_coro()
    )

asyncio.run(main())

注意：该方案依赖asyncio内部Task对象的实现逻辑，Python版本升级可能出现兼容性问题，仅适合临时调试场景使用。

注意事项 #

• 不要直接修改asyncio内置Task类的私有属性，避免版本兼容问题
• 上下文变量方案支持跨线程、跨协程的上下文传递，生产环境优先使用

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