Python 비동기 프로그래밍에서 asyncio.gather vs asyncio.create_task

문제 상황

외부 API 10개를 동시에 호출하는 배치 작업에서 하나의 API가 실패하면 전체가 멈추는 문제가 발생했다. asyncio.gather를 사용하고 있었는데, 기본 동작이 예상과 달랐다.

# 기존 코드 - 하나만 실패해도 전체 중단
results = await asyncio.gather(
    fetch_api_1(),
    fetch_api_2(),
    fetch_api_3()
)

해결 과정

1. gather의 return_exceptions

첫 번째 시도는 return_exceptions=True 옵션이었다.

results = await asyncio.gather(
    fetch_api_1(),
    fetch_api_2(),
    fetch_api_3(),
    return_exceptions=True
)

# results에 정상 값과 Exception이 섞여서 반환됨
for i, result in enumerate(results):
    if isinstance(result, Exception):
        logger.error(f"API {i} failed: {result}")

이 방식은 간단하지만 개별 태스크의 라이프사이클 제어가 어려웠다.

2. create_task로 분리

더 세밀한 제어가 필요해서 태스크를 개별 생성하는 방식으로 변경했다.

tasks = [
    asyncio.create_task(fetch_api(url))
    for url in api_urls
]

results = []
for task in asyncio.as_completed(tasks):
    try:
        result = await task
        results.append(result)
    except Exception as e:
        logger.error(f"Failed: {e}")
        results.append(None)

차이점 정리

• gather: 모든 태스크를 하나의 Future로 묶음. 취소 시 전체 취소.
• create_task: 개별 태스크로 관리. 독립적인 라이프사이클.

실무에서는 대부분 gather로 충분하지만, 개별 재시도나 타임아웃이 필요한 경우 create_task가 유용했다.

추가: 타임아웃 적용

async def fetch_with_timeout(url, timeout=5):
    try:
        return await asyncio.wait_for(fetch_api(url), timeout)
    except asyncio.TimeoutError:
        logger.warning(f"Timeout: {url}")
        return None

Python 3.9부터는 asyncio.timeout() 컨텍스트 매니저도 고려할 만하다.