Python中的协程任务取消与超时处理机制

在异步编程中，协程任务的取消和超时处理是保证程序健壮性的重要机制。当某个异步操作执行时间过长或不再需要时，我们需要能够安全地终止它。

1. 任务取消的基本概念

• 在asyncio中，每个运行的协程都被封装在Task对象中
• 取消任务是通过调用Task.cancel()方法实现的
• 取消操作实际上是向任务内部发送一个CancelledError异常

2. 任务取消的具体流程

import asyncio

async def long_running_task():
    try:
        await asyncio.sleep(10)  # 模拟长时间运行的操作
        print("任务正常完成")
    except asyncio.CancelledError:
        print("任务被取消")
        raise  # 必须重新抛出异常

async def main():
    task = asyncio.create_task(long_running_task())
    await asyncio.sleep(1)  # 等待1秒
    task.cancel()  # 取消任务
    
    try:
        await task
    except asyncio.CancelledError:
        print("主程序捕获到取消异常")

3. 超时处理的实现方式

• 使用asyncio.wait_for()设置单个操作的超时
• 使用asyncio.wait()配合timeout参数处理多个操作的超时

4. 单个任务的超时控制

async def slow_operation():
    await asyncio.sleep(5)  # 模拟耗时操作
    return "操作完成"

async def main():
    try:
        # 设置3秒超时
        result = await asyncio.wait_for(slow_operation(), timeout=3.0)
        print(f"结果: {result}")
    except asyncio.TimeoutError:
        print("操作超时")

5. 多个任务的超时控制

async def task_with_timeout(task_func, timeout):
    try:
        return await asyncio.wait_for(task_func(), timeout=timeout)
    except asyncio.TimeoutError:
        return f"任务在{timeout}秒后超时"

async def concurrent_tasks():
    tasks = [
        task_with_timeout(lambda: asyncio.sleep(2, "任务1"), 3),
        task_with_timeout(lambda: asyncio.sleep(4, "任务2"), 3)
    ]
    
    results = await asyncio.gather(*tasks, return_exceptions=True)
    for i, result in enumerate(results):
        print(f"任务{i+1}: {result}")

6. 取消操作的资源清理

• 在任务被取消时，需要确保资源得到正确释放
• 使用try-finally或异步上下文管理器

async def resource_intensive_task():
    # 模拟资源分配
    resource = acquire_resource()
    try:
        while True:
            await asyncio.sleep(1)
            print("任务执行中...")
    except asyncio.CancelledError:
        print("收到取消信号，开始清理资源...")
        # 执行清理操作
        release_resource(resource)
        raise
    finally:
        # 确保资源被释放
        if resource:
            release_resource(resource)

7. 屏蔽取消的保护机制

• 在某些关键操作中，可能需要临时屏蔽取消操作
• 使用asyncio.shield()保护重要操作不被中断

async def critical_operation():
    await asyncio.sleep(2)
    return "关键操作完成"

async def main():
    critical_task = asyncio.create_task(critical_operation())
    
    # 使用shield保护关键操作
    try:
        result = await asyncio.wait_for(
            asyncio.shield(critical_task), 
            timeout=1.0
        )
    except asyncio.TimeoutError:
        print("外部操作超时，但关键操作仍在继续")
        # 等待关键操作完成
        result = await critical_task
    
    print(f"最终结果: {result}")

8. 实际应用中的最佳实践

• 为每个异步操作设置合理的超时时间
• 在任务取消时进行适当的资源清理
• 使用return_exceptions=True避免单个任务异常影响其他任务
• 对关键操作使用shield进行保护

这种机制确保了异步程序的稳定性和可靠性，是构建健壮异步应用的基础。