Python 비동기 메서드를 사용하는 방법

비동기 프로그래밍이 필요한 이유는 무엇인가요?

비동기 프로그래밍의 동기를 이해하려면 먼저 코드 실행 속도를 제한하는 요소가 무엇인지 이해해야 합니다. 이상적으로는 코드가 빛의 속도로 실행되어 지연 없이 즉시 코드를 건너뛰기를 원합니다. 그러나 코드는 다음 두 가지 요인으로 인해 실제로 훨씬 느리게 실행됩니다.

• CPU 시간(프로세서가 명령을 실행하는 데 소비하는 시간)

• IO 시간(네트워크 요청 또는 저장소 읽기/쓰기를 기다리는 시간)

언제 우리 코드는 IO를 기다리고 있고 CPU는 기본적으로 유휴 상태이며 외부 장치의 응답을 기다리고 있습니다. 일반적으로 커널은 이를 감지하고 즉시 시스템의 다른 스레드로 실행을 전환합니다. 따라서 IO 집약적인 작업 집합의 속도를 높이려면 각 작업에 대한 스레드를 만들 수 있습니다. 스레드 중 하나가 중지되어 IO를 기다리면 커널은 처리를 계속하기 위해 다른 스레드로 전환합니다.

실제로는 잘 작동하지만 두 가지 단점이 있습니다.

• 스레드에는 오버헤드가 있습니다(특히 Python의 경우).

• 커널이 스레드 간 전환을 선택하는 시점을 제어할 수 없습니다.

예를 들어, 10,000개의 작업을 실행하려면 많은 RAM을 소비하는 10,000개의 스레드를 생성해야 하거나, 더 적은 수의 작업자 스레드를 생성하고 더 적은 동시성으로 작업을 실행해야 합니다. 또한 처음에 이러한 스레드를 생성하면 CPU 시간이 소모됩니다.

커널은 언제든지 스레드 간 전환을 선택할 수 있으므로 코드에서 언제든지 경합이 발생할 수 있습니다.

비동기성 소개

기존 동기식 스레드 기반 코드에서 커널은 스레드가 IO 바인딩되는 시기를 감지하고 마음대로 스레드 간에 전환하도록 선택해야 합니다. Python 비동기를 사용하면 프로그래머는 await 키워드를 사용하여 IO 바인딩을 선언하는 코드 줄을 확인하고 다른 작업을 수행할 수 있는 권한이 부여되었는지 확인합니다. 예를 들어, 웹 요청을 수행하는 다음 코드를 생각해 보세요. await确认声明 IO 绑定的代码行，并确认授予执行其他任务的权限。例如，考虑以下执行Web请求的代码：

async def request_google():
    reader, writer = await asyncio.open_connection('google.com', 80)
    writer.write(b'GET / HTTP/2\n\n')
    await writer.drain()
    response = await reader.read()
    return response.decode()

在这里，在这里，我们看到该代码在两个地方await。因此，在等待我们的字节被发送到服务器（writer.drain()）时，在等待服务器用一些字节（reader.read()）回复时，我们知道其他代码可能会执行，全局变量可能会更改。然而，从函数开始到第一次等待，我们可以确保代码逐行运行，而不会切换到运行程序中的其他代码。这就是异步的美妙之处。

asyncio是一个标准库，可以让我们用这些异步函数做一些有趣的事情。例如，如果我们想同时向Google执行两个请求，我们可以：

async def request_google_twice():
    response_1, response_2 = await asyncio.gather(request_google(), request_google())
    return response_1, response_2

当我们调用request_google_twice()时，神奇的asyncio.gather会启动一个函数调用，但是当我们调用时await writer.drain()，它会开始执行第二个函数调用，这样两个请求就会并行发生。然后，它等待第一个或第二个请求的writer.drain()调用完成并继续执行该函数。

最后，有一个重要的细节被遗漏了：asyncio.run。要从常规的 [同步] Python 函数实际调用异步函数，我们将调用包装在asyncio.run(...)：

async def async_main():
    r1, r2 = await request_google_twice()
    print('Response one:', r1)
    print('Response two:', r2)
    return 12

return_val = asyncio.run(async_main())

请注意，如果我们只调用async_main()而不调用await ...或者 asyncio.run(...)，则不会发生任何事情。这只是由异步工作方式的性质所限制的。

那么，异步究竟是如何工作的，这些神奇的asyncio.run和asyncio.gather函数有什么作用呢？阅读下文以了解详情。

异步是如何工作的

要了解async的魔力，我们首先需要了解一个更简单的 Python 构造：生成器

生成器

生成器是 Python 函数，它逐个返回一系列值（可迭代）。例如：

