3. 异步任务超时控制

任务说明

在异步编程中，协程可能因网络请求、外部服务调用等原因长时间无响应。本案例要求使用 asyncio.wait_for 为协程设置超时时间，当协程执行超过指定时限时自动取消任务，并妥善处理 asyncio.TimeoutError 异常，确保程序不会因单个任务卡死而阻塞整个事件循环。

核心实现

import asyncio

async def slow_operation(task_id: int, delay: float) -> str:
    """模拟一个耗时操作的协程"""
    await asyncio.sleep(delay)
    return f"任务 {task_id} 完成，耗时 {delay} 秒"

async def run_with_timeout(task_id: int, delay: float, timeout: float) -> None:
    """为协程设置超时，超时后取消任务并处理异常"""
    try:
        result = await asyncio.wait_for(slow_operation(task_id, delay), timeout=timeout)
        print(f"成功: {result}")
    except asyncio.TimeoutError:
        print(f"任务 {task_id} 执行超过 {timeout} 秒，已取消")

async def main() -> None:
    # 场景1: 任务在超时前完成
    await run_with_timeout(task_id=1, delay=1.0, timeout=3.0)
    # 场景2: 任务超时被取消
    await run_with_timeout(task_id=2, delay=5.0, timeout=2.0)
    # 场景3: 多个任务并发执行，各自独立超时
    await asyncio.gather(
        run_with_timeout(task_id=3, delay=2.0, timeout=1.0),
        run_with_timeout(task_id=4, delay=0.5, timeout=3.0),
    )

if __name__ == "__main__":
    asyncio.run(main())

关键机制说明

asyncio.wait_for(aw, timeout) 接收一个可等待对象和超时秒数。当超时发生时，wait_for 会向内部任务发送取消请求（调用 task.cancel()），随后抛出 asyncio.TimeoutError。被取消的协程在其挂起点（如 await asyncio.sleep()）收到 CancelledError，从而终止执行。通过 try/except asyncio.TimeoutError 捕获超时异常，程序可以继续处理后续逻辑，不会因单个任务超时而崩溃。

SVG：超时控制时序

运行与验证

运行方式：将代码保存为 timeout_demo.py，在命令行执行 python timeout_demo.py（需 Python 3.7+）。

预期输出：

成功: 任务 1 完成，耗时 1.0 秒
任务 2 执行超过 2.0 秒，已取消
任务 3 执行超过 1.0 秒，已取消
成功: 任务 4 完成，耗时 0.5 秒

判定标准：程序正常退出无异常堆栈；任务1和任务4在超时前完成并打印成功信息；任务2和任务3因超过各自超时时间被取消，打印对应的超时提示；asyncio.TimeoutError 被正确捕获，事件循环未因任务取消而中断。