如何在Python中使用asyncio.as_completed流式处理结果？-Python教程-PHP中文网

asyncio.as_completed不能直接流式获取结果，因为它返回的是task对象而非结果值，必须显式await每个task才能得到实际结果；若不await则停留在task状态，且异常未捕获会导致整个流中断。

如何在python中使用asyncio.as_completed流式处理结果？

asyncio.as_completed 为什么不能直接“流式”获取结果？

asyncio.as_completed 本身确实按完成顺序返回 Task 对象，但它返回的是协程对象（await task 才能得到结果），不是结果本身。很多人误以为它像生成器一样“yield 结果”，实际要手动 await 每个 task 才能拿到值——不 await，就卡在 Task 状态，没法真正“流式处理”。

常见错误现象：for f in asyncio.as_completed(tasks): print(f) 输出一堆 <task pending ...></task>，而不是你想要的计算结果。

必须用 await f 或 f.result()（后者需确保已 done 且无异常）
如果任务抛异常，f.result() 会直接 raise；而 await f 同样会传播异常，需 try/catch
不要在同步循环里混用 await：整个外层函数必须是 async def

正确写法：用 async for + await 获取每个完成结果

最简可行流式处理模式就是在一个 async def 函数里，遍历 asyncio.as_completed 并逐个 await：

import asyncio
<p>async def fetch(url):
await asyncio.sleep(1)  # 模拟网络延迟
return f"done: {url}"</p><p>async def main():
tasks = [fetch(u) for u in ["a", "b", "c"]]
for f in asyncio.as_completed(tasks):
result = await f  # ← 关键：这里 await 才真正取值
print(result)</p><p>asyncio.run(main())</p>

注意：这个 for 循环是同步语法，但内部 await 让它实际是异步调度的——as_completed 保证了谁先 finish 谁先被 await 到，所以输出顺序不依赖列表顺序。

361SCAN security check your skill python scripts

扫描并分析已安装技能，用于查看指定技能目录的名称、描述和详情，或列出所有已安装的技能。

下载

性能影响：没有额外线程或进程开销，纯事件循环调度
兼容性：Python 3.7+ 都支持；3.6 需用 loop.run_until_complete
如果想边收结果边做后续处理（比如写文件、发 HTTP），就在这里 print(result) 的位置插入逻辑

如何安全处理异常并继续流式消费？

只要有一个 Task 抛异常，await f 就会中断当前迭代。想“跳过失败任务、继续收其他结果”，得显式捕获：

for f in asyncio.as_completed(tasks):
    try:
        result = await f
        print("success:", result)
    except Exception as e:
        print("failed:", type(e).__name__)

常见错误场景：TimeoutError、ConnectionError、JSON 解析失败等。不加 try，整个流就停了。

f.exception() 可以提前检查是否失败（不用 await），但依然要调用一次 await f 或 f.done() 才能确认状态
别用 f.cancel() 在这里——as_completed 返回的 task 已经在跑，cancel 可能引发 CancelledError，反而干扰流式逻辑
如果失败任务需要重试，建议在 fetch() 内部封装重试逻辑，而不是在外层循环里重新创建 task