import asyncioasync def task(name, delay): print(f"{name} 开始") await asyncio.sleep(delay) print(f"{name} 结束") return nameasync def main(): tasks = [ task("A", 2), task("B", 1), task("C", 3) ] done, pending = await asyncio.wait( tasks, return_when=asyncio.FIRST_COMPLETED ) print("已有任务完成") for t in done: print("结果:", t.result()) print("剩余任务数量:", len(pending))asyncio.run(main())
问题
asyncio.wait() 和 asyncio.gather() 的区别是什么?FIRST_COMPLETED、FIRST_EXCEPTION、ALL_COMPLETED 分别适用于什么场景?
标准答案
第一问:代码不能正常运行
运行后会报错:
TypeError: Passing coroutines is forbidden, use tasks explicitly.
原因:
tasks = [ task("A", 2), task("B", 1), task("C", 3)]
这里创建的是 协程对象(coroutine),而不是 Task对象。
Python 3.11+ 中:
要求传入的是 Task 或 Future,而不能直接传协程。
第二问:正确写法
tasks = [ asyncio.create_task(task("A", 2)), asyncio.create_task(task("B", 1)), asyncio.create_task(task("C", 3))]
完整代码:
done, pending = await asyncio.wait( tasks, return_when=asyncio.FIRST_COMPLETED)
第三问:输出结果
大概率输出:
A 开始B 开始C 开始B 结束已有任务完成结果: B剩余任务数量: 2
因为:
B 最先完成。
第四问:wait 和 gather 区别
例如:
await asyncio.gather( task1(), task2(), task3())
会等待全部完成:
才继续执行。
面试加分题
如何实现:
谁先返回结果,就采用谁的结果,其他任务全部取消?
参考答案:
done, pending = await asyncio.wait( tasks, return_when=asyncio.FIRST_COMPLETED)result = next(iter(done)).result()for p in pending: p.cancel()print("最终结果:", result)
这个模式在实际开发中非常常见:
业内通常称为:
Race Pattern(竞速模式)
面试官最喜欢追问的一句话
await asyncio.sleep(1) 和 time.sleep(1) 的区别是什么?
标准回答:
time.sleep(1)await asyncio.sleep(1)
一句话总结:
sleep 是让线程睡觉,await asyncio.sleep 是让协程让路。 🚀