使用 Asyncio 进行异步 Python 简介-Toy模板网

公平地说，大多数Hackers 和 Slackers读者都有一个共同点：我们喜欢用 Python 编写东西。这并不使我们与众不同；相反，我们是独一无二的。它反映了一个众所周知且易于解释的现象，即数据科学家/工程师进入（以及最近离开）以前为软件工程保留的空间：多用途编程语言。尽管这些学科彼此之间有多么独特，但我们有一个共同的特征。引用披头士乐队的话，我们需要的是 Python™️。

然而，每一次旅程都有一个决定性的时刻，很明显，这门语言已经度过了它的30 岁生日。自 Guido 释放 Serpent 来解决1991 年的问题以来，已经过去了三十年。这是冷战时代的最后一年：历史上的一个不同时期，在计算领域更是如此。Python 背后的大多数设计决策在当时都是明智的，但其中一些决策已成为当今的“怪癖”。最具争议性的怪癖很容易就是并发话题。

事实上，我指的是全局解释器锁（GIL）。我会让你免去贬低 GIL 的痛苦，因为其他人在这方面做得比我好得多（如果你对细节感兴趣并且有时间，我强烈推荐一篇题为The GIL 的文章，以及它对 Python 多线程的影响）。GIL 的长处和短处在于它限制了 Python 有效利用多个 CPU 核心，让您的 8 核笔记本电脑在单个核心上运行 Python 脚本，而其他核心则闲置。

并发性是一个远远超出 Python 范围的复杂问题。大多数编程语言都有相似的命运。但我们来这里并不是为了哀叹我们的处境；而是为了我们的处境。我们在这里讨论异步 I/O。

在 Python 中同时做多件事

并发是编程中的一个广泛概念，可以归结为“同时做一堆事情”。并发运行的代码通常采用两种可能的形式之一：

任务轮流执行，以尽量减少同事任务的停机时间。
真正并行的任务同时并行运行。

Python 附带了两个模块，分别处理这两种方法：

线程（单进程）：Python 的线程模块仅限于在给定时间使用单个处理器。线程模块可以管理占用给定线程的任务。任务 X 一直运行，直到被外部因素阻止（例如等待对 HTTP 请求的响应）。同时，任务 Y 优先执行它，直到任务 X 准备好继续（因此“阻塞 I/O”）。
多处理（多个处理器）：多处理模块使代码能够并行运行。脚本被初始化并在n 个CPU上同时运行n次。将单条道路扩展为 8 车道高速公路具有明显的性能优势，直到需要整合每项任务的结果为止。多个进程无法同时将数据写入同一目标（数据库、文件等）而不创建相互阻塞的锁。对于旨在产生输出的脚本来说，尝试解决这个问题几乎肯定是一项难以克服的努力。

Asyncio 适用于哪里？

Asyncio是上述方法的第三种也是通常首选的替代方法。尽管仅限于单个线程，Asyncio可以比 Python 的本机单线程执行速度快得多地执行大量操作。为了说明这是如何可能的，请考虑人类如何倾向于“同时处理多项任务”。当人们声称自己是“多任务处理”时，他们通常是通过在任务之间切换来完成工作，而不是同时做多件事。单线程程序以同样的方式异步：通过重叠工作来优化输出。

虽然人类在多任务处理方面表现不佳是出了名的，但机器可以从这种做法中看到显着的性能优势。它们通常更适合在没有额外开销的情况下启动和停止工作。这个概念是FastAPI等框架的“秘密武器” ，FastAPI 是一个异步 Python 框架，它自称性能“与NodeJS和Go相当 ” （我保证下次会在 FastAPI 上写一篇公平的文章）。

我花了一段时间才打消了我的怀疑，即单个线程如何同时处理多个任务可以提供值得一写的性能优势。直到我偶然发现事件循环的概念，事情才开始有意义。

事件循环

我们从积压的 I/O“任务”开始。它们可以是 HTTP 请求，将内容保存到磁盘，或者在我们的例子中，两者的混合。同步 Python 工作流程（读作：标准 Python）将从开始到结束一次执行一项任务。这就像在车管所排队等候，那里只有一根线，而这位女士讨厌你们所有人。

事件循环以不同的方式处理事情，以便更快地完成任务。给定许多任务，事件“循环”通过获取新任务并将它们委托给线程来工作。该循环不断检查正在进行的任务是否有停机（或完成）。当委派任务“等待”外部因素（例如 HTTP 请求）时，事件循环会通过启动线程中的另一个任务来填充死区时间。如果事件循环发现分配的任务已完成，则该任务将从其线程中删除，收集该任务的输出，并且循环从队列中选择另一个任务来占用该线程：

