[Python] 비동기(async) 프로그래밍 5

comgu 2025. 5. 23. 15:46

2025. 5. 23. 15:46

들어가며

4편에서 여러 비동기 작업을 동시에 처리하는 방법을 학습했다. 이번 편에서는 파이썬 비동기 프로그래밍의 고급 문법들을 다룬다. async with, async for, 비동기 제너레이터 등 실무에서 자주 사용하는 비동기 문법들을 익혀보자.

async with: 비동기 컨텍스트 매니저

기본 개념

일반적인 with 문은 파일 열기/닫기나 락 획득/해제 같은 자원 관리를 동기적으로 처리한다. 하지만 데이터베이스 연결이나 HTTP 클라이언트 같은 비동기 자원들은 생성과 정리 과정에서도 await가 필요하다.

async with는 이런 비동기 자원의 생성과 정리를 자동으로 처리해준다. 객체가 __aenter__()와 __aexit__() 메서드를 구현하면 비동기 컨텍스트 매니저가 된다. 진입 시 __aenter__()가, 종료 시 __aexit__()가 자동으로 호출되어 예외 발생 여부와 관계없이 안전한 자원 정리가 보장된다.

import asyncio
import time

class TimeMeasurer:
    def __init__(self, name: str):
        self.name = name
    
    async def __aenter__(self):
        print(f"{self.name} 시작")
        self.start_time = time.time()
        return self
    
    async def __aexit__(self, exc_type, exc_val, exc_tb):
        duration = time.time() - self.start_time
        print(f"{self.name} 완료 ({duration:.2f}초)")

async def context_example():
    async with TimeMeasurer("API 호출"):
        await asyncio.sleep(1)  # 실제 작업
        print("작업 수행 중...")

asyncio.run(context_example())

실용적 활용

실무에서는 연결 풀 관리, 세마포어 제어, 트랜잭션 처리 등에서 비동기 컨텍스트 매니저가 자주 사용된다. 특히 동시 접속 수를 제한하거나 자원의 사용량을 제어할 때 유용하다. 아래 예제는 최대 연결 수가 제한된 자원 풀을 구현한 것이다.

class ResourcePool:
    def __init__(self, max_connections: int = 3):
        self.max_connections = max_connections
        self.active = 0
    
    async def __aenter__(self):
        while self.active >= self.max_connections:
            await asyncio.sleep(0.1)  # 연결 대기
        
        self.active += 1
        print(f"연결 획득 ({self.active}/{self.max_connections})")
        return self
    
    async def __aexit__(self, exc_type, exc_val, exc_tb):
        self.active -= 1
        print(f"연결 해제 ({self.active}/{self.max_connections})")

async def pool_example():
    pool = ResourcePool()
    
    async with pool:
        await asyncio.sleep(0.5)
        print("리소스 사용 중")

asyncio.run(pool_example())

async for: 비동기 반복

비동기 이터레이터 구현

일반적인 for 문은 모든 데이터가 메모리에 준비된 상태에서 순차적으로 반복한다. 하지만 대용량 파일 처리, 실시간 데이터 스트림, 페이지네이션 API 호출 등에서는 데이터를 필요할 때마다 비동기적으로 가져와야 한다.

비동기 이터레이터는 __aiter__()와 __anext__() 메서드를 구현한 객체다. __aiter__()는 이터레이터 자신을 반환하고, __anext__()는 각 항목을 비동기적으로 생성하여 반환한다. 반복이 끝나면 StopAsyncIteration 예외를 발생시켜 종료를 알린다.

class AsyncRange:
    def __init__(self, end: int, delay: float = 0.1):
        self.end = end
        self.current = 0
        self.delay = delay
    
    def __aiter__(self):
        return self
    
    async def __anext__(self):
        if self.current >= self.end:
            raise StopAsyncIteration
        
        await asyncio.sleep(self.delay)
        value = self.current
        self.current += 1
        return value

async def async_iteration():
    print("비동기 반복 시작")
    async for i in AsyncRange(3):
        print(f"값: {i}")

asyncio.run(async_iteration())

실용적 활용 - 데이터 스트리밍

실제 환경에서는 파일을 한 줄씩 읽거나, API에서 페이지별로 데이터를 가져오거나, 실시간 로그를 처리할 때 비동기 이터레이터가 유용하다. 각 데이터 항목을 가져오는 데 시간이 걸리는 작업에서 다른 코루틴들이 블로킹되지 않도록 해준다.

class DataStreamer:
    def __init__(self, data_list: list):
        self.data = data_list
        self.index = 0
    
    def __aiter__(self):
        return self
    
    async def __anext__(self):
        if self.index >= len(self.data):
            raise StopAsyncIteration
        
        # 데이터 처리 시뮬레이션
        await asyncio.sleep(0.2)
        item = self.data[self.index]
        self.index += 1
        
        return f"처리완료: {item}"

