Python 및 코루틴 마스터하기: 코드 효율성 및 성능 향상-파이썬 튜토리얼-php.cn

Mastering Python

Python에서 코루틴과 구조화된 동시성의 흥미로운 세계를 탐험해 보겠습니다. 이러한 강력한 기능은 동시 코드 작성 방식을 혁신하여 보다 효율적이고 관리하기 쉽게 만들었습니다.

코루틴은 실행을 일시 중지하고 다른 코루틴에 제어권을 양보할 수 있는 특수 함수입니다. async 키워드를 사용하여 정의되며 wait 키워드를 사용하여 기다릴 수 있습니다. 간단한 예는 다음과 같습니다.

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

async def main():
    await greet("Alice")
    await greet("Bob")

asyncio.run(main())

로그인 후 복사

이 코드에서 Greeting 함수는 인사말을 인쇄하고 잠시 기다린 후 작별 인사를 하는 코루틴입니다. 메인 함수는 Greeting을 두 번 호출하고, asyncio.run을 사용하여 메인 코루틴을 실행합니다.

그런데 코루틴이 그렇게 특별한 이유는 무엇일까요? 이를 통해 동기 코드처럼 보이고 동작하지만 실제로는 여러 작업을 동시에 수행할 수 있는 동시 코드를 작성할 수 있습니다. 이는 네트워크 작업이나 파일 처리와 같은 I/O 중심 작업에 특히 유용합니다.

Python에서 비동기 프로그래밍의 기초를 제공하는 asyncio 라이브러리에 대해 더 자세히 살펴보겠습니다. 그 핵심에는 코루틴 실행을 관리하는 이벤트 루프가 있습니다. 다음에 어떤 코루틴을 실행할지 결정하는 스케줄러라고 생각하시면 됩니다.

asyncio로 작업을 생성하고 사용하는 방법은 다음과 같습니다.

import asyncio

async def fetch_data(url):
    print(f"Fetching data from {url}")
    await asyncio.sleep(2)  # Simulating network delay
    return f"Data from {url}"

async def main():
    urls = ['http://example.com', 'http://example.org', 'http://example.net']
    tasks = [asyncio.create_task(fetch_data(url)) for url in urls]
    results = await asyncio.gather(*tasks)
    for result in results:
        print(result)

asyncio.run(main())

로그인 후 복사

이 예에서는 여러 URL에서 동시에 데이터를 가져오는 것을 시뮬레이션합니다. asyncio.create_task 함수는 코루틴을 작업으로 변환한 다음 asyncio.gather를 사용하여 동시에 실행됩니다.

이제 구조적 동시성에 대해 이야기해 보겠습니다. 이는 동시 코드를 보다 예측 가능하고 추론하기 쉽게 만드는 것을 목표로 하는 패러다임입니다. Python 3.11에는 작업 그룹과 같은 구조화된 동시성을 지원하는 몇 가지 새로운 기능이 도입되었습니다.

작업 그룹을 사용하는 방법은 다음과 같습니다.

import asyncio

async def process_item(item):
    await asyncio.sleep(1)
    return f"Processed {item}"

async def main():
    async with asyncio.TaskGroup() as tg:
        task1 = tg.create_task(process_item("A"))
        task2 = tg.create_task(process_item("B"))
        task3 = tg.create_task(process_item("C"))

    print(task1.result())
    print(task2.result())
    print(task3.result())

asyncio.run(main())

로그인 후 복사

TaskGroup은 계속 진행하기 전에 모든 작업이 완료(또는 취소)되었는지 확인합니다. 이는 잊어버린 작업이나 동시 작업 간의 예상치 못한 상호 작용과 같은 문제를 방지하는 데 도움이 됩니다.

코루틴의 가장 강력한 측면 중 하나는 I/O 작업을 효율적으로 처리하는 능력입니다. 간단한 비동기 웹 서버의 예를 살펴보겠습니다.

import asyncio
from aiohttp import web

async def handle(request):
    name = request.match_info.get('name', "Anonymous")
    text = f"Hello, {name}!"
    return web.Response(text=text)

async def main():
    app = web.Application()
    app.add_routes([web.get('/', handle),
                    web.get('/{name}', handle)])

    runner = web.AppRunner(app)
    await runner.setup()
    site = web.TCPSite(runner, 'localhost', 8080)
    await site.start()

    print("Server started at http://localhost:8080")
    await asyncio.Event().wait()

asyncio.run(main())

로그인 후 복사

이 서버는 코루틴의 강력한 기능 덕분에 여러 연결을 동시에 처리할 수 있습니다. 각 요청은 자체 코루틴에서 처리되므로 부하가 높은 경우에도 서버가 계속 응답할 수 있습니다.

좀 더 고급 개념을 살펴보겠습니다. 취소는 동시 작업을 처리할 때 중요한 기능입니다. 때로는 작업이 완료되기 전에 작업을 중지해야 할 때도 있습니다. 이를 수행하는 방법은 다음과 같습니다.

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

async def main():
    await greet("Alice")
    await greet("Bob")

asyncio.run(main())

로그인 후 복사

이 예에서는 장기 실행 작업을 생성하고 5초 후에 취소합니다. 작업은 CancelledError를 포착하고 종료하기 전에 필요한 정리를 수행합니다.

또 다른 강력한 기능은 맞춤 이벤트 루프를 생성하는 기능입니다. 대부분의 경우 기본 이벤트 루프로 충분하지만 때로는 더 많은 제어가 필요합니다. 다음은 맞춤 이벤트 루프의 간단한 예입니다.

import asyncio

async def fetch_data(url):
    print(f"Fetching data from {url}")
    await asyncio.sleep(2)  # Simulating network delay
    return f"Data from {url}"

