Python 异步代码如何正确处理超时?

12次阅读

python异步超时处理核心是用asyncio.wait_for()包裹协程并指定timeout;不能对await加timeout参数。它会取消任务并抛出asyncio.TimeoutError,但要求协程本身真正异步(如用asyncio.sleep而非time.sleep、aiohttp而非requests)。

Python 异步代码如何正确处理超时?

Python 异步代码中处理超时,核心是用 asyncio.wait_for() 包裹协程,并指定超时时间;不能直接对 await 表达式加 timeout= 参数——它不支持。

asyncio.wait_for() 包裹协程

这是最标准、最可靠的方式。它会在超时后取消被等待的协程任务,并抛出 asyncio.TimeoutError

示例:

import asyncio 

async def fetch_data(): await asyncio.sleep(3) # 模拟耗时操作 return "done"

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 请求用 aiohttphttpx.AsyncClient,别用 requests
  • 数据库操作用异步驱动(如 asyncpgaiomysql

为整个 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 

@with_timeout(3) async def risky_api_call(): await asyncio.sleep(4) return "success"

调用时直接 await risky_api_call() 即可,异常类型更明确。

text=ZqhQzanResources