异步 I/O 事件循环

与 DMV 线路不同，事件循环的工作方式与餐厅有些相似（请继续我的说法）。尽管有许多桌子和一个厨房，但服务器通过在桌子（任务）和厨房（线程）之间轮换来处理“积压”。在厨房里连续准备多个食物订单比接受单个订单并等待它们被创建/提供给下一个顾客之前要高效得多。

协程：异步运行的函数

异步 Python 脚本不定义函数- 它们定义协程。协程（用定义async def，而不是def）可以在完成之前停止执行，通常等待另一个协程完成。下面的代码片段演示了协程的最简单示例：

"""Define a Coroutine function to be executed asynchronously."""
import asyncio

from logger import LOGGER


async def simple_coroutine(number: int):
    """
    Wait for a time delay & display number associated with coroutine.

    :param int number: Number to identify the current coroutine.
    """
    await asyncio.sleep(1)
    LOGGER.info(f"Coroutine {number} has finished executing.")

协程.py

simple_coroutine()在记录消息之前暂停执行 1 秒。协程不能像常规函数那样被调用；除非在 asyncio 事件循环内运行，否则尝试运行simple_coroutine(1)将不起作用。幸运的是，创建事件循环很容易：

import asyncio

from coroutines import simple_coroutine  # Import our coroutine


asyncio.run(simple_coroutine(1))

运行协程

asyncio.run()创建一个事件循环，并运行传递给它的协程。当您的脚本有一个所有逻辑源自的入口点时，创建事件循环是最好的。或者，asyncio.gather()如果您只想执行少量协程，则可以接受任意数量的协程：

import asyncio

from coroutines import simple_coroutine  # Import our coroutine


asyncio.gather(
    simple_coroutine(1)
    simple_coroutine(2)
    simple_coroutine(3)
)

在事件循环内运行 3 个协程

运行此脚本将执行所有三个协程并记录以下内容：

1
2
3

asyncio.gather()三个协程的输出

您认为完成上述操作需要多长时间？3秒，也许？或者我们是否能够通过魔法来优化我们的代码？

您可能会惊讶地发现，运行上述代码始终能在几乎一秒内执行（或者在糟糕的一天偶尔会执行1.01 秒）。如果我们使用 Python 的内置time.perf_counter()来计算函数的执行时间，我们可以直接看到这一点：

import asyncio
import time

from coroutines import simple_coroutine  # Import our coroutine


def async_gather_example()
    start_time = time.perf_counter()
    asyncio.gather(
        simple_coroutine(1)
        simple_coroutine(2)
        simple_coroutine(3)
    )
    print(
        f"Executed {__name__} in {time.perf_counter() - start_time:0.2f} seconds."
    )


async_example()

跟踪执行 3 个休眠 1 秒的协程的执行时间

果然，该脚本几乎只花了1 秒：

Executed async_example in 1.01 seconds.

输出async_example()

我们的协程simple_coroutine()需要 1 秒才能自行执行。上面令人印象深刻的是，我们调用了这个协程 3 次，运行时间接近 1 秒，而同步Python 脚本确实需要 3 秒。更重要的是，执行这些任务的开销仅不到.01几秒，这意味着我们的协程几乎同时完成。

使用任务

asyncio.gather()在上面的示例中，我们回避了 Asyncio 中的一个基本数据结构：Task.

协程是可以异步运行的函数。当以特定方式运行数百或数千个此类函数时，如果能够“管理”这些函数，那就太好了。了解协程何时失败（以及如何处理它），或者只是检查循环当前正在处理哪个协程，特别是当我们的事件循环可能需要几分钟或几小时才能执行或有可能失败时。

管理任务

在更复杂的工作流程中，任务提供了几种有用的方法来帮助我们管理正在执行的任务：

.set_name([name])（和.get_name()）：为任务命名，以便以人类可读的方式来识别哪个任务。
.cancel(msg=[message])：取消事件循环中的任务，允许循环继续执行其他任务。对于无响应或不太可能完成的任务很有用。
.canceled()：返回True任务是否被取消，否则False返回。
.done()：返回True任务是否成功完成，否则False返回。
.result()：返回任务的结果。canceled任务将包含有关任务被取消原因的异常消息，而done任务将仅返回done。尚未调用的任务将返回InvalidStateError异常。
许多其他方法都可以在Asyncio 的 Task 文档中找到。

