Python取消操作更方便了吗

wen python案例 1

本文目录导读:

Python取消操作更方便了吗

  1. asyncio 中的任务取消(Python 3.9+)
  2. TaskGroup 结构(Python 3.11+)
  3. 超时取消更方便
  4. contextvars.ContextVar 的取消传播
  5. 更清晰的异常处理

是的,Python的取消操作在近年来确实变得更方便了,主要体现在以下几个方面:

asyncio 中的任务取消(Python 3.9+)

import asyncio
async def my_task():
    try:
        await asyncio.sleep(10)
    except asyncio.CancelledError:
        print("任务被取消")
        raise  # 重新抛出以完成取消
async def main():
    task = asyncio.create_task(my_task())
    await asyncio.sleep(1)
    task.cancel()  # 取消任务
    await task  # 这会抛出 CancelledError
asyncio.run(main())

TaskGroup 结构(Python 3.11+)

async def main():
    async with asyncio.TaskGroup() as tg:
        task1 = tg.create_task(task1_func())
        task2 = tg.create_task(task2_func())
        # 任何任务失败,所有任务都会被取消

超时取消更方便

# Python 3.11+ 使用 asyncio.timeout
try:
    async with asyncio.timeout(5):
        await long_running_operation()
except TimeoutError:
    print("操作超时")

contextvars.ContextVar 的取消传播

import contextvars
cancel_flag = contextvars.ContextVar('cancel_flag', default=False)
async def cancellable_task():
    for item in range(100):
        if cancel_flag.get():
            raise asyncio.CancelledError()
        await process(item)

更清晰的异常处理

# Python 3.12+ 更好的 CancelledError 堆栈跟踪
try:
    await some_cancellable_op()
except* asyncio.CancelledError:
    print("操作被取消")

最显著的改进是:

  • TaskGroup 让批量取消更简单
  • asyncio.timeout() 让超时取消更直观
  • 更好的异常处理 让取消逻辑更清晰

这些改进使得编写健壮的异步代码和处理取消操作比以往更方便了。

抱歉,评论功能暂时关闭!