Python asyncio.as_completed 怎样按完成顺序处理结果

批量请求多个接口时，最先完成的结果通常最有价值。如果仍然按输入列表逐个等待，前面的慢任务会挡住后面已经完成的任务。asyncio.as_completed() 的用途就是把 awaitable 按完成先后交给调用方。

官方文档：https://docs.python.org/3/library/asyncio-task.html#asyncio.as_completed

关键点是：as_completed() 改变的是结果的消费顺序，不会改变任务本身的并发执行方式；超时发生时，它也不会替你取消还没完成的任务。

为什么它能拿到先完成的结果

输入可以是一组协程、Task 或 Future。函数立即返回一个可迭代对象，迭代时每次取到“下一个已完成任务”的结果。输入顺序只决定任务集合，不决定输出顺序。

例如三个任务的预计耗时分别是 0.20、0.05、0.10 秒，消费顺序应当是第二、第三、第一。下面的数字只是示例输入，不能把它理解成固定性能承诺。

普通迭代写法兼容 Python 3.12 及更早版本

普通 for 迭代得到的是一个新的协程，需要 await 它才能取得下一个结果。这个协程不是原始 Task，所以如果需要知道结果属于哪个请求，应当在任务结果中携带标识。

import asyncio

async def fetch(name: str, delay: float) -> str:
    # 用延迟模拟不同请求的完成时间，实际项目中替换为 I/O 调用。
    await asyncio.sleep(delay)
    if name == "库存":
        raise RuntimeError("库存服务暂时不可用")
    return f"{name}完成"

async def main() -> None:
    tasks = [
        asyncio.create_task(fetch("详情", 0.20)),
        asyncio.create_task(fetch("库存", 0.05)),
        asyncio.create_task(fetch("推荐", 0.10)),
    ]
    try:
        # 每次 await 的都是最先完成的那个结果，不是列表中的下一个位置。
        for next_result in asyncio.as_completed(tasks):
            try:
                result = await next_result
            except Exception as exc:
                # 单个任务失败只影响本次消费，是否中止由业务决定。
                print(f"单项失败: {exc}")
            else:
                print(result)
    finally:
        # 如果外层提前退出，主动取消并等待剩余任务，避免留下悬挂任务。
        for task in tasks:
            if not task.done():
                task.cancel()
        await asyncio.gather(*tasks, return_exceptions=True)

asyncio.run(main())

这段代码的输出顺序大致会先看到“库存”的失败，再看到“推荐完成”和“详情完成”。重点不是把异常吞掉，而是把异常处理位置放在每次 await 附近，让其它已经排队的结果仍有机会被消费。

Python 3.13 及更新版本怎样保留 Task 身份

当前官方文档说明，Python 3.13 起 as_completed() 还可以作为异步迭代器使用。使用 async for 时，如果输入本来就是 Task 或 Future，迭代得到的就是这些原始对象，便于从映射表中找回请求名。

async def collect_by_task() -> dict[str, str]:
    tasks = {
        asyncio.create_task(fetch("详情", 0.20)): "详情",
        asyncio.create_task(fetch("库存", 0.05)): "库存",
        asyncio.create_task(fetch("推荐", 0.10)): "推荐",
    }
    results: dict[str, str] = {}
    # 异步迭代时拿到原始 Task，可以直接关联业务键。
    async for task in asyncio.as_completed(tasks):
        service = tasks[task]
        try:
            results[service] = await task
        except Exception as exc:
            results[service] = f"失败: {exc}"
    return results

如果项目还要兼容 Python 3.12 或更早版本，继续使用普通迭代，并让协程返回 (业务键, 结果)；不要假设普通迭代返回的对象可以与原始 Task 做身份比较。

timeout 只报告未完成，不负责清理

给 as_completed() 传入 timeout 后，如果规定时间内没有全部完成，迭代会抛出 TimeoutError。官方文档明确说明，剩余任务不会因此自动取消，它们仍可能继续运行。

async def collect_with_timeout(tasks: list[asyncio.Task[str]]) -> list[str]:
    values: list[str] = []
    try:
        # timeout 是整个结果集合的等待上限，不是每个任务各自的上限。
        for next_result in asyncio.as_completed(tasks, timeout=0.15):
            values.append(await next_result)
    except TimeoutError:
        # 只记录超时；下面仍要决定未完成任务是取消还是继续交给后台管理。
        print("结果集合未在截止时间内全部完成")
    finally:
        for task in tasks:
            if not task.done():
                # 当前调用方不再需要迟到结果，因此取消并等待清理。
                task.cancel()
        await asyncio.gather(*tasks, return_exceptions=True)
    return values

如果业务允许迟到结果继续写入缓存，就不要在这里取消，而是把剩余 Task 交给有生命周期管理的后台组件；如果请求已经结束，通常应取消并等待它们完成清理。

几个容易混淆的边界

实际选择可以归纳为：只关心谁先完成时用普通迭代即可；需要把结果和原始 Task 关联时，优先使用 Python 3.13+ 的异步迭代；存在截止时间时，始终单独设计未完成任务的清理策略。

相关问题

as_completed 和 gather 有什么区别？ gather 更适合一次等待并按输入顺序收集；as_completed 适合边完成边消费。

as_completed 超时后还能拿到迟到结果吗？ 可以，但要保留 Task 引用并由其它生命周期管理逻辑继续等待；超时本身不会替你收集迟到结果。

为什么不能直接把普通迭代返回值和原始 Task 比较？ 因为普通迭代返回的是用于取得下一个结果的协程，不是原始 Task；需要身份关联时使用异步迭代或在结果中带业务键。