创建任务

Coroutine使用 Asyncio包装sTask很简单。之前的运行asyncio.gather()为我们解决了这个问题，但这只是一种捷径，使我们无法利用任务的优势，因为任务被实例化为通用对象并立即执行。如果我们事先创建任务，我们可以将元数据与它们关联起来，并在准备好时在事件循环中执行它们。

我们将创建一个名为的新协程 create_tasks()，该协程将：

创建n 个Task 实例simple_coroutine()。
在创建时为每个任务分配一个名称。
以 Python 列表的形式返回所有任务，稍后可以通过事件循环执行：

"""Create multiple tasks from a Coroutine."""
import asyncio
from asyncio import Task
from typing import List

from logger import LOGGER

from asyncio_intro_part1.coroutines import simple_coroutine


async def create_tasks(num_tasks: int) -> List[Task]:
    """
    Create n number of asyncio tasks to be executed.

    :param int num_tasks: Number of tasks to create.

    :returns: List[Task]
    """
    task_list = []
    LOGGER.info(f"Creating {num_tasks} tasks to be executed...")
    for i in range(num_tasks):
        task = asyncio.create_task(
            simple_coroutine(i),
            name=f"Task #{i}"
        )
        task_list.append(task)
        LOGGER.info(f"Created Task: {task}")
    return task_list

任务.py

行动中的任务

定义了我们的create_tasks()方法后，就到了有趣的部分了：查看任务的创建、执行和完成。在项目的根部，我们将定义最后一个函数async_tasks_example()来演示这一点：

...
from .tasks import create_tasks


async def async_tasks_example():
    """Create and inspect tasks to wrap simple functions."""
    task_list = await create_tasks(5)
    done, pending = await asyncio.wait(task_list)
    if done:
        LOGGER.success(
            f"{len(done)} tasks completed: {[task.get_name() for task in done]}."
        )
    if pending:
        LOGGER.warning(
            f"{len(done)} tasks pending: {[task.get_name() for task in pending]}."
        )

__init__.py

我们首先将通过创建的 5 个任务分配create_tasks()给该task_list变量。发生这种情况时，我们会看到在tasks.py中添加的正确日志记录：

17:00:53 PM | INFO: Creating 5 tasks to be executed...

17:00:53 PM | INFO: Created Task: <Task pending name='Task #0' coro=<simple_coroutine() running at /Users/toddbirchard/Projects/asyncio-tutorial-part1/asyncio_intro_part1/coroutines.py:7>>

17:00:53 PM | INFO: Created Task: <Task pending name='Task #1' coro=<simple_coroutine() running at /Users/toddbirchard/Projects/asyncio-tutorial-part1/asyncio_intro_part1/coroutines.py:7>>

17:00:53 PM | INFO: Created Task: <Task pending name='Task #2' coro=<simple_coroutine() running at /Users/toddbirchard/Projects/asyncio-tutorial-part1/asyncio_intro_part1/coroutines.py:7>>

17:00:53 PM | INFO: Created Task: <Task pending name='Task #3' coro=<simple_coroutine() running at /Users/toddbirchard/Projects/asyncio-tutorial-part1/asyncio_intro_part1/coroutines.py:7>>

17:00:53 PM | INFO: Created Task: <Task pending name='Task #4' coro=<simple_coroutine() running at /Users/toddbirchard/Projects/asyncio-tutorial-part1/asyncio_intro_part1/coroutines.py:7>>

在tasks.py中创建5个任务的输出

我们随后通过执行这五个任务asyncio.wait(task_list)。asyncio.wait()尝试完成 task_list 中的所有任务并返回“已完成”和“待处理”任务的元组。由于添加了一些日志记录和任务名称的存在，我们可以确认所有任务均已成功完成：文章来源地址https://www.toymoban.com/diary/python/583.html

17:00:54 PM | INFO: Coroutine 0 has finished executing.
17:00:54 PM | INFO: Coroutine 1 has finished executing.
17:00:54 PM | INFO: Coroutine 2 has finished executing.
17:00:54 PM | INFO: Coroutine 3 has finished executing.
17:00:54 PM | INFO: Coroutine 4 has finished executing.
17:00:54 PM | SUCCESS: 5 tasks completed: ['Task #1', 'Task #4', 'Task #3', 'Task #0', 'Task #2'].

到此这篇关于使用 Asyncio 进行异步 Python 简介的文章就介绍到这了,更多相关内容可以在右上角搜索或继续浏览下面的相关文章，希望大家以后多多支持TOY模板网！