Quick Flow

import asyncio

async def fetch_data(name: str) -> str:
    await asyncio.sleep(0.5)
    return f"data:{name}"

async def main() -> None:
    results = await asyncio.gather(fetch_data("a"), fetch_data("b"))
    print(results)

asyncio.run(main())

async def 호출은 값을 바로 실행하지 않고 coroutine 객체를 만들며, 실제 실행은 await 또는 task로 event loop에 맡길 때 시작합니다. 독립 I/O를 함께 기다릴 때는 TaskGroup 또는 gather를 고르고, 취소는 CancelledError를 다시 전파할 수 있게 정리합니다.

기본 흐름

기본형은 coroutine 정의, await, 최상위 실행, 함께 기다리기, 먼저 시작하기를 같이 보면 됩니다.

import asyncio

async def work():
    return 1

async def main():
    value = await work()
    results = await asyncio.gather(work(), work())
    task = asyncio.create_task(work())
    result = await task

asyncio.run(main())

async def는 호출 즉시 끝까지 실행되는 함수가 아니라 coroutine 객체를 만듭니다.
실제 실행은 이벤트 루프가 맡고, await는 그 진행을 조율하는 지점입니다.

coroutine 모델

async def를 호출하면 coroutine 객체가 생긴다

async def 함수는 호출 즉시 끝까지 실행되지 않습니다.
우선 coroutine 객체를 만들고, 실제 실행은 이벤트 루프가 담당합니다.

async def work():
    return 1

coro = work()
print(coro)

이 점 때문에 async 코드는 "지금 바로 실행"보다 "스케줄링해서 진행" 감각으로 읽는 편이 맞습니다.

await는 현재 coroutine이 제어권을 잠시 넘기는 지점이다

await는 프로그램 전체를 멈추는 문법이 아니라, 현재 coroutine이 다른 작업이 돌 수 있게 제어권을 넘기는 지점입니다.

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

그래서 async는 기다리는 시간이 많은 문제에 잘 맞고, CPU 계산이 많은 작업을 자동으로 해결해 주지는 않습니다.

동시 대기

여러 결과를 같이 기다리면 gather()를 쓴다

여러 awaitable을 동시에 진행하고 최종 결과를 함께 받고 싶다면 asyncio.gather()가 가장 직접적입니다.

results = await asyncio.gather(fetch_data("a"), fetch_data("b"))

즉 "이 작업들을 묶어서 같이 끝낼까?"라는 질문에는 gather()를 먼저 떠올리면 됩니다. gather()의 기본 동작은 첫 예외를 호출자에게 전달하지만, 이미 제출한 다른 awaitable을 자동으로 취소하지는 않습니다. 한 작업 실패 시 sibling task까지 정리해야 하는 작업 단위라면 TaskGroup이 더 직접적입니다.

async with asyncio.TaskGroup() as group:
    user = group.create_task(fetch_data("user"))
    orders = group.create_task(fetch_data("orders"))

# 둘 다 성공했을 때만 result를 읽는다.
print(user.result(), orders.result())

TaskGroup 안에서 취소가 아닌 예외가 나면 나머지 task를 취소하고 기다린 뒤 ExceptionGroup을 올립니다. 이 차이 때문에 독립 task를 묶은 요청 처리에는 TaskGroup을 먼저 검토합니다.

작업 시작

먼저 시작해 두고 나중에 합류하면 create_task()가 맞다

백그라운드로 작업을 시작해 두고, 중간에 다른 일을 하다가 나중에 합류하고 싶다면 create_task()가 자연스럽습니다.

task = asyncio.create_task(fetch_data("c"))
print("do other work")
result = await task

즉 gather()는 함께 기다리는 패턴, create_task()는 먼저 시작해 두는 패턴으로 나누면 헷갈림이 줄어듭니다.

results = await asyncio.gather(fetch_data("a"), fetch_data("b"))

task = asyncio.create_task(fetch_data("c"))
await do_other_work()
result = await task

둘 다 동시에 시작할 수는 있지만, gather()는 "같이 묶어 기다리기", create_task()는 "지금 시작만 해 두고 합류하기"에 더 가깝습니다.

실행 경계

최상위에서는 보통 asyncio.run()으로 시작한다

스크립트나 프로그램의 시작점에서 async 코드를 실행할 때는 보통 asyncio.run(main())으로 이벤트 루프를 엽니다.
다만 프레임워크나 REPL 환경에서는 이미 루프를 관리할 수 있으므로 실행 환경을 같이 봐야 합니다.

주의할 점

# ❌ coroutine만 만들고 끝냄
async def fetch():
    await asyncio.sleep(1)
    return "done"

fetch()

# ✅ await하거나 task로 스케줄링
result = await fetch()
# 또는
asyncio.create_task(fetch())

# ❌ CPU 계산이 많다고 async만 붙인다고 빨라지지 않음
async def crunch():
    return sum(i * i for i in range(10_000_000))

async def receive() -> None:
    try:
        await asyncio.sleep(60)
    except asyncio.CancelledError:
        await close_connection()
        raise