비동기 입문 (asyncio)

2부의 마지막 챕터입니다. 11장 이터러블, 제너레이터, yield from에서 잠깐 본 send / yield 메커니즘 위에 만들어진 도구가 asyncio와 async/await입니다. 본 챕터는 그 첫 걸음을 한곳에 정리합니다.

본 챕터는 3부의 18장 비동기 깊이 — 이벤트 루프, gather/wait, async generator와 19장 GIL과 동시성의 전제가 되는 모델을 깝니다. 그리고 4부 FastAPI의 거의 모든 코드가 async def 위에 올라가므로, 본 챕터를 분명히 잡아두는 게 책 후반부 전체에 도움이 됩니다.

동기 vs 비동기 — 한 그림 #

import time

def fetch(url: str) -> str:
    time.sleep(1)             # 네트워크 응답 대기 (1초)
    return f"data from {url}"

def main():
    a = fetch("a")            # 1초
    b = fetch("b")            # 1초
    c = fetch("c")            # 1초
    print(a, b, c)            # 총 3초

main()

세 개의 요청을 차례로 합니다. 각각 1초씩, 총 3초.

비동기로 같은 일을 하면:

import asyncio

async def fetch(url: str) -> str:
    await asyncio.sleep(1)
    return f"data from {url}"

async def main():
    results = await asyncio.gather(
        fetch("a"),
        fetch("b"),
        fetch("c"),
    )
    print(results)            # 총 1초

asyncio.run(main())

세 요청이 동시에 진행됩니다. 총 1초.

이게 비동기의 가치입니다. CPU가 아니라 기다림이 병목인 경우 — 네트워크, DB, 파일 — 에서 동시성으로 시간을 법니다. CPU가 병목인 경우는 19장 GIL과 동시성에서 threading / multiprocessing과 비교합니다.

핵심 개념 — 한곳에 #

1) 코루틴 함수와 코루틴 객체 #

async def fetch(url):    # 코루틴 함수 (coroutine function)
    return ...

c = fetch("a")            # c 는 코루틴 객체 (coroutine object)
                          # 함수 본문은 아직 실행 안 됨!

async def로 정의하면 호출 시 본문을 실행하지 않고 코루틴 객체를 반환 합니다. 제너레이터 함수와 같은 패턴입니다 — 11장에서 본 일시정지 / 재개 모델 그대로.

코루틴이 실제로 실행되려면 이벤트 루프가 그것을 돌려야 합니다.

2) await — 결과를 기다림 #

async def main():
    data = await fetch("a")    # fetch 가 끝날 때까지 기다림
    print(data)

await가 하는 일:

• 표현식의 값(코루틴, Task 등)을 이벤트 루프에 양도
• 그것이 끝나면 결과를 받아 다음 줄을 진행
• 양도 동안 다른 코루틴이 실행될 수 있음

await는 async def 안에서만 쓸 수 있습니다.

3) 이벤트 루프 #

코루틴들을 번갈아 실행하는 스케줄러. 한 번에 한 코루틴만 돌리지만, 그게 await로 양도하면 다른 코루틴이 실행을 이어받습니다. 단일 스레드에서 동시성을 만드는 메커니즘입니다. 깊이는 18장 비동기 깊이에서.

asyncio.run(coro)가 이벤트 루프를 만들고 coro를 끝까지 돌린 뒤 정리해 줍니다. 프로그램 진입점에서 한 번만 호출 하는 게 일반적입니다.

동시 실행의 두 표준 — gather와 TaskGroup #

asyncio.gather — 클래식 형태 #

async def main():
    a, b, c = await asyncio.gather(
        fetch("a"),
        fetch("b"),
        fetch("c"),
    )

gather는 여러 코루틴을 동시에 시작하고, 모두 끝나면 결과를 리스트로 줍니다. 옛날부터 있던 표준 도구입니다.

TaskGroup (3.11+) — 새 표준 #

async def main():
    async with asyncio.TaskGroup() as tg:
        a = tg.create_task(fetch("a"))
        b = tg.create_task(fetch("b"))
        c = tg.create_task(fetch("c"))

    print(a.result(), b.result(), c.result())

같은 동작인데 두 가지 면에서 더 안전합니다.

1. 중간에 예외가 나면 다른 task 들도 cleanup — gather는 옵션 없으면 그냥 진행
2. **여러 task가 동시에 실패하면 6장 예외 처리의 ExceptionGroup**으로 묶어줌

