冲刺模拟期 20 天规划：Python 异步进阶 - asyncio 任务取消与超时处理

在蓝桥杯的备考过程中，Python 的异步编程是一个重要的知识点，尤其是在处理高并发和 I/O 密集型任务时。本文将详细讲解 asyncio 库中的任务取消与超时处理，帮助考生在 20 天的冲刺模拟期内高效掌握这一内容。

一、asyncio 任务取消

在异步编程中，任务取消是一个常见的需求。asyncio 提供了 cancel() 方法来实现任务的取消。以下是 cancel() 方法的基本用法：

import asyncio

async def my_task():
    try:
        await asyncio.sleep(10)
        print("Task completed")
    except asyncio.CancelledError:
        print("Task was cancelled")

async def main():
    task = asyncio.create_task(my_task())
    await asyncio.sleep(2)
    task.cancel()
    try:
        await task
    except asyncio.CancelledError:
        print("Main function also knows the task was cancelled")

asyncio.run(main())

在这个例子中，my_task 任务会在执行 2 秒后被取消，并捕获 CancelledError 异常。

学习方法

理解 cancel() 方法的工作原理：cancel() 方法会向任务发送一个取消请求，任务需要在适当的地方捕获 CancelledError 异常来处理取消逻辑。
实践编写取消逻辑：通过编写不同场景下的任务取消代码，熟悉如何处理取消请求。

二、asyncio 超时处理

在异步编程中，超时处理同样重要。asyncio 提供了 wait_for() 函数来实现任务的超时控制。以下是 wait_for() 函数的基本用法：

import asyncio

async def my_task():
    await asyncio.sleep(10)
    print("Task completed")

async def main():
    try:
        await asyncio.wait_for(my_task(), timeout=2)
    except asyncio.TimeoutError:
        print("Task timed out")

asyncio.run(main())

在这个例子中，my_task 任务会在执行 2 秒后被强制终止，并抛出 TimeoutError 异常。

学习方法

理解 wait_for() 函数的工作原理：wait_for() 函数会在指定时间内等待任务完成，如果超时则会抛出 TimeoutError 异常。
实践编写超时处理逻辑：通过编写不同场景下的超时处理代码，熟悉如何处理超时异常。

三、异步任务异常处理逻辑

在异步任务中，异常处理是一个关键环节。无论是任务取消还是超时处理，都需要正确捕获和处理异常。以下是一个综合示例：

import asyncio

async def my_task():
    try:
        await asyncio.sleep(10)
        print("Task completed")
    except asyncio.CancelledError:
        print("Task was cancelled")
    except Exception as e:
        print(f"Task encountered an error: {e}")

async def main():
    try:
        await asyncio.wait_for(my_task(), timeout=2)
    except asyncio.TimeoutError:
        print("Task timed out")

asyncio.run(main())

在这个例子中，my_task 任务会在执行 2 秒后被取消，并捕获 CancelledError 异常。