避免在Python异步编程中出现CancelledError()异常的五个技巧

在Python异步编程中，CancelledError异常通常在协程被取消时引发。它是一个常见的异常，因为在并发环境中很容易取消正在执行的协程。为了避免CancelledError异常的出现，可以采用以下五个技巧：

1. 使用try-except块捕获CancelledError异常：

import asyncio

async def coroutine():
    try:
        while True:
            await asyncio.sleep(1)
    except asyncio.CancelledError:
        print("Coroutine cancelled")

async def main():
    task = asyncio.create_task(coroutine())
    await asyncio.sleep(5)
    task.cancel()
    await task

asyncio.run(main())

在上面的例子中，我们在coroutine函数中使用try-except块来捕获CancelledError异常。当协程被取消时，except块将被执行，打印出"Coroutine cancelled"。

2. 在协程中使用ensure_future()或create_task()而不是直接使用run_until_complete()来调度协程的执行：

import asyncio

async def coroutine():
    while True:
        await asyncio.sleep(1)

async def main():
    coro = asyncio.ensure_future(coroutine())
    await asyncio.sleep(5)
    coro.cancel()
    await coro

asyncio.run(main())

在上面的例子中，我们使用ensure_future()函数来创建一个协程对象，然后使用cancel()方法取消它。这样做可以避免使用run_until_complete()方法，因为它在协程被取消时会引发CancelledError异常。

3. 使用add_done_callback()方法来处理取消事件：

import asyncio

async def coroutine():
    while True:
        await asyncio.sleep(1)

def handle_cancel(task):
    print("Coroutine cancelled")

async def main():
    coro = asyncio.ensure_future(coroutine())
    coro.add_done_callback(handle_cancel)
    await asyncio.sleep(5)
    coro.cancel()
    await coro

asyncio.run(main())

在上面的例子中，我们使用add_done_callback()方法将handle_cancel函数注册为协程完成时的回调函数。当协程被取消时，回调函数将被执行。

4. 在协程中使用asyncio.shield()函数保护关键部分：

import asyncio

async def coroutine():
    while True:
        await asyncio.sleep(1)
        await asyncio.shield(do_something())

async def do_something():
    # Some critical code
    pass

async def main():
    task = asyncio.create_task(coroutine())
    await asyncio.sleep(5)
    task.cancel()
    await task

asyncio.run(main())

在上面的例子中，我们在coroutine函数中使用了asyncio.shield()函数来保护do_something()函数的关键部分。这样做可以确保在协程被取消时，关键部分不会被中断。

5. 使用asyncio.TimeoutError异常替代CancelledError异常：

import asyncio

async def coroutine():
    try:
        while True:
            await asyncio.sleep(1)
    except asyncio.TimeoutError:
        print("Coroutine cancelled")

async def main():
    task = asyncio.create_task(coroutine())
    await asyncio.sleep(5)
    task.cancel()
    await task

asyncio.run(main())

在上面的例子中，我们将try-except块中的CancelledError异常替换为asyncio.TimeoutError异常。这样做是为了将CancelledError异常从应用程序逻辑中分离出来，并使代码更具可读性。

通过以上五个技巧，可以在Python异步编程中避免CancelledError异常的出现。无论是捕获异常、使用适当的调度方法、注册回调函数、保护关键部分还是使用替代异常，都是为了确保协程的正常执行。