def get_numbers():
    print("|| get_numbers begin")
    print("|| get_numbers Giving 1...")
    yield 1
    print("|| get_numbers Giving 2...")
    yield 2
    print("|| get_numbers Giving 3...")
    yield 3
    print("|| get_numbers end")

print("| for begin")
for number in get_numbers():
    print(f"| Got {number}.")
print("| for end")

| for begin
|| get_numbers begin
|| get_numbers Giving 1...
| Got 1.
|| get_numbers Giving 2...
| Got 2.
|| get_numbers Giving 3...
| Got 3.
|| get_numbers end
| for end

因此，我们看到，对于for循环的每个迭代，我们在生成器中只执行一次。我们可以使用Python的next()

In [3]: generator = get_numbers()                                                                                                                                                            

In [4]: next(generator)                                                                                                                                                                      
|| get_numbers begin
|| get_numbers Giving 1...
Out[4]: 1

In [5]: next(generator)                                                                                                                                                                      
|| get_numbers Giving 2...
Out[5]: 2

In [6]: next(generator)                                                                                                                                                                      
|| get_numbers Giving 3...
Out[6]: 3

In [7]: next(generator)                                                                                                                                                                      
|| get_numbers end
---------------------------------------
StopIteration       Traceback (most recent call last)
<ipython-input-154-323ce5d717bb> in <module>
----> 1 next(generator)

StopIteration:

여기서 두 곳에서 await 코드를 볼 수 있습니다. 따라서 우리의 바이트가 서버로 전송되기를 기다리는 동안(writer.drain()), 서버가 일부 바이트로 응답하기를 기다리는 동안(reader.read() ), 우리는 다른 코드가 실행될 수 있고 전역 변수가 변경될 수 있다는 것을 알고 있습니다. 그러나 함수 시작부터 첫 번째 대기까지 실행 중인 프로그램에서 다른 코드로 전환하지 않고도 코드가 한 줄씩 실행되도록 할 수 있습니다. 이것이 비동기의 아름다움입니다.

asyncio는 이러한 비동기 함수로 몇 가지 흥미로운 작업을 수행할 수 있게 해주는 표준 라이브러리입니다. 예를 들어 Google에 대한 두 가지 요청을 동시에 수행하려는 경우 다음을 수행할 수 있습니다. 🎜

def send_updates(count: int, interval_seconds: float):
    for i in range(1, count + 1):
        time.sleep(interval_seconds)
        print('[{}] Sending update {}/{}.'.format(interval_seconds, i, count))

request_google_twice()를 호출하면 마법이 발생합니다. asyncio.gather는 함수 호출을 시작하지만 awaitwriter.drain()을 호출하면 두 번째 함수 호출 실행을 시작하므로 두 요청은 평행한 . 그런 다음 요청된 첫 번째 또는 두 번째 writer.drain() 호출이 완료될 때까지 기다렸다가 함수 실행을 계속합니다. 🎜

마지막으로 중요한 세부정보가 빠져 있습니다: asyncio.run. 실제로 일반 [동기] Python 함수에서 비동기 함수를 호출하려면 asyncio.run(...)에 호출을 래핑합니다. 🎜

[1.0] Sending update 1/3.
[1.0] Sending update 2/3.
[1.0] Sending update 3/3.

await ... 또는 asyncio.run(...)을 호출하지 않고 async_main()만 호출하면 무슨 일이든 일어납니다. 이는 비동기식 작동 방식의 특성에 따라 제한됩니다. 🎜

그렇다면 비동기식 작업이 정확히 어떻게 작동하며 이러한 마법 같은 asyncio.run 및 asyncio.gather의 기능은 무엇입니까? 기능 모직물? 자세한 내용은 아래를 읽어보세요. 🎜

비동기 작동 방식 🎜

비동기의 마법을 이해하려면 먼저 더 간단한 Python 구성을 이해해야 합니다. Generator 🎜

Generator 🎜

Generator는 다음을 반환하는 Python 함수입니다. 하나씩 값의 시퀀스(반복 가능). 예: 🎜

threads = [
    threading.Thread(target=send_updates, args=(10, 1.0)),
    threading.Thread(target=send_updates, args=(5, 2.0)),
    threading.Thread(target=send_updates, args=(4, 3.0))
]
for i in threads:
    i.start()
for i in threads:
    i.join()

class AsyncTimer:
    def __init__(self, duration: float):
        self.done_time = time.time() + duration

따라서 for 루프의 각 반복에 대해 생성기에서 한 번만 실행된다는 것을 알 수 있습니다. Python의 next() 함수를 사용하여 이 반복을 보다 명시적으로 수행할 수 있습니다. 🎜

In [3]: generator = get_numbers()                                                                                                                                                            

In [4]: next(generator)                                                                                                                                                                      
|| get_numbers begin
|| get_numbers Giving 1...
Out[4]: 1