async def streaming_example():
    data = ["파일1", "파일2", "파일3"]
    
    async for processed_item in DataStreamer(data):
        print(processed_item)

asyncio.run(streaming_example())

비동기 제너레이터 (async def + yield)

기본 사용법

일반 제너레이터는 def 함수에서 yield를 사용하지만, 비동기 제너레이터는 async def 함수에서 yield를 사용한다. 가장 큰 차이점은 각 yield 사이사이에서 await를 사용할 수 있다는 것이다.

이는 데이터를 생성하는 과정 자체가 비동기 작업을 포함할 때 매우 유용하다. 예를 들어, 외부 API를 호출해서 데이터를 가져온 후 가공하여 반환하거나, 데이터베이스에서 배치 단위로 조회하여 처리하는 경우에 활용된다.

async def async_counter(start: int, end: int):
    for i in range(start, end):
        await asyncio.sleep(0.1)  # 비동기 작업
        yield f"카운트: {i}"

async def generator_example():
    async for count in async_counter(1, 4):
        print(count)

asyncio.run(generator_example())

실용적 활용 - 배치 처리

대용량 데이터를 처리할 때는 메모리 효율성을 위해 배치 단위로 나누어 처리한다. 비동기 제너레이터를 사용하면 배치를 생성하는 과정과 처리하는 과정을 모두 비동기적으로 수행할 수 있어, 전체적인 처리량이 크게 향상된다.

async def batch_processor(items: list, batch_size: int = 2):
    batch = []
    
    for item in items:
        batch.append(item)
        
        if len(batch) >= batch_size:
            # 배치 처리
            await asyncio.sleep(0.3)  # 처리 시간
            yield f"배치 처리: {batch}"
            batch = []
    
    # 남은 항목 처리
    if batch:
        await asyncio.sleep(0.3)
        yield f"마지막 배치: {batch}"

async def batch_example():
    items = ["A", "B", "C", "D", "E"]
    
    async for result in batch_processor(items):
        print(result)

asyncio.run(batch_example())

asyncio.as_completed() 활용

asyncio.gather()는 모든 작업이 완료될 때까지 기다리지만, as_completed()는 완료되는 순서대로 결과를 받을 수 있다. 이는 빠른 작업의 결과를 먼저 처리하고 싶거나, 진행 상황을 실시간으로 표시하고 싶을 때 유용하다.

특히 여러 외부 API 호출에서 응답 시간이 다를 때, 느린 API를 기다리지 않고 빠른 API의 결과부터 먼저 사용자에게 보여줄 수 있다.

async def task_with_delay(name: str, delay: float):
    await asyncio.sleep(delay)
    return f"{name} 완료"

async def as_completed_example():
    tasks = [
        task_with_delay("작업A", 1.0),
        task_with_delay("작업B", 0.3),
        task_with_delay("작업C", 0.7),
    ]
    
    # 완료되는 순서대로 처리
    for completed_task in asyncio.as_completed(tasks):
        result = await completed_task
        print(result)

asyncio.run(as_completed_example())

복합 패턴 활용

실제 프로덕션 환경에서는 여러 비동기 문법을 조합하여 사용하는 경우가 많다. 예를 들어, 데이터 처리 파이프라인에서는 자원 관리(async with)와 스트리밍 처리(async for)를 함께 사용한다.

이런 복합 패턴을 사용하면 코드의 가독성을 높이면서도 자원 안전성과 성능을 모두 확보할 수 있다. 특히 컨텍스트 매니저로 전체 프로세스를 관리하고, 비동기 제너레이터로 데이터를 스트리밍하는 조합은 매우 일반적이다.

class AsyncProcessor:
    def __init__(self, name: str):
        self.name = name
        self.processed = 0
    
    async def __aenter__(self):
        print(f"{self.name} 프로세서 시작")
        return self
    
    async def __aexit__(self, exc_type, exc_val, exc_tb):
        print(f"{self.name} 완료 (처리: {self.processed}개)")
    
    async def process_items(self, items: list):
        for item in items:
            await asyncio.sleep(0.1)
            self.processed += 1
            yield f"처리: {item} -> 결과{self.processed}"

async def complex_pattern():
    items = ["데이터1", "데이터2", "데이터3"]
    
    # 컨텍스트 매니저 + 비동기 제너레이터 + async for
    async with AsyncProcessor("메인") as proc:
        async for result in proc.process_items(items):
            print(result)

asyncio.run(complex_pattern())

에러 처리

