使用 Asyncio 创建和管理任务

asyncio 允许开发者轻松地用 python 编写异步程序。该模块还提供了多种异步任务的方法，并且由于执行方法多种多样，因此可能会让人困惑于使用哪一种。

在本文中，我们将讨论使用 asyncio 创建和管理任务的多种方法。

什么是异步任务？

在 asyncio 中，task 是一个包装协程并安排其在事件循环内运行的对象。简而言之，任务是一种与其他任务同时运行协程的方式。创建任务后，事件循环将运行它，并根据需要暂停和恢复它以允许其他任务运行。

创建和管理 asyncio 任务的方法

现在，我们可以讨论创建和管理任务的方法。首先，要使用 asyncio 在 python 中创建任务，请使用 asyncio.create_task 方法，该方法采用以下参数：

• coro（必填）：要调度的协程对象。这是你想要异步运行的函数。

• 名称（可选）：可用于调试或日志记录目的的任务名称。您可以为该参数分配一个字符串。

  • 您还可以稍后使用 task.set_name(name) 和 task.get_name() 设置或获取名称。

• context（可选）：在python 3.11中引入，用于为任务设置上下文变量，从而启用任务本地存储。它类似于线程本地存储，但用于异步任务。

  • 这个参数并不常用，除非你正在处理需要上下文管理的高级场景。

这里是asyncio.create_task的使用示例：

import asyncio

# define a coroutine
async def greet(name):
    await asyncio.sleep(1)  # simulate an i/o-bound operation
    print(f"hello, {name}!")

async def main():
    # create tasks
    task1 = asyncio.create_task(greet("alice"), name="greetingalice")
    task2 = asyncio.create_task(greet("bob"), name="greetingbob")

    # check task names
    print(f"task 1 name: {task1.get_name()}")
    print(f"task 2 name: {task2.get_name()}")

    # wait for both tasks to complete
    await task1
    await task2

# run the main function
asyncio.run(main())

创建任务时，可以执行很多方法，例如：

• .cancel(): 取消任务。

• .add_done_callback(cb): 添加任务完成时运行的回调函数。

• .done(): 检查任务是否完成。

• .result()：任务完成后检索结果。

现在我们了解了如何创建任务，让我们看看如何处理等待一个任务或多个任务。

等待任务完成

在本节中，我们将讨论如何等待一个或多个任务的任务完成。异步编程基于这样一个事实：如果正在运行异步任务，我们可以继续执行程序。有时您可能想要更好地控制流程，并希望确保在安全地继续执行程序之前获得可以使用的结果。

要等待单个任务完成，可以使用asyncio.wait_for。它需要两个参数：

• awaitable（必需）：这是您想要等待的协程、任务或未来。它可以是任何可以等待的对象，例如协程函数调用、asyncio.task 或 asyncio.future。

• timeout（可选）：这指定等待 aw 完成的最大秒数。如果达到超时并且等待尚未完成，asyncio.wait_for 会引发 timeouterror。如果超时设置为 none，该函数将无限期地等待等待完成。

这是使用此方法的示例：

import asyncio

async def slow_task():
    print("task started...")
    await asyncio.sleep(5)  # simulating a long-running task
    print("task finished!")
    return "completed"

async def main():
    try:
        # wait for slow_task to finish within 2 seconds
        result = await asyncio.wait_for(slow_task(), timeout=2)
        print(result)
    except asyncio.timeouterror:
        print("the task took too long and was canceled!")

asyncio.run(main())

上面的代码中，slow_task() 是一个协程，通过休眠 5 秒来模拟长时间运行的任务。 asyncio.wait_for(slow_task(), timeout=2) 行等待任务完成，但将等待时间限制为 2 秒，从而导致超时，因为任务需要更长的时间。当超过超时时，会引发 timeouterror，任务被取消，并通过打印一条指示任务花费太长时间的消息来处理异常。

我们还可以等待多个或一组任务完成。这可以使用 asyncio.wait、asyncio.gather 或 asyncio.as_completed 来实现。让我们探索每种方法。

异步等待

asyncio.wait 方法等待一组任务并返回两组：一组用于已完成的任务，一组用于待处理的任务。它需要以下参数：

• aws（必需，可迭代）：您想要等待的协程对象、任务或 future 的集合。

• timeout（float 或 none，可选）：等待的最大秒数。如果没有提供，它将无限期地等待。

• return_when （常量，可选）：指定 asyncio.wait 何时返回。选项包括：

  • asyncio.all_completed（默认）：当所有任务完成时返回。
  • asyncio.first_completed：第一个任务完成时返回。
  • asyncio.first_exception：当第一个任务引发异常时返回。

让我们看看它是如何在示例中使用的。

import asyncio
import random

async def task():
    await asyncio.sleep(random.uniform(1, 3))

async def main():
    tasks = [asyncio.create_task(task()) for _ in range(3)]
    done, pending = await asyncio.wait(tasks, return_when=asyncio.first_completed)
    print(f"done tasks: {len(done)}, pending tasks: {len(pending)}")

asyncio.run(main())

在上面的代码中，asyncio.wait 等待一组任务并返回两组：一组包含已完成的任务，另一组包含仍待处理的任务。您可以控制它何时返回，例如第一个任务完成后或所有任务完成后。在示例中，asyncio.wait 在第一个任务完成时返回，将其余任务留在待处理集中。

