Python 비동기 처리에서 asyncio.gather() vs TaskGroup 선택 기준

문제 상황

외부 API 100개를 병렬로 호출하는 배치 작업에서 일부 API 실패 시 전체가 중단되는 문제가 있었다. 기존 asyncio.gather()로 구현되어 있었는데, 에러 핸들링 개선이 필요했다.

asyncio.gather()의 한계

# 기존 코드
results = await asyncio.gather(
    *[fetch_api(url) for url in urls],
    return_exceptions=True
)

return_exceptions=True로 예외를 결과에 포함시킬 수 있지만, 어떤 태스크가 실패했는지 추적하기 번거로웠다.

TaskGroup 도입 (Python 3.11+)

async with asyncio.TaskGroup() as tg:
    tasks = [tg.create_task(fetch_api(url)) for url in urls]

# 하나라도 실패하면 ExceptionGroup 발생

TaskGroup은 구조화된 동시성(Structured Concurrency)을 제공한다. 하나의 태스크가 실패하면 나머지를 자동으로 취소하고 ExceptionGroup으로 모든 예외를 수집한다.

선택 기준

결국 요구사항에 따라 구분했다:

  • 부분 실패 허용: gather(return_exceptions=True) 사용
  • All-or-Nothing: TaskGroup 사용
  • 세밀한 제어: 개별 태스크 관리

우리 케이스는 부분 실패를 허용해야 해서 gather()를 유지했다. 대신 결과 처리 로직을 개선해 예외 발생 건을 별도 로깅하도록 수정했다.

성능 차이

100개 API 호출 기준 성능 차이는 미미했다(~5ms). 선택 기준은 성능보다 에러 핸들링 전략이 더 중요했다.