asyncio.gather()默认不支持部分任务超时而其余继续，需用with_timeout等包装函数捕获TimeoutError并返回默认值，避免异常冒泡中断其他任务。
在 asyncio.gather() 中，**默认不支持“部分任务超时、其余继续”**——一旦某个任务被 asyncio.wait_for() 包裹后超时抛出 TimeoutError，而该异常未被单独捕获，整个 gather() 就会立即中止，其他未完成任务会被取消。
要实现“某几个任务可超时，其余照常运行”，关键在于：**不让超时异常向上冒泡到 gather() 层级**，而是提前拦截并转为普通返回值（如 None、False 或自定义占位符）。
用 async wrapper 封装单个任务并处理超时
为每个可能超时的任务写一个带 try/except TimeoutError 的异步包装函数：
import asyncioasync def with_timeout(coro, timeout, default=None): try: return await asyncio.wait_for(coro, timeout) except asyncio.TimeoutError: return default
示例任务
async def fetch_user(user_id): await asyncio.sleep(2 if user_id != 3 else 5) # userid=3 故意慢 return f"user{user_id}"
并发执行，但只对 user_id=3 设 3 秒超时，其余不限
tasks = [ fetch_user(1), fetch_user(2), with_timeout(fetch_user(3), timeout=3, default="timeout"), fetch_user(4), ]
results = await asyncio.gather(*tasks)
结果类似：['user_1', 'user_2', 'timeout', 'user_4']
所有任务都跑完（包括 1/2/4），user_3 超时后返回 'timeout'，不中断别人
避免意外取消：确保超时任务本身可安全中断
如果被包装的协程内部有重要清理逻辑（如释放锁、关闭连接），需确保它能响应取消（即不屏蔽 CancelledError）。否则 wait_for 超时时强行取消，可能导致资源泄漏。
- 不要在协程里用
try/except CancelledError: pass吞掉取消信号 - 推荐用
asyncio.shield()保护关键清理段（但慎用，可能让超时失效） - 更稳妥的做法：在协程内部主动检查
asyncio.current_task().cancelled()并做清理
需要区分“超时”和“失败”？用 sentinel 类型或元组返回
若业务上需明确知道是超时而非报错，可返回结构化结果：
from dataclasses import dataclass@dataclass class TimeoutResult: is_timeout: bool = True
@dataclass class SuccessResult: value: any is_timeout: bool = False
async def fetch_with_status(coro, timeout): try: res = await asyncio.wait_for(coro, timeout) return SuccessResult(res) except asyncio.TimeoutError: return TimeoutResult()
使用
tasks = [ fetch_with_status(fetch_user(1), 10), fetch_with_status(fetch_user(3), 3), ] results = await asyncio.gather(*tasks)
每个 result 是 SuccessResult 或 TimeoutResult，可清晰分支处理
替代方案：用 asyncio.wait() 手动控制
若逻辑复杂（比如动态增减任务、按完成顺序处理），可绕过 gather，直接用 asyncio.wait() 监听完成集，并对已完成任务立刻处理，对超时任务单独标记：
pending = {
asyncio.create_task(fetch_user(1)): "user1",
asyncio.create_task(fetch_user(2)): "user2",
asyncio.create_task(fetch_user(3)): "user3",
}
done, pending = await asyncio.wait(
pending.keys(),
timeout=3,
return_when=asyncio.FIRST_COMPLETED
)
处理已 done 的结果
for t in done:
try:
result = await t
print(f"✅ {pending[t]}: {result}")
except Exception as e:
print(f"❌ {pending[t]}: {e}")
剩余 pending 任务继续等或取消
for t in pending:
t.cancel()
这种方式更灵活，但代码量增加，适合精细化调度场景。
核心就一点：超时不能让异常逃出单个任务上下文。封装 + 捕获，是最简单可靠的解法。