3.11+ 환경이면 TaskGroup이 더 권장되는 형태입니다.

Task — 코루틴을 백그라운드로 #

코루틴을 즉시 시작하고 백그라운드로 돌리고 싶을 때 Task로 만듭니다.

async def main():
    task = asyncio.create_task(fetch("a"))
    # 여기서 task 는 이미 실행 중

    other_work()              # 다른 일 진행

    result = await task        # 필요할 때 결과 회수

코루틴 객체 자체는 누가 await 해 주기 전엔 안 도는 단순 값입니다. Task로 감싸야 이벤트 루프에 등록되고 실행이 시작됩니다.

흔한 함정 — 잊지 말아야 할 것 #

1) 동기 함수 안에서 await 못 씀 #

def main():
    data = await fetch("a")    # ✗ SyntaxError

await는 async def 안에서만. 진입점은 asyncio.run(...)으로 동기 코드에서 비동기를 시작.

2) time.sleep은 비동기를 막는다 #

async def fetch(url):
    time.sleep(1)              # 이벤트 루프 전체가 멈춤
    return ...

async def fetch(url):
    await asyncio.sleep(1)
    return ...

time.sleep은 OS 스레드 전체를 잠재웁니다. 비동기 안에서 쓰면 동시성이 깨집니다. 동기 블로킹 함수 (DB 드라이버, requests, 파일 I/O 일부)도 같은 문제가 있습니다 — 비동기용 라이브러리 (asyncpg, httpx, aiofiles)를 써야 합니다.

3) 코루틴 만들고 await 안 함 #

async def main():
    fetch("a")                # 만들고 await 안 함 — 안 돌아감
    print("done")

이러면 경고가 뜹니다 (coroutine 'fetch' was never awaited). 코루틴은 값일 뿐이라 누군가 await 또는 Task로 만들어주지 않으면 절대 안 돕니다.

4) 동기 코드에서 동기처럼 호출하면 안 됨 #

async def fetch(url): ...

result = fetch("a")    # 코루틴 객체. 결과 아님.

이걸 풀어내려면 진입점에서 asyncio.run(...).

동기 코드와 섞기 #

동기 함수가 오래 걸린다면 — to_thread #

import asyncio

def heavy_calc(n: int) -> int:
    # CPU 바운드든 블로킹 I/O 든
    ...

async def main():
    result = await asyncio.to_thread(heavy_calc, 100)

asyncio.to_thread(fn, *args)가 동기 함수를 별도 스레드에서 실행 하고 결과를 await 가능한 형태로 만들어줍니다. 비동기 코드 안에서 동기 라이브러리를 어쩔 수 없이 써야 할 때 자주 등장.

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

10장 컨텍스트 매니저에서 짧게 본 도구입니다.

import httpx

async def main():
    async with httpx.AsyncClient() as client:
        resp = await client.get("https://api.example.com")
        data = resp.json()

__enter__ / __exit__ 대신 __aenter__ / __aexit__가 정의된 객체. @asynccontextmanager로 함수에서 만들 수도 있습니다.

from contextlib import asynccontextmanager

@asynccontextmanager
async def db_transaction(conn):
    tx = await conn.begin()
    try:
        yield tx
        await tx.commit()
    except Exception:
        await tx.rollback()
        raise

비동기 이터레이터 — async for #

async def stream_lines(url: str):
    async with httpx.AsyncClient() as client:
        async with client.stream("GET", url) as resp:
            async for line in resp.aiter_lines():
                yield line

__iter__ / __next__ 대신 __aiter__ / __anext__ — 비동기로 다음 값을 받는 패턴. 스트리밍에 자주 등장합니다. async generator의 깊이는 18장 비동기 깊이에서.

타임아웃 — asyncio.timeout (3.11+) #

async def main():
    try:
        async with asyncio.timeout(5):
            data = await fetch("https://slow.example.com")
    except TimeoutError:
        print("5초 안에 못 받음")

3.10 이전에는 asyncio.wait_for(coro, timeout=5)가 표준이었습니다. 새 코드는 asyncio.timeout 컨텍스트 매니저가 권장됩니다.

자주 만나는 패턴 모음 #

동시 N 개로 제한 — Semaphore #

