Python 异步编程进阶：Asyncio 任务调度与异常处理最佳实践

魁首哥

作者

在现代应用开发中，异步编程已经成为提升性能和用户体验的重要手段。Python 的 asyncio 框架作为异步编程的核心库，提供了丰富的功能和灵活的调度机制。然而，对于开发者来说，如何高效地管理任务调度以及优雅地处理异常，仍然是一个需要深入理解的课题。本文将围绕这两个主题，分享一些最佳实践和实用技巧。

一、Asyncio 任务调度：让代码更高效

1. 任务调度的基本概念

在 asyncio 中，任务（Task）是协程（Coroutine）的封装，用于跟踪协程的执行状态。通过任务调度，我们可以同时执行多个协程，从而充分利用系统的资源。

常见的任务调度方式包括：

asyncio.create_task()：直接将协程包装为任务并添加到事件循环中。
asyncio.gather()：将多个协程或任务组合成一个任务组，等待所有任务完成。
asyncio.TaskGroup()：在 Python 3.11+ 中引入的新特性，提供更简洁的任务管理方式。

2. 任务调度的最佳实践

（1）合理使用 `TaskGroup`

TaskGroup 是 asyncio 的一大亮点，能够简化任务的创建和管理。它通过上下文管理器的方式，自动处理任务的启动和等待。

import asyncioasync def task_func(name):    print(f"Task {name} started")    await asyncio.sleep(1)    print(f"Task {name} completed")async def main():    async with asyncio.TaskGroup() as tg:        tg.create_task(task_func("A"))        tg.create_task(task_func("B"))    print("All tasks completed")asyncio.run(main())

优势：

自动管理任务，避免手动等待。
支持异常传播，任务中的异常会自动传递给 main() 函数。

（2）避免任务失控

在复杂的场景中，如果不小心忘记等待任务完成，可能会导致任务被丢弃。为了避免这种情况，可以使用 asyncio.gather() 显式地收集所有任务。

async def main():    task1 = asyncio.create_task(task_func("A"))    task2 = asyncio.create_task(task_func("B"))    await asyncio.gather(task1, task2)

注意事项：

如果任务数量较多，建议使用 TaskGroup 代替手动管理。
对于长时间运行的任务，可以考虑设置超时机制。

（3）任务优先级与资源管理

在某些场景中，可能需要为任务设置优先级，或者限制资源的使用。asyncio 提供了 asyncio.PriorityQueue 和 asyncio.Semaphore 等工具，帮助开发者更好地管理任务。

import asyncioasync def worker(name, semaphore):    async with semaphore:        print(f"Worker {name} is working")        await asyncio.sleep(2)        print(f"Worker {name} finished")async def main():    max_workers = 2    semaphore = asyncio.Semaphore(max_workers)    tasks = [asyncio.create_task(worker(f"Worker {i}", semaphore)) for i in range(4)]    await asyncio.gather(*tasks)asyncio.run(main())

解释：

Semaphore 用于限制同时执行的任务数量。
通过这种方式，可以避免资源被过度占用。

二、Asyncio 异常处理：优雅应对意外情况

1. 异常处理的基本方法

在 asyncio 中，协程的异常处理与普通函数类似，但需要注意以下几点：

协程内部的异常不会自动传播到事件循环。
必须在协程内部或外部显式地处理异常。

（1）在协程内部捕获异常

async def task_func():    try:        await asyncio.sleep(1)        1 / 0  # 故意引发异常    except ZeroDivisionError:        print("ZeroDivisionError caught")async def main():    task = asyncio.create_task(task_func())    await taskasyncio.run(main())

（2）在任务外部捕获异常

如果不想在协程内部处理异常，可以在任务完成后检查其结果。

async def task_func():    await asyncio.sleep(1)    1 / 0  # 故意引发异常async def main():    task = asyncio.create_task(task_func())    await task    if task.exception() is not None:        print(f"Task failed with exception: {task.exception()}")asyncio.run(main())

2. 异常处理的最佳实践

（1）使用 `add_done_callback`

通过 add_done_callback，可以在任务完成后执行回调函数，从而优雅地处理异常。

async def task_func():    await asyncio.sleep(1)    1 / 0  # 故意引发异常def handle_result(task):    if task.exception() is not None:        print(f"Task failed with exception: {task.exception()}")async def main():    task = asyncio.create_task(task_func())    task.add_done_callback(handle_result)    await taskasyncio.run(main())

优势：

回调函数可以在任务完成后立即执行。
适用于需要异步处理异常的场景。

（2）避免全局异常捕获

虽然可以使用 asyncio.get_event_loop().set_exception_handler() 设置全局异常处理器，但这种方式并不推荐，因为它会掩盖潜在的问题。

（3）结合 `TaskGroup` 处理异常

TaskGroup 可以自动传播任务中的异常，简化异常处理的逻辑。

async def task_func():    await asyncio.sleep(1)    1 / 0  # 故意引发异常async def main():    try:        async with asyncio.TaskGroup() as tg:            tg.create_task(task_func())            tg.create_task(task_func())    except Exception as e:        print(f"Exception caught in main: {e}")asyncio.run(main())

解释：