异步编程中：asyncio任务被取消时该捕获CancelledError还是Exception？

import asyncio

async def my_task():
    try:
        print("任务开始执行")
        await asyncio.sleep(5)  # 模拟耗时操作
        print("任务执行完成")
    except asyncio.CancelledError:
        print("任务被取消了！")
        # 在这里进行清理工作，例如关闭文件、释放资源等
    finally:
        print("无论如何都会执行的清理工作")

async def main():
    task = asyncio.create_task(my_task())
    await asyncio.sleep(1)  # 等待一段时间后取消任务
    task.cancel()
    try:
        await task  # 等待任务结束，会抛出 CancelledError
    except asyncio.CancelledError:
        print("main 函数也捕获了 CancelledError")

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

import asyncio

async def child_task():
    try:
        print("子任务开始执行")
        await asyncio.sleep(3)
        print("子任务执行完成")
    except asyncio.CancelledError:
        print("子任务被取消了！")
        await asyncio.sleep(1) # 模拟清理时间
        print("子任务清理完成")

async def parent_task():
    try:
        print("父任务开始执行")
        task = asyncio.create_task(child_task())
        await asyncio.sleep(1)
        print("父任务准备取消子任务")
        task.cancel()
        await task  # 等待子任务结束
        print("父任务等待子任务取消完成")
    except asyncio.CancelledError:
        print("父任务也被取消了！")

async def main():
    await parent_task()

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

import asyncio

async def task_a(lock):
    async with lock:
        # 访问或修改共享状态
        print("Task A acquired the lock")
        await asyncio.sleep(1)
        print("Task A releasing the lock")

async def task_b(lock):
    async with lock:
        # 访问或修改共享状态
        print("Task B acquired the lock")
        await asyncio.sleep(1)
        print("Task B releasing the lock")

async def main():
    lock = asyncio.Lock()
    task1 = asyncio.create_task(task_a(lock))
    task2 = asyncio.create_task(task_b(lock))

    await asyncio.sleep(0.5)
    task2.cancel()

    await asyncio.gather(task1, task2, return_exceptions=True)

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