使用 Asyncio 创建和管理任务

百变鹏仔 3天前 #Python

文章标签 Asyncio

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

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

什么是异步任务？

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

创建和管理 asyncio 任务的方法

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

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

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

import asyncio# define a coroutineasync 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 functionasyncio.run(main())

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

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

等待任务完成

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

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

这是使用此方法的示例：

import asyncioasync 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 方法等待一组任务并返回两组：一组用于已完成的任务，一组用于待处理的任务。它需要以下参数：

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

import asyncioimport randomasync 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 方法同时运行多个可等待对象并返回其结果列表，可以选择处理异常。让我们看看它所需要的论据。

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

import asyncioimport randomasync 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 方法用于返回一个迭代器，该迭代器在任务完成时生成任务，从而允许立即处理结果。它需要以下参数：

例子

import asyncioimport randomasync 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.taskgroup 以及如何使用它来管理一组任务。

异步任务组

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

这是如何使用它的示例：

import asyncioasync 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 提供的多种创建和管理任务的方法。方法总结如下：

结论

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

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

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

快乐编码！

文章推荐

使用 Asyncio 创建和管理任务

什么是异步任务？

创建和管理 asyncio 任务的方法

等待任务完成

异步等待

异步收集

asyncio.as_completed 已完成

例子

异步任务组

结论

Python中使用Beautiful Soup进行网页抓取：基础知识探究

使用Python计算神经机器翻译的BLEU分数

探索 Python 包的使用

使用Python获取文件中的字符数、单词数、空格数和行数

封装在Python中是什么？