asyncio.wait_for 是首选方案，因其原生支持超时取消、自动抛出 TimeoutError 且无需手动管理任务状态；错误用法包括混用同步 sleep 或误信 create_task 自带超时。
asyncio.wait_for 为什么是首选方案
asyncio.wait_for 是 asyncio 原生支持超时取消的最直接方式，它不依赖手动管理任务状态或信号，也不需要自己写 timeout 逻辑。它在超时发生时会自动 cancel() 被包装的协程，并抛出 asyncio.TimeoutError。
常见错误是试图用 time.sleep() 或同步计时器套在协程外，这会阻塞整个事件循环；或者误以为 asyncio.create_task() 自带超时能力 —— 它没有。
使用场景包括：等待网络响应、调用第三方异步 API、等待某个条件变量就绪。
import asyncio
<p>async def fetch_data():
await asyncio.sleep(3)
return "done"</p><p>try:
result = await asyncio.wait_for(fetch_data(), timeout=2.0)
except asyncio.TimeoutError:
print("fetch_data timed out")</p>注意：timeout 参数单位是秒（float），不是毫秒；超时后原协程会被取消，但已执行的 await 点（如底层 socket read）是否真正中断，取决于被 await 对象是否响应取消 —— 比如 aiohttp.ClientSession.get() 支持，而某些自定义协程若没检查 asyncio.current_task().cancelled() 就可能“假取消”。
立即学习“Python免费学习笔记（深入）”；
什么时候该用 asyncio.shield + wait_for 组合
当你想防止超时取消波及到某些必须完成的清理逻辑时，asyncio.shield() 就派上用场了。比如你启动了一个协程做资源释放（如关闭连接、写日志），又希望主逻辑有超时，但不希望清理动作被连带取消。
常见错误是把整个业务逻辑包进 shield() —— 这会让超时失效；正确做法是只 shield 清理部分，主流程仍由 wait_for 控制。
async def main_work():
# 主工作，可能耗时
await asyncio.sleep(5)
<p>async def cleanup():</p><div class="aritcle_card flexRow">
<div class="artcardd flexRow">
<a class="aritcle_card_img" href="/xiazai/skill7351" title="testing-python"><img
src="https://img.php.cn/upload/skill/000/000/081/179143938488980.jpg" alt="testing-python" onerror="this.onerror='';this.src='/static/lhimages/moren/morentu.png'" ></a>
<div class="aritcle_card_info flexColumn">
<a href="/xiazai/skill7351" title="testing-python">testing-python</a>
<p>使用pytest编写和评估有效的Python测试。适用于编写测试、审查测试代码、调试测试失败或提高测试覆盖率。</p>
</div>
<a href="/xiazai/skill7351" title="testing-python" class="aritcle_card_btn flexRow flexcenter"><b></b><span>下载</span> </a>
</div>
</div><h1>必须运行完的清理</h1><pre class='brush:python;toolbar:false;'>await asyncio.sleep(1)
print("cleanup done")async def run_with_timeout(): task = asyncio.create_task(main_work()) try: await asyncio.wait_for(task, timeout=2.0) except asyncio.TimeoutError: print("main_work cancelled")
cleanup 不受 task 取消影响
await asyncio.shield(cleanup())
性能影响很小，shield() 本质只是让被包裹的协程忽略外部取消请求，直到它自己结束或抛出异常。
loop.call_later + cancel() 手动实现的风险点
有人会想到用 loop.call_later(timeout, task.cancel) 配合 asyncio.create_task() 来模拟超时。这种方式可行，但容易踩坑：
-
loop.call_later的回调在事件循环线程中执行，如果任务已在运行中并处于不可取消点（如等待一个不响应取消的await），cancel()只会标记状态，不会立即退出； - 没有统一异常类型，需手动判断
task.cancelled()并处理； - 如果任务提前完成，忘记取消定时器，会导致
task.cancel()在已完成任务上调用 —— 这不会报错，但属于冗余操作，且可能掩盖真实问题； - 多个并发任务时，每个都要维护自己的 timer handle，容易漏清理。
# 不推荐：手动管理取消定时器
loop = asyncio.get_running_loop()
task = asyncio.create_task(some_coro())
handle = loop.call_later(2.0, task.cancel)
<p>try:
result = await task
except asyncio.CancelledError:
print("cancelled by timer")
finally:
handle.cancel() # 必须手动取消定时器，否则可能触发已结束任务的 cancel()</p>除非你在写底层库或调试事件循环行为，否则没必要绕过 wait_for。
超时嵌套与子协程取消传播要注意什么
如果被 wait_for 包裹的协程内部又 await 了另一个协程，而那个协程本身也用了 wait_for，超时取消会逐层向内传播。但前提是：各层都未用 shield 或忽略取消。
容易被忽略的是：某些异步库（如早期版本的 aiomysql 或自定义协议解析器）在 await 底层 socket 时未适配 CancelledError，导致即使外层超时，内层仍在等待，最终整个任务卡住。
验证方法是在协程开头加一句：
if asyncio.current_task().cancelled():
raise asyncio.CancelledError
并在每个 await 后检查返回值或异常。
另外，wait_for 的 timeout 是针对整个协程执行时间，不是某一次 await 的耗时。如果你需要“每次网络请求单独超时”，得在每次 await 前套一层 wait_for，而不是只包最外层。
