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