Python asyncio.wait 怎样只处理先完成的任务

想让多个异步任务“谁先完成就先处理谁”，可以使用 asyncio.wait()，并把 return_when 设为 asyncio.FIRST_COMPLETED。它返回 (done, pending) 两个集合：done 中是已经完成或被取消的任务，pending 中仍在运行。关键点是，wait() 只负责等待和分组，不会因为超时自动取消剩余任务。

官方地址：https://docs.python.org/3/library/asyncio-task.html#asyncio.wait

FIRST_COMPLETED 返回的两个集合怎么理解

asyncio.wait() 的返回值不是一个“最先完成的结果”，而是两个集合。使用 FIRST_COMPLETED 时，只要有一个任务完成、返回异常或被取消，等待就结束；同一轮中已经完成的其他任务也会一起进入 done。因此不要把 done 当成只有一个元素的列表。

它与 asyncio.gather() 的目标不同：gather() 更适合等待一组任务并按输入顺序收集结果，wait() 更适合把“先完成的一批”作为当前处理边界。

先完成的任务怎样安全读取结果

传入 wait() 前先创建 Task。这样既满足 API 的输入要求，也能保留每个任务对象，后续用 task.result() 读取成功值或捕获任务异常。遍历集合时不要假设完成顺序，集合本身没有业务顺序。

import asyncio

async def fetch(name: str, delay: float) -> str:
    # 用不同延迟模拟多个并发 I/O 任务。
    await asyncio.sleep(delay)
    if name == "节点-2":
        # 让调用方看到 done 中的异常任务如何处理。
        raise RuntimeError("节点-2 暂时不可用")
    return f"{name} 返回结果"

async def main() -> None:
    # 先创建 Task，wait() 从 Python 3.11 起不接受 coroutine 对象。
    tasks = {
        asyncio.create_task(fetch("节点-1", 0.20)),
        asyncio.create_task(fetch("节点-2", 0.05)),
        asyncio.create_task(fetch("节点-3", 0.40)),
    }

    done, pending = await asyncio.wait(
        tasks,
        return_when=asyncio.FIRST_COMPLETED,
    )

    for task in done:
        # result() 会重新抛出该 Task 内部保存的异常。
        try:
            print("完成：", task.result())
        except Exception as exc:
            print("失败：", type(exc).__name__, str(exc))

    # 当前示例明确取消剩余任务，并等待它们收尾，避免留下悬挂 Task。
    for task in pending:
        task.cancel()
    await asyncio.gather(*pending, return_exceptions=True)

asyncio.run(main())

这段代码中，最先完成的可能是成功任务，也可能是抛异常的任务；两者都会触发 FIRST_COMPLETED。真正的业务判断应该看 task.result() 是否抛异常，而不是只看“它已经进入 done”。

加上 timeout 后 pending 会发生什么

如果只想等待一小段时间，可以给 wait() 传入 timeout。超时返回时，未完成任务仍在运行，函数只是把它们放进 pending；不会自动取消它们，也不会像 wait_for() 一样把超时转换成异常。

import asyncio

async def wait_first_with_budget(tasks: set[asyncio.Task[str]]) -> None:
    # timeout 是本轮等待预算，不是对每个 Task 的强制取消命令。
    done, pending = await asyncio.wait(
        tasks,
        timeout=0.15,
        return_when=asyncio.FIRST_COMPLETED,
    )

    for task in done:
        # 先消费已经完成的任务，异常也必须被取出。
        try:
            print("首批完成：", task.result())
        except Exception as exc:
            print("首批失败：", repr(exc))

    if pending:
        # 业务选择是停止竞速：显式取消并等待取消完成。
        for task in pending:
            task.cancel()
        await asyncio.gather(*pending, return_exceptions=True)
    else:
        print("没有剩余任务")

如果 done 为空，说明在预算内没有任务完成；如果 done 非空但 pending 也非空，说明已经得到一批结果，但还有任务没有结束。两种情况都不能直接把 pending 丢掉，否则任务可能继续占用连接、文件句柄或其他资源。

pending 应该继续等待还是取消

处理 pending 没有一个对所有场景都正确的答案，取决于业务目标：

• 竞速请求：拿到第一个可用结果后取消其余任务，并用 gather(..., return_exceptions=True) 等待清理。
• 分批消费：处理当前 done 后，把 pending 继续传给下一轮 wait()。
• 软超时：超时只记录状态，保留 pending 让它们继续完成，但必须由集合、任务组或其他所有者持有引用。

如果选择继续等待，可以把 pending 作为下一轮输入：

async def drain_in_batches(tasks: set[asyncio.Task[str]]) -> list[str]:
    results: list[str] = []
    remaining = tasks

    while remaining:
        # 每一轮只取先完成的一批，直到所有 Task 都被消费。
        done, remaining = await asyncio.wait(
            remaining,
            return_when=asyncio.FIRST_COMPLETED,
        )
        for task in done:
            # 失败任务被记录后继续处理其他已完成任务。
            try:
                results.append(task.result())
            except Exception as exc:
                print("忽略一个失败任务：", repr(exc))

    return results

这里每轮都会重新绑定 remaining，避免把已经完成的任务再次传入。若业务要求保留原始输入顺序，需要额外把任务和索引绑定；FIRST_COMPLETED 本身只表达完成时机，不提供输入顺序。

接入前需要检查的兼容边界

首先确认调用对象是 Task 或 Future，而不是裸 coroutine；Python 3.11 起直接传 coroutine 给 wait() 已被禁止。其次区分 wait(timeout=...) 和 wait_for(...)：前者返回集合并保留未完成任务，后者在超时路径取消目标并抛出 TimeoutError。

还要注意三点：输入集合不能是空集合；任务失败后要取出异常，避免“Task exception was never retrieved”；调用方要明确 pending 的所有者，不能因为函数返回就让它们无人管理。若只是想按完成顺序逐个获得结果，也可以比较 asyncio.as_completed()，它的返回模型不同，不应和 wait() 的两个集合混用。

常见问题

FIRST_COMPLETED 会只返回一个任务吗？

不一定。它会在任意任务完成或取消后返回，同一时刻已经完成的其他任务也可能一起进入 done，所以应遍历整个集合。

wait 超时后要不要取消 pending？

由业务决定。竞速请求通常取消并等待回收；后台任务可以继续运行，但必须保留任务引用并由明确的生命周期管理者负责。

可以直接把协程列表传给 wait 吗？

不建议，也不符合 Python 3.11 之后的接口要求。先用 asyncio.create_task() 创建 Task，再把 Task 集合传给 wait()。