sayu.day

· 4분 읽기

Python asyncio 비동기 프로그래밍 가이드

#python#concurrency

asyncio의 목적은 멀티스레드 대체가 아니라 "I/O 대기 시간을 겹쳐 처리"하는 것입니다. 외부 API 호출, DB 질의, 네트워크 응답 대기처럼 기다리는 시간이 긴 작업일수록 효과가 큽니다. 이 글은 asyncio를 서비스 코드에 적용할 때 필요한 핵심 원칙과 자주 쓰는 패턴, 흔한 실수, 운영 체크리스트를 정리합니다.

핵심 원칙

  1. I/O 바운드 작업에 사용하고, CPU 바운드 작업에는 별도 executor를 사용합니다.
  2. await 지점이 없는 코드는 동시성이 아닙니다. 이벤트 루프는 await에서만 다른 작업으로 전환할 수 있습니다.
  3. 취소(cancellation)와 타임아웃을 예외 상황이 아니라 기본 경로로 설계합니다.

자주 쓰는 패턴

asyncio.gather: 여러 작업 동시 실행

import asyncio

async def fetch(path: str) -> str:
    await asyncio.sleep(0.1)  # 실제로는 HTTP 호출 등 I/O
    return f"result:{path}"

async def main() -> None:
    results = await asyncio.gather(
        fetch("/users"),
        fetch("/orders"),
        return_exceptions=True,  # 일부 실패해도 나머지 결과 수집
    )
    print(results)

asyncio.run(main())

return_exceptions 사용 여부에 따라 실패 전파 방식이 달라지므로, 호출부의 실패 처리 규칙을 먼저 정해야 합니다.

TaskGroup(3.11+): 구조적 동시성

async def main() -> None:
    async with asyncio.TaskGroup() as tg:
        tg.create_task(fetch("/users"))
        tg.create_task(fetch("/orders"))
    # 블록을 벗어나면 모든 태스크의 완료가 보장됩니다.
    # 하나라도 실패하면 나머지는 취소되고 예외가 묶여 전파됩니다.

gather와 달리 실패 시 형제 태스크가 자동으로 취소되므로 실패 전파가 명확해집니다.

Semaphore: 외부 API 동시 호출 제한

sem = asyncio.Semaphore(10)  # 동시 호출 10개로 제한

async def call_api(path: str) -> str:
    async with sem:
        return await fetch(path)

async def main() -> None:
    paths = [f"/items/{i}" for i in range(100)]
    await asyncio.gather(*(call_api(p) for p in paths))

동시 실행 상한이 없으면 순간적으로 태스크가 몰릴 때 외부 시스템과 자기 자신 모두를 위협합니다.

Queue: producer/consumer 파이프라인

async def producer(q: asyncio.Queue) -> None:
    for i in range(100):
        await q.put(i)  # 큐가 가득 차면 대기(배압)

async def consumer(q: asyncio.Queue) -> None:
    while True:
        item = await q.get()
        await handle(item)  # 실제 처리 로직(I/O)
        q.task_done()

async def main() -> None:
    q = asyncio.Queue(maxsize=50)
    workers = [asyncio.create_task(consumer(q)) for _ in range(5)]
    await producer(q)
    await q.join()  # 큐에 남은 작업의 처리 완료 대기
    for w in workers:
        w.cancel()

maxsize를 지정하면 생산 속도가 소비 속도를 앞지를 때 자연스럽게 배압이 걸립니다.

실무에서 흔한 실수

  1. 동기 라이브러리를 async 코드에서 직접 호출해 이벤트 루프 전체를 멈추게 합니다.
  2. 무제한 태스크 생성으로 메모리와 소켓을 고갈시킵니다.
  3. 백그라운드 태스크의 예외를 수거하지 않아 실패가 조용히 유실됩니다.
  4. 종료 시 태스크 취소와 정리를 누락해 리소스가 새거나 종료가 지연됩니다.

운영 체크리스트

  1. 외부 호출에 타임아웃 기본값을 적용합니다.
  2. 재시도는 idempotency가 보장되는 경로에만 사용합니다.
  3. 태스크 수, 큐 길이, 실패율 메트릭을 수집합니다.
  4. shutdown 시 graceful cancellation을 구현합니다.

정리

asyncio 품질은 문법보다 제어흐름에서 결정됩니다. 동시 실행 개수와 실패 처리 규칙을 먼저 고정하면 안정성이 크게 올라갑니다. 패턴 선택 기준은 단순합니다. 전부 모아서 기다리면 gather, 실패 전파를 명확히 하려면 TaskGroup, 동시 개수를 제한하려면 Semaphore, 흐름을 분리하고 배압이 필요하면 Queue입니다.