async def fetch_all(urls: list[str], concurrency: int = 10):
    sem = asyncio.Semaphore(concurrency)

    async def fetch_one(url):
        async with sem:
            return await fetch(url)

    return await asyncio.gather(*[fetch_one(u) for u in urls])

URL 1000개를 한꺼번에 동시 요청하면 서버가 요청을 제한하거나 메모리 / 연결이 고갈될 수 있습니다. Semaphore로 동시에 N 개까지만 요청되도록 제한합니다.

첫 번째로 끝나는 것만 받기 — as_completed #

async def race(tasks):
    for coro in asyncio.as_completed(tasks):
        result = await coro
        if result.is_winner:
            return result

여러 동시 작업 중 먼저 끝난 것부터 처리하고 싶을 때.

취소 #

task = asyncio.create_task(fetch("a"))
await asyncio.sleep(0.5)
task.cancel()    # CancelledError 가 task 안에서 던져짐

타임아웃, 사용자 취소, 부모 취소 등이 모두 본 메커니즘 위에 만들어집니다. 비동기 코드는 취소될 수 있다는 가정으로 짜는 게 좋습니다 — try/except CancelledError로 정리 코드를 두는 패턴이 자주 등장합니다.

어디까지가 입문인가 #

본 챕터는 개념 소개 까지입니다. 실전에서는 다음 주제들이 추가로 들어옵니다.

• 이벤트 루프의 내부 동작과 정책 → 18장 비동기 깊이
• Future vs Task의 정확한 차이 → 18장
• 동기 / 비동기 라이브러리 선택과 통합 → 19장 GIL과 동시성
• 성능 디버깅 (어디서 await가 길어지는지) → 21장 성능
• 백그라운드 작업, 워커 패턴 → 27장 비동기와 백그라운드 작업
• 웹 프레임워크 (FastAPI)와의 통합 → 4부 전체

연습문제 #

1. async def fetch(url: str) -> str:가 await asyncio.sleep(1) 후 더미 결과를 반환한다고 가정하세요. 세 URL을 (1) 순차로 await 하기 와 (2) asyncio.gather로 동시에 실행하기 두 버전을 작성하고 실제 걸리는 시간을 time.perf_counter()로 측정합니다.
2. asyncio.TaskGroup으로 같은 일을 다시 작성하세요. 한 task에서 일부러 예외를 던지도록 만들고, 나머지 task 들이 어떻게 cleanup 되는지, 최종 예외가 ExceptionGroup으로 묶이는지 확인합니다.
3. Semaphore(5)로 동시성을 5로 제한해 URL 100개를 처리하는 함수를 작성하세요. 동시성을 1 / 5 / 100으로 바꿔가며 총 시간 변화를 비교합니다. 실제 운영에서 어떤 값을 고를지 직접 가설을 세워 보겠습니다.

한 줄 요약: 비동기는 CPU가 아니라 대기가 병목인 I/O 동시성 도구. async def = 코루틴 함수, await가 이벤트 루프에 양도. 진입점은 asyncio.run. 동시 실행은 gather (옛) / TaskGroup (3.11+ 권장). create_task로 백그라운드, to_thread로 동기 함수 변환, asyncio.timeout으로 타임아웃, Semaphore로 동시성 제한, cancel()로 취소. time.sleep 금지.

2부 마무리 #

2부 7장을 거쳐 코드 구조화 도구가 채워졌습니다.

• 데이터 모양 — @dataclass, __slots__
• 타입 — Generic, Protocol, TypedDict, Literal
• 자원 관리 — with, contextlib
• 흐름 표현 — 이터러블, 제너레이터, yield from
• 함수 감싸기 — 데코레이터의 모든 패턴
• 분기 표현 — match-case 깊이
• 동시성 — asyncio 입문

다음 3부 깊이 · 동시성은 라이브러리 / 프레임워크에서 만나는 깊이 — 매직 메소드, 디스크립터, 메타클래스, 비동기 깊이, GIL / 동시성, typing 고급, 성능 — 으로 들어갑니다.

다음 챕터 #

다음 15장 매직 메소드 깊이와 프로토콜에서는 객체가 언어 기능과 통합되는 모든 후크 — __call__, __getitem__, __hash__, __format__, __getattr__ 등을 다룹니다. 8장 dataclass가 자동으로 만들어주던 매직 메소드들을 직접 만드는 수준으로 들어갑니다.