asyncio.wait_for超时不准是因为它只在协程挂起点检查超时，CPU密集型任务无法被抢占；需配合手动拆分、await让点、task.cancel()及正确处理CancelledError才能实现可靠超时。
asyncio.wait_for 为什么经常超时不准
直接用 asyncio.wait_for(task, timeout=1.0) 看似简单，但实际中常出现“明明设了 1 秒超时，却等了 1.2 秒才抛出 TimeoutError”——根本原因是它只在协程**挂起点**检查超时，而协程若正在执行 CPU 密集型逻辑（比如循环计算、正则匹配大文本），事件循环无法抢占，计时器就“卡住”了。
这不是 bug，是 asyncio 的设计前提：协程必须主动让出控制权。所以 wait_for 的超时精度依赖于协程是否频繁 await。
- 适合 I/O 类协程（
await aiohttp.get(...)、await asyncio.sleep(...)） - 不适合含长循环、
time.sleep()、或未分块的解析逻辑 - 超时后，原 task 仍可能在后台继续运行（除非显式
cancel()）
需要硬性截断时，用 asyncio.create_task + cancel
当任务可能卡死（比如第三方异步库没做良好取消支持，或内部用了阻塞调用），仅靠 wait_for 不够。得自己管理 task 生命周期，并在超时后强制终止：
import asyncio
<p>async def risky_job():
await asyncio.sleep(3) # 模拟慢操作
return "done"</p><p>async def run_with_hard_timeout(coro, timeout):
task = asyncio.create_task(coro)
try:
return await asyncio.wait_for(task, timeout)
except asyncio.TimeoutError:
task.cancel()
try:
await task # 等它响应 cancel（可选，但推荐）
except (asyncio.CancelledError, Exception):
pass
raise</p><h1>使用</h1><p>try:
result = await run_with_hard_timeout(risky_job(), 1.0)
except asyncio.TimeoutError:
print("真的被干掉了")
注意：task.cancel() 发送的是 CancelledError，能否真正中断取决于协程内部是否捕获并处理它，或是否在可取消点（如 await）及时响应。
立即学习“Python免费学习笔记（深入）”；
对 CPU 密集型操作，必须拆解 + await yield
如果协程里有纯 Python 循环、大数组处理、或调用非异步的阻塞函数（如 json.loads 大字符串），wait_for 完全失效。唯一办法是手动插入让点：
- 把大循环拆成小批次，每批后
await asyncio.sleep(0)或await asyncio.yield_()（3.12+） - 用
loop.run_in_executor把阻塞调用移到线程池，再await结果 - 避免在协程里写
time.sleep()—— 它会彻底阻塞整个 event loop
例如处理一个 10 万行列表：
async def process_lines(lines):
for i, line in enumerate(lines):
# 处理单行...
if i % 100 == 0: # 每 100 行让出一次
await asyncio.sleep(0)
timeout 参数不是魔法，它依赖协程的协作性
asyncio 的超时机制本质是“协作式抢占”，不是操作系统级的硬中断。这意味着：
-
asyncio.timeout()（3.11+ 上下文管理器）和wait_for行为一致，只是语法更清晰 - 第三方库若没正确传播
CancelledError（比如某些旧版aiofiles或自定义协议实现），cancel 可能静默失败 - 在
async with块内启动的子任务，需确保其生命周期被父协程管理，否则超时 cancel 可能漏掉它们
真正可靠的超时，永远由「合理拆分工作」+「显式 cancel」+「防御性 await」共同构成，而不是依赖某一个参数。
