Python 및 비동기화 마스터하기: 코루틴 및 이벤트 루프를 사용하여 앱 성능 향상-파이썬 튜토리얼-php.cn

Mastering Python

Python의 비동기 프로그래밍은 고성능 애플리케이션 구축에 획기적인 변화를 가져왔습니다. 저는 수년 동안 이 제품을 사용해 왔으며 올바르게 사용하면 얼마나 강력한지 계속해서 놀라게 됩니다.

Python 비동기 모델의 핵심은 코루틴과 이벤트 루프입니다. 코루틴은 실행을 일시 중지하고 재개할 수 있는 특수 함수로, 스레드의 오버헤드 없이 효율적인 멀티태스킹을 가능하게 합니다. 반면 이벤트 루프는 이러한 코루틴을 구동하고 실행을 관리하며 I/O 작업을 처리하는 엔진입니다.

코루틴부터 시작해 보겠습니다. Python에서는 async def 구문을 사용하여 이를 정의합니다. 간단한 예는 다음과 같습니다.

async def greet(name):
    print(f"Hello, {name}!")
    await asyncio.sleep(1)
    print(f"Goodbye, {name}!")

로그인 후 복사

이 코루틴은 사람에게 인사하고 잠시 기다린 후 작별 인사를 합니다. 여기서 wait 키워드는 매우 중요합니다. 이를 통해 코루틴이 실행을 일시 중지하고 이벤트 루프에 다시 제어권을 줄 수 있습니다.

그런데 코루틴은 내부적으로 어떻게 작동하나요? 실제로 Python의 생성기 기능을 기반으로 구축되었습니다. 코루틴을 호출하면 즉시 실행되지 않습니다. 대신 코루틴 객체를 반환합니다. 이 객체는 생성기처럼 값을 전송하고 값을 산출할 수 있습니다.

이벤트 루프는 이러한 코루틴을 실제로 실행하는 역할을 합니다. 작업 대기열(코루틴을 둘러싼 래퍼)을 유지하고 하나씩 실행합니다. 코루틴이 Wait 문에 도달하면 이벤트 루프는 이를 일시 중지하고 다음 작업으로 이동합니다. 이것이 협력적 멀티태스킹의 핵심입니다. 작업은 자발적으로 제어권을 포기하고 다른 사람이 실행하도록 허용합니다.

이벤트 루프 작동 방식을 단순화한 버전은 다음과 같습니다.

class EventLoop:
    def __init__(self):
        self.ready = deque()
        self.sleeping = []

    def call_soon(self, callback):
        self.ready.append(callback)

    def call_later(self, delay, callback):
        deadline = time.time() + delay
        heapq.heappush(self.sleeping, (deadline, callback))

    def run_forever(self):
        while True:
            self.run_once()

    def run_once(self):
        now = time.time()
        while self.sleeping and self.sleeping[0][0] <= now:
            _, callback = heapq.heappop(self.sleeping)
            self.ready.append(callback)

        if self.ready:
            callback = self.ready.popleft()
            callback()
        else:
            time.sleep(0.1)  # Avoid busy waiting

로그인 후 복사

이 이벤트 루프는 두 가지 유형의 작업, 즉 실행 준비가 완료된 작업(준비된 deque에 있음)과 휴면 중인 작업(휴면 목록에 있음)을 관리합니다. run_forever 메소드는 더 이상 남은 작업이 없을 때까지 작업을 계속 실행합니다.

이제 작업 예약에 대해 이야기해 보겠습니다. Python의 asyncio 모듈은 고급 예약 기능을 갖춘 보다 정교한 이벤트 루프를 제공합니다. I/O 작업을 처리하고 하위 프로세스를 실행하며 다른 이벤트 루프와 통합할 수도 있습니다.

asyncio를 사용하여 여러 코루틴을 동시에 실행하는 방법은 다음과 같습니다.

import asyncio

async def task1():
    print("Task 1 starting")
    await asyncio.sleep(2)
    print("Task 1 finished")

async def task2():
    print("Task 2 starting")
    await asyncio.sleep(1)
    print("Task 2 finished")

async def main():
    await asyncio.gather(task1(), task2())

asyncio.run(main())

로그인 후 복사

이 스크립트는 두 작업을 모두 시작하지만, task2는 대기 시간이 더 짧기 때문에 task1보다 먼저 완료됩니다.

비동기 프로그래밍의 가장 강력한 응용 프로그램 중 하나는 네트워크 작업입니다. 간단한 비동기 웹 서버를 살펴보겠습니다.

import asyncio

async def handle_client(reader, writer):
    data = await reader.read(100)
    message = data.decode()
    addr = writer.get_extra_info('peername')

    print(f"Received {message!r} from {addr!r}")

    response = f"Echo: {message}\n"
    writer.write(response.encode())
    await writer.drain()

    print("Close the connection")
    writer.close()

async def main():
    server = await asyncio.start_server(
        handle_client, '127.0.0.1', 8888)

    addr = server.sockets[0].getsockname()
    print(f'Serving on {addr}')

    async with server:
        await server.serve_forever()

asyncio.run(main())

로그인 후 복사

이 서버는 스레드를 사용하지 않고 동시에 여러 클라이언트를 처리할 수 있어 매우 효율적입니다.

그러나 비동기 프로그래밍은 서버만을 위한 것이 아닙니다. 특히 여러 네트워크 요청을 해야 하는 경우 클라이언트에게도 좋습니다. 다음은 여러 페이지를 동시에 가져올 수 있는 간단한 웹 스크레이퍼입니다.

