Python中的异步编程陷阱与解决方案：任务取消、异常处理与资源管理

知识点描述：
在Python异步编程（尤其使用asyncio时），开发者常遇到一些隐蔽的陷阱，涉及任务取消的传播机制、异常处理的不同行为、资源泄露风险等。理解这些陷阱的底层原理和标准解决方案，是编写健壮异步代码的关键。

详细讲解：

一、任务取消的传播陷阱

1. 问题现象：

   import asyncio
   
   async def subtask():
       try:
           await asyncio.sleep(10)
       except asyncio.CancelledError:
           print("Subtask cancelled")
           raise
   
   async def main():
       task = asyncio.create_task(subtask())
       await asyncio.sleep(0.1)
       task.cancel()
       await task  # 等待任务完成
   
   asyncio.run(main())
   # 输出: Subtask cancelled
   # 然后程序正常结束
   

2. 陷阱分析：

   • 当调用task.cancel()时，会在协程内部抛出CancelledError
   • 但如果在捕获CancelledError后不重新抛出，或者进行了其他处理，任务实际上不会被取消

   async def bad_subtask():
       try:
           await asyncio.sleep(10)
       except asyncio.CancelledError:
           print("捕获了取消异常，但不重新抛出")
           # 这里不重新抛出，任务会继续运行！
   
   async def main():
       task = asyncio.create_task(bad_subtask())
       await asyncio.sleep(0.1)
       task.cancel()
       try:
           await task
           print("任务'完成'了")  # 这行会执行！
       except asyncio.CancelledError:
           print("任务被取消了")  # 这行不会执行
   

3. 解决方案：

   • 始终重新抛出CancelledError，除非有特殊理由
   • 使用finally块进行清理

   async def good_subtask():
       try:
           await asyncio.sleep(10)
       except asyncio.CancelledError:
           print("执行取消清理")
           raise  # 必须重新抛出！
       finally:
           print("最终清理")
   

二、异常处理的隐蔽行为

1. 未捕获异常的问题：

   async def failing_task():
       raise ValueError("任务内部异常")
   
   async def main():
       task = asyncio.create_task(failing_task())
       await asyncio.sleep(1)  # 异常被记录但未传播
       print("主程序继续运行")
   
   asyncio.run(main())
   # 异常被记录到stderr，但程序继续运行
   

2. 陷阱的根本原因：

   • 任务的异常不会自动传播到创建它的上下文
   • 必须显式地await任务或检查task.exception()

3. 推荐解决方案：

   async def safe_main():
       task = asyncio.create_task(failing_task())
       try:
           await task
       except Exception as e:
           print(f"捕获到任务异常: {e}")
   
   # 或者使用任务组（Python 3.11+）
   async def with_taskgroup():
       async with asyncio.TaskGroup() as tg:
           task1 = tg.create_task(failing_task())
           task2 = tg.create_task(another_task())
       # TaskGroup会等待所有任务完成，并传播第一个异常
   

三、资源管理陷阱

1. 异步上下文管理器使用不当：

   async def bad_resource_usage():
       # 忘记使用async with
       writer = asyncio.open_connection("localhost", 8888)
       # 如果这里发生异常，连接永远不会关闭
       await writer
   
   async def good_resource_usage():
       # 正确方式
       async with asyncio.open_connection("localhost", 8888) as (reader, writer):
           # 确保连接会被正确关闭
           await writer.write(b"data")
   

2. 信号量泄露陷阱：

   async def with_semaphore(sem):
       # 错误：如果await操作被取消，semaphore不会释放
       await sem.acquire()
       try:
           await do_work()
       finally:
           sem.release()
   
   async def correct_semaphore_usage(sem):
       # 正确：使用async with确保释放
       async with sem:
           await do_work()
   

四、asyncio.wait的常见陷阱

1. 不指定return_when参数：

   async def risky_wait():
       tasks = [task1(), task2(), task3()]
       # 默认return_when=FIRST_COMPLETED
       done, pending = await asyncio.wait(tasks)
       # pending中的任务仍在运行，可能造成资源泄露
   
   async def safe_wait():
       tasks = [task1(), task2(), task3()]
       # 明确指定等待所有任务
       done, pending = await asyncio.wait(tasks, return_when=asyncio.ALL_COMPLETED)
       # 或者明确取消pending任务
       done, pending = await asyncio.wait(tasks, return_when=asyncio.FIRST_COMPLETED)
       for task in pending:
           task.cancel()
   

五、最佳实践总结

1. 任务取消规范：

   • 总是重新抛出CancelledError，除非明确要忽略取消
   • 在finally块中清理资源

2. 异常处理规范：

   • 总是await或检查任务结果
   • 使用asyncio.TaskGroup（Python 3.11+）管理相关任务
   • 设置异常处理器：loop.set_exception_handler

3. 资源管理规范：

   • 对支持异步上下文管理器的资源，总是使用async with
   • 避免手动调用acquire()/release()，优先使用上下文管理器

4. 超时处理：

   async def with_timeout():
       try:
           async with asyncio.timeout(5.0):  # Python 3.11+
               await long_operation()
       except TimeoutError:
           print("操作超时")
   
   # Python 3.10及以下
   async def with_timeout_old():
       try:
           await asyncio.wait_for(long_operation(), timeout=5.0)
       except asyncio.TimeoutError:
           print("操作超时")
   

5. 任务监控：

   async def monitor_tasks():
       task = asyncio.create_task(risky_operation())
       # 添加完成回调
       task.add_done_callback(lambda t: 
           print(f"Task completed with {t.exception()}" if t.exception() else "Task succeeded"))
       return await task
   

通过理解这些陷阱和遵循最佳实践，可以编写出更健壮、可维护的异步Python代码，避免资源泄露、未处理异常和不可预测的行为。