异步收集

asyncio.gather 方法同时运行多个可等待对象并返回其结果列表，可以选择处理异常。让我们看看它所需要的论据。

• *aws（必需，多个可等待）：并发运行的可变数量的可等待对象（如协程、任务或 future）。

• return_exceptions（bool，可选）：如果为 true，任务中的异常将作为结果列表的一部分返回，而不是引发。

让我们看看如何在示例中使用它。

import asyncio
import random

async def task(id):
    await asyncio.sleep(random.uniform(1, 3))
    return f"task {id} done"

async def main():
    results = await asyncio.gather(task(1), task(2), task(3))
    print(results)

asyncio.run(main())

在上面的代码中，asyncio.gather 同时运行多个可等待对象，并按照传入的顺序返回结果列表。如果 return_exceptions 设置为 true，它允许您优雅地处理异常。在示例中，三个任务同时运行，所有任务完成后，它们的结果将在列表中返回。

asyncio.as_completed 已完成

asyncio.as_completed 方法用于返回一个迭代器，该迭代器在任务完成时生成任务，从而允许立即处理结果。它需要以下参数：

• aws（可迭代的可等待对象）：协程对象、任务或 future 的集合。

• timeout（float 或 none，可选）：等待任务完成的最大秒数。如果没有提供，它将无限期地等待。

例子

import asyncio
import random

async def task(id):
    await asyncio.sleep(random.uniform(1, 3))
    return f"task {id} done"

async def main():
    tasks = [task(i) for i in range(3)]
    for coro in asyncio.as_completed(tasks):
        result = await coro
        print(result)

asyncio.run(main())

在上面的示例中，asyncio.as_completed 返回一个迭代器，该迭代器在每个任务完成时生成结果，使您可以立即处理它们。当您希望在结果可用时立即对其进行处理，而不是等待所有任务完成时，这非常有用。在示例中，任务同时运行，并且在每个任务完成时按完成顺序打印其结果。

因此，为了进行总结，您可以使用：

• asyncio.wait：当您需要处理多个任务并想要跟踪哪些任务已完成、哪些任务仍待处理时。当您分别关心每个任务的状态时，它很有用。

• asyncio.gather：当你想要同时运行多个任务并需要列表中的结果时，特别是当结果的顺序很重要或者你需要优雅地处理异常时。

• asyncio.as_completed：当您想要在每个任务完成后立即处理结果，而不是等待所有任务完成时。这对于按可用顺序处理结果很有用。

但是，这些方法不采用带有内置错误处理的原子任务管理。在下一节中，我们将了解 asyncio.taskgroup 以及如何使用它来管理一组任务。

异步任务组

asyncio.taskgroup 是 python 3.11 中引入的上下文管理器，可简化对多个任务作为一个组的管理。它确保如果组内的任何任务失败，所有其他任务都会被取消，从而提供了一种通过强大的错误处理来处理复杂任务管理的方法。该类有一个名为created_task 的方法，用于创建任务并将其添加到任务组中。您将协程传递给此方法，它会返回一个由组管理的 asyncio.task 对象。

这是如何使用它的示例：

import asyncio

async def task1():
    await asyncio.sleep(1)
    return "Task 1 done"

async def task2():
    await asyncio.sleep(2)
    return "Task 2 done"

async def task_with_error():
    await asyncio.sleep(1)
    raise ValueError("An error occurred")

async def main():
    try:
        async with asyncio.TaskGroup() as tg:
            task1 = tg.create_task(task1())
            task2 = tg.create_task(task2())
            error_task = tg.create_task(task_with_error())
    except Exception as e:
        print(f"Error: {e}")

    # Print results from completed tasks
    print("Task 1 result:", task1.result())
    print("Task 2 result:", task2.result())

asyncio.run(main())

asyncio.taskgroup 管理多个任务，并确保如果任何任务失败，组中的所有其他任务都将被取消。在示例中，出现错误的任务会导致整个组被取消，并且只打印已完成任务的结果。

此用途可以用于网络抓取。您可以使用 asyncio.taskgroup 来处理多个并发 api 请求，并确保如果任何请求失败，则所有其他请求都被取消，以避免数据不完整。

我们到了文章的最后，我们已经了解了 asyncio 提供的多种创建和管理任务的方法。方法总结如下：

• asyncio.wait_for：等待超时的任务。

• asyncio.wait：等待多个任务，完成条件灵活。

• asyncio.gather：将多个任务聚合到一个等待中。

• asyncio.as_completed：处理完成的任务。

• asyncio.taskgroup：管理一组任务，并在失败时自动取消。

结论

异步编程可以改变您在 python 中处理并发任务的方式，使您的代码更加高效且响应迅速。在本文中，我们浏览了 asyncio 提供的各种方法来创建和管理任务，从简单的超时到复杂的任务组。了解何时以及如何使用每种方法（asyncio.wait_for、asyncio.wait、asyncio.gather、asyncio.as_completed 和 asyncio.taskgroup）将帮助您充分利用异步编程的潜力，使您的应用程序更加健壮和可扩展。

要更深入地了解异步编程和更多实际示例，请在此处浏览我们的详细指南。

如果您喜欢这篇文章，请考虑订阅我的时事通讯，这样您就不会错过未来的更新。

快乐编码！