async def greet(name):
    print(f"Hello, {name}!")
    await asyncio.sleep(1)
    print(f"Goodbye, {name}!")

로그인 후 복사

이 스크레이퍼는 여러 페이지를 동시에 가져올 수 있어 동기식 접근 방식에 비해 프로세스 속도가 크게 향상됩니다.

이제 좀 더 고급 개념을 살펴보겠습니다. Python 비동기 모델의 흥미로운 기능 중 하나는 자신만의 이벤트 루프를 만들 수 있다는 것입니다. 이는 비동기 코드를 다른 프레임워크와 통합해야 하거나 특정 사용 사례에 맞게 최적화하려는 경우 유용할 수 있습니다.

다음은 동기 및 비동기 콜백을 모두 실행할 수 있는 간단한 사용자 정의 이벤트 루프입니다.

class EventLoop:
    def __init__(self):
        self.ready = deque()
        self.sleeping = []

    def call_soon(self, callback):
        self.ready.append(callback)

    def call_later(self, delay, callback):
        deadline = time.time() + delay
        heapq.heappush(self.sleeping, (deadline, callback))

    def run_forever(self):
        while True:
            self.run_once()

    def run_once(self):
        now = time.time()
        while self.sleeping and self.sleeping[0][0] <= now:
            _, callback = heapq.heappop(self.sleeping)
            self.ready.append(callback)

        if self.ready:
            callback = self.ready.popleft()
            callback()
        else:
            time.sleep(0.1)  # Avoid busy waiting

로그인 후 복사

이 사용자 정의 루프는 매우 기본적이지만 핵심 원칙을 보여줍니다. I/O 작업이나 타이머와 같은 더 복잡한 시나리오를 처리하기 위해 이를 확장할 수 있습니다.

비동기 코드 디버깅은 어려울 수 있으며, 특히 복잡한 애플리케이션을 다룰 때는 더욱 그렇습니다. 제가 찾은 유용한 기술 중 하나는 asyncio의 디버그 모드를 사용하는 것입니다. 다음과 같이 활성화할 수 있습니다:

import asyncio

async def task1():
    print("Task 1 starting")
    await asyncio.sleep(2)
    print("Task 1 finished")

async def task2():
    print("Task 2 starting")
    await asyncio.sleep(1)
    print("Task 2 finished")

async def main():
    await asyncio.gather(task1(), task2())

asyncio.run(main())

로그인 후 복사

이렇게 하면 완료되지 않는 코루틴이나 실행하는 데 너무 오래 걸리는 콜백과 같은 항목에 대한 더 자세한 오류 메시지와 경고가 제공됩니다.

또 다른 유용한 디버깅 기술은 asyncio의 작업 검사 기능을 사용하는 것입니다. 예를 들어, 실행 중인 모든 작업 목록을 얻을 수 있습니다.

import asyncio

async def handle_client(reader, writer):
    data = await reader.read(100)
    message = data.decode()
    addr = writer.get_extra_info('peername')

    print(f"Received {message!r} from {addr!r}")

    response = f"Echo: {message}\n"
    writer.write(response.encode())
    await writer.drain()

    print("Close the connection")
    writer.close()

async def main():
    server = await asyncio.start_server(
        handle_client, '127.0.0.1', 8888)

    addr = server.sockets[0].getsockname()
    print(f'Serving on {addr}')

    async with server:
        await server.serve_forever()

asyncio.run(main())

로그인 후 복사

이는 특정 순간에 프로그램이 수행하는 작업을 이해하는 데 도움이 됩니다.

비동기 코드를 최적화할 때 중요한 원칙 중 하나는 동기 작업에 소요되는 시간을 최소화하는 것입니다. 장기 실행 동기 코드는 이벤트 루프를 차단하여 다른 코루틴이 실행되지 않도록 합니다. CPU 집약적인 작업이 있는 경우 별도의 스레드나 프로세스에서 실행하는 것이 좋습니다.

또 다른 최적화 기술은 동시에 실행할 수 있는 코루틴이 여러 개 있는 경우 asyncio.gather를 사용하는 것입니다. 이는 하나씩 기다리는 것보다 더 효율적입니다.

import asyncio
import aiohttp

async def fetch_page(session, url):
    async with session.get(url) as response:
        return await response.text()

async def main():
    urls = [
        'http://example.com',
        'http://example.org',
        'http://example.net'
    ]

    async with aiohttp.ClientSession() as session:
        tasks = [fetch_page(session, url) for url in urls]
        pages = await asyncio.gather(*tasks)

    for url, page in zip(urls, pages):
        print(f"Page from {url}: {len(page)} bytes")

asyncio.run(main())

로그인 후 복사

마지막으로 비동기 프로그래밍이 항상 최선의 솔루션은 아니라는 점을 기억하세요. 대기 시간이 많은 I/O 중심 작업의 경우 성능이 크게 향상될 수 있습니다. 그러나 CPU 바인딩 작업의 경우 기존 멀티스레딩 또는 멀티프로세싱이 더 적합할 수 있습니다.

결론적으로 코루틴과 이벤트 루프를 기반으로 구축된 Python의 비동기 프로그래밍 모델은 효율적이고 확장 가능한 애플리케이션을 작성하는 강력한 방법을 제공합니다. 웹 서버, 네트워크 클라이언트 또는 데이터 처리 파이프라인을 구축하는 경우 이러한 개념을 이해하면 Python의 비동기 기능을 최대한 활용하는 데 도움이 될 수 있습니다. 다른 강력한 도구와 마찬가지로 효과적으로 사용하려면 연습과 신중한 생각이 필요하지만 결과는 정말 인상적일 수 있습니다.