비동기 문법에서도 예외 처리는 매우 중요하다. 특히 async for 문을 사용할 경우, 내부에서 발생하는 예외는 루프를 즉시 종료시키므로, 각 반복마다 예외를 개별적으로 처리하려면 async for 대신 __anext__()를 직접 호출하는 while 루프 구조가 필요하다.

import asyncio

class SafeAsyncIterator:
    def __init__(self, items: list):
        self.items = items
        self.index = 0

    def __aiter__(self):
        return self

    async def __anext__(self):
        if self.index >= len(self.items):
            raise StopAsyncIteration

        item = self.items[self.index]
        self.index += 1

        # 에러 시뮬레이션
        if "error" in item:
            raise ValueError(f"처리 실패: {item}")

        await asyncio.sleep(0.1)
        return f"성공: {item}"

async def error_handling():
    items = ["정상1", "error", "정상2"]
    iterator = SafeAsyncIterator(items)
    async_iter = iterator.__aiter__()

    while True:
        try:
            item = await async_iter.__anext__()
            print(item)
        except StopAsyncIteration:
            break
        except ValueError as e:
            print(f"에러 처리: {e}")

asyncio.run(error_handling())

결과

성공: 정상1
에러 처리: 처리 실패: error
성공: 정상2

비동기 컨텍스트 매니저에서는 __aexit__() 메서드의 매개변수를 통해 예외 정보를 받을 수 있어, 정리 작업 시 예외 상황에 따른 다른 처리가 가능하다. 예외를 억제하려면 True를, 재발생시키려면 False를 반환하면 된다.

import asyncio

class SafeAsyncContext:
    async def __aenter__(self):
        print("🔓 리소스 열기")
        return self

    async def __aexit__(self, exc_type, exc_value, traceback):
        if exc_type:
            print(f"⚠️ 예외 발생 감지: {exc_value}")
            # 예외를 억제 (False를 반환하면 예외가 다시 발생함)
            return True  # True를 반환하면 예외를 억제함
        print("🔒 정상 종료: 리소스 닫기")
        return False

    async def risky_operation(self):
        await asyncio.sleep(0.1)
        raise RuntimeError("예기치 못한 오류 발생!")

async def main():
    async with SafeAsyncContext() as ctx:
        await ctx.risky_operation()
        print("이 줄은 실행되지 않음 (예외로 인해)")

asyncio.run(main())

결과

🔓 리소스 열기
⚠️ 예외 발생 감지: 예기치 못한 오류 발생!

__aenter__()는 async with가 시작될 때 실행되고, 리소스를 반환하거나 초기화 작업을 수행한다. __aexit__()는 async with 블록이 종료될 때 호출되며, 예외 발생 여부에 관계없이 실행된다.

예외가 발생한 경우 exc_type, exc_value, traceback에 예외 정보가 전달된다.
True를 반환하면 예외가 억제(suppressed) 되어 바깥으로 전파되지 않고 처리된다.
False를 반환하거나 아무것도 반환하지 않으면 예외가 다시 발생한다.

마무리

비동기 문법들(async with, async for, 비동기 제너레이터)은 복잡한 비동기 워크플로우를 깔끔하게 구현할 수 있게 해준다. 자원 관리, 스트리밍 처리, 배치 작업에서 특히 유용하다. 이러한 문법들을 적절히 조합하면 강력하고 읽기 쉬운 비동기 코드를 작성할 수 있다.

다음 편에서는 비동기 프로그래밍의 성능을 측정하고 최적화하는 방법을 알아보겠다.

다음 편 예고

6편에서는 다음 내용을 다룰 예정이다:

비동기 성능 측정 도구와 방법론
프로덕션 환경에서의 모니터링
성능 최적화 팁과 베스트 프랙티스
실제 성능 비교 실험과 결과 분석

저작자표시 (새창열림)

'Python > 문법' 카테고리의 다른 글

[Python] 비동기(async) 프로그래밍 4 (0)	2025.05.22
[Python] 비동기(async) 프로그래밍 3 (0)	2025.05.22
[Python] 비동기(async) 프로그래밍 2 (0)	2025.05.22
[Python] 비동기(async) 프로그래밍 1 (0)	2025.05.22
[Python] Enum 정리 (0)	2025.04.12

내 블로그 - 관리자 홈 전환	`Q` `Q`
새 글 쓰기	`W` `W`

글 수정 (권한 있는 경우)	`E` `E`
댓글 영역으로 이동	`C` `C`

이 페이지의 URL 복사	`S` `S`
맨 위로 이동	`T` `T`
티스토리 홈 이동	`H` `H`
단축키 안내	`Shift` + `/` `⇧` + `/`

comgu