asyncio.gather() 不支持部分超时，遇异常即中断全部任务；应改用 asyncio.wait_for() 逐个封装任务并捕获 TimeoutError，再通过 asyncio.gather() 收集结果。
asyncio.gather() 里怎么让个别任务超时而不影响其他任务
直接说结论：asyncio.gather() 本身不支持“部分超时、其余继续”，它默认是「全等」行为——只要一个任务抛出 TimeoutError（或任何异常），整个 gather() 就立刻中断，其余未完成的协程会被取消。这不是 bug，是设计使然：它本质是“原子性并发等待”。想实现“某几个超时就放弃，其他照常跑完”，得绕开 gather() 的默认异常传播逻辑。
用 asyncio.wait() + timeout 参数控制单个任务生命周期
asyncio.wait() 更底层、更可控，适合这种“差异化超时”场景。你可以把每个任务包一层 asyncio.wait_for(task, timeout=...)，再把它们丢进 wait() 或直接 await ——关键在于：对每个任务单独设超时，且把超时异常吃掉，不让它冒泡到外层。
常见做法：
- 用
asyncio.create_task()启动所有任务，得到 task 对象列表 - 对每个 task，用
asyncio.wait_for(task, timeout=...)包裹，并用try/except捕获asyncio.TimeoutError - 在 except 块里返回默认值（如
None）或标记失败，确保不 raise - 最后用
await asyncio.gather(*wrapped_tasks)收集结果（这时已无超时风险）
示例片段：
立即学习“Python免费学习笔记（深入）”；
import asyncioasync def fetch_user(user_id): await asyncio.sleep(2) return {"id": user_id, "name": "Alice"}
async def fetch_posts(user_id): await asyncio.sleep(5) # 故意慢 return [{"post_id": 1}]
async def safe_with_timeout(coro, timeout, default=None): try: return await asyncio.wait_for(coro, timeout) except asyncio.TimeoutError: return default
async def main(): tasks = [ safe_with_timeout(fetch_user(123), timeout=3), safe_with_timeout(fetch_posts(123), timeout=3), # 这个会超时 ] results = await asyncio.gather(*tasks) print(results) # [ {...}, None ]
asyncio.run(main())
为什么不用 asyncio.shield() 或 asyncio.current_task().cancel()
有人试过用 shield() 保护某个任务不被取消，但这解决不了根本问题：gather() 在遇到第一个异常时，会主动调用其余 task 的 cancel()。而 shield() 只防外部取消，防不住 gather() 自己发起的取消信号。同理，手动调用 current_task().cancel() 是反模式——你无法精准控制哪个 task 该取消、哪个该保留，还容易引发竞态。
真正可控的方式只有一种：让每个任务的超时处理完全独立，互不干扰。这意味着必须在协程入口处做超时拦截，而不是依赖 gather() 的聚合逻辑。
注意 asyncio.wait() 的 done/pending 分离陷阱
如果改用 asyncio.wait(tasks, return_when=asyncio.FIRST_COMPLETED) 等模式，要注意：pending 里的 task 并不会自动继续运行，它们仍处于挂起状态，需要你显式 await 或再次传给 wait()。否则它们就“卡住”了，既没结果也没报错。这不是 bug，是 wait() 的语义：它只告诉你哪些完成了，不负责调度剩余任务。
所以更稳妥的做法仍是：对每个任务预设超时 + 错误兜底，然后统一 gather()。复杂度低、可读性强、不易漏掉 pending 协程。
超时逻辑写在哪一层，决定了整个并发流的健壮性。别指望 gather() 做它不承诺的事。