async def main():
    urls = ['http://example.com', 'http://example.org', 'http://example.net']
    tasks = [asyncio.create_task(fetch_data(url)) for url in urls]
    results = await asyncio.gather(*tasks)
    for result in results:
        print(result)

asyncio.run(main())

로그인 후 복사

이것은 매우 기본적인 맞춤 이벤트 루프이지만 원리를 보여줍니다. 이를 확장하여 더 나은 예약, 모니터링 또는 다른 시스템과의 통합과 같은 기능을 추가할 수 있습니다.

코루틴과 구조적 동시성을 사용할 때의 몇 가지 모범 사례에 대해 이야기해 보겠습니다. 첫째, 비동기 컨텍스트 관리자를 관리하려면 항상 async with를 사용하세요. 이렇게 하면 예외가 발생하더라도 적절한 설정 및 해제가 보장됩니다.

import asyncio

async def process_item(item):
    await asyncio.sleep(1)
    return f"Processed {item}"

async def main():
    async with asyncio.TaskGroup() as tg:
        task1 = tg.create_task(process_item("A"))
        task2 = tg.create_task(process_item("B"))
        task3 = tg.create_task(process_item("C"))

    print(task1.result())
    print(task2.result())
    print(task3.result())

asyncio.run(main())

로그인 후 복사

둘째, 차단 작업에 주의하세요. CPU 바인딩된 작업을 수행해야 하는 경우 asyncio.to_thread를 사용하여 별도의 스레드에서 실행하는 것을 고려해 보세요.

import asyncio
from aiohttp import web

async def handle(request):
    name = request.match_info.get('name', "Anonymous")
    text = f"Hello, {name}!"
    return web.Response(text=text)

async def main():
    app = web.Application()
    app.add_routes([web.get('/', handle),
                    web.get('/{name}', handle)])

    runner = web.AppRunner(app)
    await runner.setup()
    site = web.TCPSite(runner, 'localhost', 8080)
    await site.start()

    print("Server started at http://localhost:8080")
    await asyncio.Event().wait()

asyncio.run(main())

로그인 후 복사

셋째, 작업 그룹에 대해 더 많은 제어가 필요할 때 asyncio.wait를 사용하세요. 첫 번째 작업이 완료될 때까지 기다리거나 시간 초과를 설정할 수 있습니다.

import asyncio

async def long_running_task():
    try:
        while True:
            print("Working...")
            await asyncio.sleep(1)
    except asyncio.CancelledError:
        print("Task was cancelled")

async def main():
    task = asyncio.create_task(long_running_task())
    await asyncio.sleep(5)
    task.cancel()
    try:
        await task
    except asyncio.CancelledError:
        print("Main: task was cancelled")

asyncio.run(main())

로그인 후 복사

동시 코드를 디버깅하는 것은 어려울 수 있습니다. Python의 asyncio에는 몇 가지 유용한 도구가 함께 제공됩니다. 디버그 모드를 활성화하면 더 자세한 로깅을 얻을 수 있습니다.

import asyncio

class MyEventLoop(asyncio.BaseEventLoop):
    def __init__(self):
        self._running = False
        self._ready = asyncio.Queue()

    def run_forever(self):
        self._running = True
        while self._running:
            coro = self._ready.get_nowait()
            if coro:
                coro.send(None)

    def stop(self):
        self._running = False

    def call_soon(self, callback, *args):
        self._ready.put_nowait(callback(*args))

# Usage
loop = MyEventLoop()
asyncio.set_event_loop(loop)

async def my_coroutine():
    print("Hello from my coroutine!")

loop.call_soon(my_coroutine)
loop.run_forever()

로그인 후 복사

또한 고급 디버깅 기능을 위해 aiodebug 라이브러리를 사용할 수도 있습니다.

더 복잡한 예인 병렬 데이터 처리 파이프라인을 살펴보겠습니다. 이는 대규모 데이터 세트 처리 또는 스트리밍 데이터 처리와 같은 작업에 유용할 수 있습니다.

async with aiohttp.ClientSession() as session:
    async with session.get('http://example.com') as response:
        html = await response.text()

로그인 후 복사

이 파이프라인은 대기열을 사용하여 동시에 실행되는 다양한 처리 단계 간에 데이터를 전달하는 방법을 보여줍니다.

코루틴과 구조적 동시성은 Python 프로그래밍에 새로운 가능성을 열어주었습니다. 이를 통해 우리는 더 쉽게 추론하고 유지 관리할 수 있는 효율적인 동시 코드를 작성할 수 있습니다. 웹 서버, 데이터 처리 파이프라인, 반응형 GUI 등 무엇을 구축하든 이러한 도구는 강력한 고성능 애플리케이션을 만드는 데 도움이 될 수 있습니다.

이러한 개념을 익히는 열쇠는 연습이라는 점을 기억하세요. 간단한 예제로 시작하여 점차적으로 더 복잡한 사용 사례로 확장해 보세요. 오류 처리 및 취소에 주의하세요. 이는 안정적인 비동기 시스템을 구축하는 데 중요합니다. 그리고 두려워하지 말고 asyncio 소스 코드를 살펴보세요. 이는 이러한 강력한 기능이 내부적으로 어떻게 작동하는지 더 깊이 이해할 수 있는 좋은 방법입니다.

코루틴과 구조화된 동시성을 계속 탐색하면서 코드를 더욱 효율적이고 표현력 있게 만들 수 있는 새로운 패턴과 기술을 발견하게 될 것입니다. 이는 Python 개발의 흥미로운 영역이자 지속적으로 발전하고 있는 영역입니다. 그러니 계속 배우고, 실험하고, 비동기 프로그래밍의 세계로의 여정을 즐겨보세요!