In [5]: next(generator)                                                                                                                                                                      
|| get_numbers Giving 2...
Out[5]: 2

In [6]: next(generator)                                                                                                                                                                      
|| get_numbers Giving 3...
Out[6]: 3

In [7]: next(generator)                                                                                                                                                                      
|| get_numbers end
---------------------------------------
StopIteration       Traceback (most recent call last)
<ipython-input-154-323ce5d717bb> in <module>
----> 1 next(generator)

StopIteration:

这与异步函数的行为非常相似。正如异步函数从函数开始直到第一次等待时连续执行代码一样，我们第一次调用next()时，生成器将从函数顶部执行到第一个yield 语句。然而，现在我们只是从生成器返回数字。我们将使用相同的思想，但返回一些不同的东西来使用生成器创建类似异步的函数。

使用生成器进行异步

让我们使用生成器来创建我们自己的小型异步框架。

但是，为简单起见，让我们将实际 IO 替换为睡眠（即。time.sleep）。让我们考虑一个需要定期发送更新的应用程序：

def send_updates(count: int, interval_seconds: float):
    for i in range(1, count + 1):
        time.sleep(interval_seconds)
        print('[{}] Sending update {}/{}.'.format(interval_seconds, i, count))

因此，如果我们调用send_updates(3, 1.0)，它将输出这三条消息，每条消息间隔 1 秒：

[1.0] Sending update 1/3.
[1.0] Sending update 2/3.
[1.0] Sending update 3/3.

现在，假设我们要同时运行几个不同的时间间隔。例如，send_updates(10, 1.0)，send_updates(5, 2.0)和send_updates(4, 3.0)。我们可以使用线程来做到这一点，如下所示：

threads = [
    threading.Thread(target=send_updates, args=(10, 1.0)),
    threading.Thread(target=send_updates, args=(5, 2.0)),
    threading.Thread(target=send_updates, args=(4, 3.0))
]
for i in threads:
    i.start()
for i in threads:
    i.join()

这可行，在大约 12 秒内完成，但使用具有前面提到的缺点的线程。让我们使用生成器构建相同的东西。

在演示生成器的示例中，我们返回了整数。为了获得类似异步的行为，而不是返回任意值，我们希望返回一些描述要等待的IO的对象。在我们的例子中，我们的“IO”只是一个计时器，它将等待一段时间。因此，让我们创建一个计时器对象，用于此目的：

class AsyncTimer:
    def __init__(self, duration: float):
        self.done_time = time.time() + duration

现在，让我们从我们的函数中产生这个而不是调用time.sleep：

def send_updates(count: int, interval_seconds: float):
    for i in range(1, count + 1):
        yield AsyncTimer(interval_seconds)
        print('[{}] Sending update {}/{}.'.format(interval_seconds, i, count))

现在，每次我们调用send_updates(...)时调用next(...)，我们都会得到一个AsyncTimer对象，告诉我们直到我们应该等待什么时候：

generator = send_updates(3, 1.5)
timer = next(generator)  # [1.5] Sending update 1/3.
print(timer.done_time - time.time())  # 1.498...

由于我们的代码现在实际上并没有调用time.sleep，我们现在可以同时执行另一个send_updates调用。

所以，为了把这一切放在一起，我们需要退后一步，意识到一些事情：

• 生成器就像部分执行的函数，等待一些 IO（计时器）。

• 每个部分执行的函数都有一些 IO（计时器），它在继续执行之前等待。

• 因此，我们程序的当前状态是每个部分执行的函数（生成器）和该函数正在等待的 IO（计时器）对的对列表

• 现在，要运行我们的程序，我们只需要等到某个 IO 准备就绪（即我们的一个计时器已过期），然后再向前一步执行相应的函数，得到一个阻塞该函数的新 IO。

实现此逻辑为我们提供了以下信息：

# Initialize each generator with a timer of 0 so it immediately executes
generator_timer_pairs = [
    (send_updates(10, 1.0), AsyncTimer(0)),
    (send_updates(5, 2.0), AsyncTimer(0)),
    (send_updates(4, 3.0), AsyncTimer(0))
]

while generator_timer_pairs:
    pair = min(generator_timer_pairs, key=lambda x: x[1].done_time)
    generator, min_timer = pair

    # Wait until this timer is ready
    time.sleep(max(0, min_timer.done_time - time.time()))
    del generator_timer_pairs[generator_timer_pairs.index(pair)]

    try:  # Execute one more step of this function
        new_timer = next(generator)
        generator_timer_pairs.append((generator, new_timer))
    except StopIteration:  # When the function is complete
        pass

有了这个，我们有了一个使用生成器的类似异步函数的工作示例。请注意，当生成器完成时，它会引发StopIteration，并且当我们不再有部分执行的函数（生成器）时，我们的函数就完成了

