使用 Asyncio 创建和管理任务

admin 阅读:94 2024-08-16

使用 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)将帮助您充分利用异步编程的潜力,使您的应用程序更加健壮和可扩展。

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

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

快乐编码!

声明

1、部分文章来源于网络,仅作为参考。
2、如果网站中图片和文字侵犯了您的版权,请联系1943759704@qq.com处理!