Python异步超时处理核心是用asyncio.wait_for()包裹协程并指定timeout；不能对await加timeout参数。它会取消任务并抛出asyncio.TimeoutError，但要求协程本身真正异步（如用asyncio.sleep而非time.sleep、aiohttp而非requests）。
Python 异步代码中处理超时，核心是用 asyncio.wait_for() 包裹协程，并指定超时时间；不能直接对 await 表达式加 timeout= 参数——它不支持。
用 asyncio.wait_for() 包裹协程
这是最标准、最可靠的方式。它会在超时后取消被等待的协程任务，并抛出 asyncio.TimeoutError。
示例：
import asyncio
<p>async def fetch_data():
await asyncio.sleep(3) # 模拟耗时操作
return "done"</p><p>async def main():
try:
result = await asyncio.wait_for(fetch_data(), timeout=2.0)
print(result)
except asyncio.TimeoutError:
print("请求超时，已取消")
注意协程是否可取消
不是所有协程都能被及时中断。如果内部用了阻塞调用（如 time.sleep()、未封装的 requests.get()），wait_for 取消后仍会继续执行，只是 await 返回失败。必须确保协程是真正异步的。
立即学习“Python免费学习笔记（深入）”；
- 用
asyncio.sleep()替代time.sleep() - HTTP 请求用
aiohttp或httpx.AsyncClient，别用requests - 数据库操作用异步驱动（如
asyncpg、aiomysql）
为整个 async with 或嵌套 await 加超时
超时作用域要覆盖完整逻辑块。比如在 async with aiohttp.ClientSession() as session: 内部多个 await session.get(...)，应分别对每个请求加 wait_for，或把整个请求+解析逻辑封装成一个协程再包一层。
错误示范（只对 session 创建加超时，没管实际请求）：
# ❌ 错误：超时只作用于 session 创建，不保护后续 await
async with await asyncio.wait_for(aiohttp.ClientSession(), timeout=1):
resp = await session.get(url) # 这里仍可能卡住
正确做法：
# ✅ 正确：对每个实际 I/O 操作设超时
async with aiohttp.ClientSession() as session:
try:
async with asyncio.wait_for(session.get(url), timeout=5) as resp:
data = await resp.text()
except asyncio.TimeoutError:
print("获取响应超时")
自定义超时装饰器（可选进阶）
若多个协程需统一超时逻辑，可写个简单装饰器：
def with_timeout(seconds):
def decorator(coro_func):
async def wrapper(*args, **kwargs):
try:
return await asyncio.wait_for(
coro_func(*args, **kwargs), timeout=seconds
)
except asyncio.TimeoutError:
raise TimeoutError(f"{coro_func.__name__} 超时 ({seconds}s)")
return wrapper
return decorator
<p>@with_timeout(3)
async def risky_api_call():
await asyncio.sleep(4)
return "success"
调用时直接 await risky_api_call() 即可，异常类型更明确。