现在，我们把它包装在一个函数中，我们得到了类似于asyncio.run的东西。结合asyncio.gather运行：

def async_run_all(*generators):
    generator_timer_pairs = [
        (generator, AsyncTimer(0))
        for generator in generators
    ]
    while generator_timer_pairs:
        pair = min(generator_timer_pairs, key=lambda x: x[1].done_time)
        generator, min_timer = pair

        time.sleep(max(0, min_timer.done_time - time.time()))
        del generator_timer_pairs[generator_timer_pairs.index(pair)]

        try:
            new_timer = next(generator)
            generator_timer_pairs.append((generator, new_timer))
        except StopIteration:
            pass

async_run_all(
    send_updates(10, 1.0),
    send_updates(5, 2.0),
    send_updates(4, 3.0)
)

使用 async/await 进行异步

实现我们的caveman版本的asyncio的最后一步是支持Python 3.5中引入的async/await语法。await的行为类似于yield，只是它不是直接返回提供的值，而是返回next((...).__await__())。async函数返回“协程”，其行为类似于生成器，但需要使用.send(None)而不是next()（请注意，正如生成器在最初调用时不返回任何内容一样，异步函数在逐步执行之前不会执行任何操作，这解释了我们前面提到的）。

因此，鉴于这些信息，我们只需进行一些调整即可将我们的示例转换为async/await。以下是最终结果：

class AsyncTimer:
    def __init__(self, duration: float):
        self.done_time = time.time() + duration
    def __await__(self):
        yield self

async def send_updates(count: int, interval_seconds: float):
    for i in range(1, count + 1):
        await AsyncTimer(interval_seconds)
        print('[{}] Sending update {}/{}.'.format(interval_seconds, i, count))

def _wait_until_io_ready(ios):
    min_timer = min(ios, key=lambda x: x.done_time)
    time.sleep(max(0, min_timer.done_time - time.time()))
    return ios.index(min_timer)

def async_run_all(*coroutines):
    coroutine_io_pairs = [
        (coroutine, AsyncTimer(0))
        for coroutine in coroutines
    ]
    while coroutine_io_pairs:
        ios = [io for cor, io in coroutine_io_pairs]
        ready_index = _wait_until_io_ready(ios)
        coroutine, _ = coroutine_io_pairs.pop(ready_index)

        try:
            new_io = coroutine.send(None)
            coroutine_io_pairs.append((coroutine, new_io))
        except StopIteration:
            pass

async_run_all(
    send_updates(10, 1.0),
    send_updates(5, 2.0),
    send_updates(4, 3.0)
)

我们有了它，我们的迷你异步示例完成了，使用async/await. 现在，您可能已经注意到我将 timer 重命名为 io 并将查找最小计时器的逻辑提取到一个名为_wait_until_io_ready. 这是有意将这个示例与最后一个主题联系起来：真实 IO。

여기서 async/await。现在，你可能已经注意到我将timer重命名为io，并将用于查找最小计时器的逻辑提取到一个名为_wait_until_io_ready的函数中。这是为了将本示例与最后一个主题：真正的IO，连接起来。

真正的 IO（而不仅仅是定时器）

所以，所有这些例子都很棒，但是它们与真正的 asyncio 有什么关系，我们希望在真正 IO 上等待 TCP 套接字和文件读/写？嗯，美丽就在那个_wait_until_io_ready功能中。为了让真正的 IO 正常工作，我们所要做的就是创建一些AsyncReadFile类似于AsyncTimer包含文件描述符的新对象。然后，AsyncReadFile我们正在等待的对象集对应于一组文件描述符。最后，我们可以使用函数 (syscall) select()等待这些文件描述符之一准备好。由于 TCP/UDP 套接字是使用文件描述符实现的，因此这也涵盖了网络请求。

所以，所有这些例子都很好，但它们与真正的异步IO有什么关系呢？我们希望等待实际的IO，比如TCP套接字和文件读/写？好吧，其优点在于_wait_until_io_ready函数。要使真正的IO工作，我们需要做的就是创建一些新的AsyncReadFile，类似于AsyncTimer，它包含一个文件描述符。然后，我们正在等待的一组AsyncReadFile对象对应于一组文件描述符。最后，我们可以使用函数（syscall）select()이러한 파일 설명자 중 하나가 준비될 때까지 기다리기를 사용하여 작은 비동기 예제를 완성했습니다. TCP/UDP 소켓은 파일 설명자를 사용하여 구현되므로 네트워크 요청도 포함됩니다.

위 내용은 Python 비동기 메서드를 사용하는 방법의 상세 내용입니다. 자세한 내용은 PHP 중국어 웹사이트의 기타 관련 기사를 참조하세요!