如何从已取消的asyncio.gather任务中获取异步生成器结果？

核心思路是让处理生成器的任务在被取消时返回已收集的部分数据，而非抛出未处理异常，同时避免依赖asyncio.gather的默认行为（它无法返回部分完成结果）。以下是修改后的完整代码：

import asyncio


async def some_task():
    try:
        yield 1
        yield 2
        yield 3
        await asyncio.sleep(10000)
    except asyncio.CancelledError:
        # 生成器被取消时的可选清理逻辑
        pass


async def run_tasks():
    async def _collect(gen):
        collected = []
        try:
            async for value in gen:
                collected.append(value)
            return collected
        except asyncio.CancelledError:
            # 捕获取消异常，返回当前已收集的所有数据
            return collected

    collection_task = asyncio.create_task(_collect(some_task()))
    try:
        return await collection_task
    except asyncio.CancelledError:
        # 主动触发任务取消，等待它返回部分结果
        collection_task.cancel()
        return await collection_task


async def amain():
    result = []
    try:
        result = await run_tasks()
        print("任务正常完成，结果:", result)
    except KeyboardInterrupt:
        result = await run_tasks()
        print("收到中断，保存部分结果:", result)
    finally:
        # 无论正常结束还是中断，都写入磁盘
        with open("collected_data.txt", "w", encoding="utf-8") as f:
            f.write(str(result))


if __name__ == '__main__':
    asyncio.run(amain())

关键细节说明 #

• _collect函数是核心：它维护已收集数据的列表，即使任务被取消，也会捕获asyncio.CancelledError并返回当前已有的数据，确保不会丢失已获取的内容。
• 放弃asyncio.gather的原因：默认情况下，gather在任一任务被取消时会直接抛出异常，无法获取部分完成的结果。改用单独任务管理后，可主动控制取消流程并获取部分结果。
• 磁盘写入逻辑放在finally块：确保无论程序是正常完成还是被用户中断，已收集的数据都会被写入磁盘。

为什么之前的asyncio.shield无效？ #

asyncio.shield仅能保护任务不被外部取消信号终止，但如果gather本身因中断被取消，它仍会抛出异常，且无法返回部分结果。只有让任务自身处理取消并返回数据，才能拿到已收集的内容。

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