python异步超时控制用asyncio.wait_for()(3.7+)或asyncio.timeout()(3.11+),可中断协程;多任务统一超时用asyncio.wait()的timeout参数,但需手动取消pending任务。

Python中用asyncio做异步编程时,超时控制不是靠time.sleep()或简单计时器,而是通过asyncio.wait_for()和asyncio.timeout()(Python 3.11+)来实现——它们能真正中断挂起的协程,避免任务无限等待。
用asyncio.wait_for()设置超时(兼容3.7+)
这是最常用、最稳妥的方式。它包装一个协程,在指定时间内未完成就抛出asyncio.TimeoutError。
说明:
- 第二个参数是秒数(支持浮点数,如2.5);
- 超时后原协程会被取消(自动调用cancel()),但需确保协程内有合理清理逻辑(比如关闭连接);
- 若被包装的协程本身已结束,wait_for直接返回结果,不触发超时。
示例:
import asyncio
<p>async def fetch_data():
await asyncio.sleep(5) # 模拟慢请求
return "done"</p><p>async def main():
try:
result = await asyncio.wait_for(fetch_data(), timeout=3)
print(result)
except asyncio.TimeoutError:
print("请求超时,已取消")</p><p>asyncio.run(main())
用asyncio.timeout()上下文管理器(Python 3.11+)
更直观、更符合“限时执行某段逻辑”的语义,推荐新项目使用。
立即学习“Python免费学习笔记(深入)”;
说明:
- 是异步上下文管理器,需配合async with;
- 超时后自动取消当前任务,并抛出TimeoutError(注意不是asyncio.TimeoutError);
- 可嵌套使用,内层超时优先于外层;
- 支持传入None表示禁用超时(方便动态控制)。
示例:
import asyncio
<p>async def risky_operation():
await asyncio.sleep(4)
return "success"</p><p>async def main():
try:
async with asyncio.timeout(2):
result = await risky_operation()
print(result)
except TimeoutError:
print("操作在2秒内未完成")</p><p>asyncio.run(main())
对多个协程统一设超时(asyncio.wait() + timeout)
当并发运行多个任务,且希望整体不超过某时限(而非每个单独设限),可用asyncio.wait()的timeout参数。
说明:
- timeout是总等待时间,不是单个任务的;
- 超时后未完成的任务仍处于pending状态,不会被自动取消(需手动处理);
- 返回值是(done, pending)集合,可遍历pending并调用cancel()。
建议做法:
- 用
asyncio.create_task()显式创建任务,便于后续取消 - 超时后检查
pending,逐个cancel()并await其完成(避免“task destroyed but it is pending”警告)
注意事项与常见陷阱
超时不是万能的,容易忽略的关键点:
-
阻塞调用无法被中断:比如在协程里写了
time.sleep(10)或调用了同步IO(如requests.get()),wait_for或timeout只能等它自己返回,无法强制退出 -
取消需要协程配合:若协程内部没响应取消信号(比如没检查
task.cancelled()或没用await asyncio.sleep(0)让出控制权),可能无法及时终止 -
I/O类库需选异步版本:用
aiohttp代替requests,用aiomysql代替pymysql,才能真正享受超时中断能力 -
超时单位是秒,不是毫秒:传
0.1代表100毫秒,别误写成100










