비동기 프로그래밍(asyncio)

동시성 모델 선택: 스레드, 프로세스, 비동기 I/O

Python에서 "여러 일을 동시에"하는 방법은 세 가지다.

asyncio는 단일 스레드 내에서 협력적 멀티태스킹(cooperative multitasking) 을 구현한다. 코루틴이 I/O를 기다리는 동안 자발적으로 제어권을 넘겨주기 때문에, OS 스케줄러 개입 없이도 수천 개의 동시 연결을 처리할 수 있다. 데이터베이스 쿼리, HTTP 호출, 파일 I/O가 많은 파이프라인 코드에 최적이다.

코루틴(Coroutine): async/await의 실체

async def로 정의한 함수는 호출 즉시 실행되지 않는다. 코루틴 객체(coroutine object)를 반환할 뿐이다.

import asyncio

async def fetch_data(url: str) -> str:
    print(f"→ 요청 시작: {url}")
    await asyncio.sleep(1)          # I/O 대기 시뮬레이션: 제어권을 이벤트 루프에 반환
    print(f"← 요청 완료: {url}")
    return f"data from {url}"

# 코루틴 객체 생성 — 아직 실행 안 됨
coro = fetch_data("https://example.com")

# 이벤트 루프를 시작하고 코루틴을 실행
result = asyncio.run(coro)
print(result)

await는 두 가지 일을 한다. ① 현재 코루틴을 일시 중단하고, ② 이벤트 루프가 다른 준비된 코루틴을 실행할 수 있도록 제어권을 넘긴다. await 뒤에는 어웨이터블(awaitable) — 코루틴, Task, Future — 이 올 수 있다.

이벤트 루프 동작 원리

이벤트 루프는 크게 세 단계를 반복한다. ① Ready 큐에 있는 콜백과 코루틴을 모두 실행, ② I/O 셀렉터를 폴링해 완료된 I/O를 Ready 큐에 추가, ③ 만료된 예약 콜백을 Ready 큐에 추가. 이 사이클이 초당 수천 번 돌면서 수많은 동시 I/O를 처리한다.

Task와 Future

Task는 코루틴을 감싸서 이벤트 루프가 관리하는 실행 단위다. Future는 아직 완료되지 않은 연산 결과를 담는 저수준 객체다. 보통 직접 Future를 만들 일은 없고, asyncio.create_task()로 Task를 만든다.

import asyncio

async def worker(name: str, delay: float) -> str:
    await asyncio.sleep(delay)
    return f"{name} 완료"

async def main() -> None:
    # create_task: 즉시 스케줄링 — main이 await하지 않아도 백그라운드에서 시작
    task_a = asyncio.create_task(worker("A", 2.0))
    task_b = asyncio.create_task(worker("B", 1.0))

    # await으로 결과를 기다린다
    result_a = await task_a   # B는 이미 완료 상태일 수 있음
    result_b = await task_b
    print(result_a, result_b)

asyncio.run(main())
# B 완료 (먼저 출력되지 않음 — await 순서가 출력 순서를 결정)
# B 완료, A 완료 (결과 자체는 병렬로 실행됨)

gather: 여러 코루틴을 동시에 실행

import asyncio, aiohttp

async def fetch(session: aiohttp.ClientSession, url: str) -> bytes:
    async with session.get(url) as resp:
        return await resp.read()

async def main(urls: list[str]) -> list[bytes]:
    async with aiohttp.ClientSession() as session:
        tasks = [fetch(session, url) for url in urls]
        # gather: 모두 끝날 때까지 기다리고 결과를 순서대로 반환
        results: list[bytes] = await asyncio.gather(*tasks)
    return results

asyncio.gather()는 모든 코루틴을 동시에 스케줄링하고, 모두 완료될 때까지 기다린다. 결과 순서는 인자 순서를 보장한다. return_exceptions=True를 주면 하나가 실패해도 나머지를 계속 실행한다.

Semaphore로 동시 실행 수 제어

네트워크 연결 수나 외부 API rate limit을 초과하지 않으려면 Semaphore를 쓴다.

import asyncio, aiohttp

async def fetch_limited(
    session: aiohttp.ClientSession,
    url: str,
    sem: asyncio.Semaphore,
) -> bytes:
    async with sem:             # 동시 실행 허용 슬롯을 1개 점유
        async with session.get(url) as resp:
            return await resp.read()

async def main(urls: list[str]) -> None:
    sem = asyncio.Semaphore(10)  # 최대 10개 동시 요청
    async with aiohttp.ClientSession() as session:
        await asyncio.gather(
            *(fetch_limited(session, url, sem) for url in urls)
        )

asyncio.Lock()은 뮤텍스처럼 동작한다(Semaphore(1)). 공유 상태를 수정할 때만 락으로 감싸고, 나머지 로직은 바깥에서 비동기로 실행한다.

타임아웃 처리

import asyncio

async def slow_query() -> str:
    await asyncio.sleep(10)
    return "결과"

async def main() -> None:
    # Python 3.11+ 권장 방식: asyncio.timeout()
    try:
        async with asyncio.timeout(3.0):
            result = await slow_query()
    except TimeoutError:
        print("3초 초과 — 쿼리 취소됨")

    # 구 버전 호환: asyncio.wait_for()
    try:
        result = await asyncio.wait_for(slow_query(), timeout=3.0)
    except asyncio.TimeoutError:
        print("타임아웃")

asyncio.run(main())

asyncio.timeout()(Python 3.11, PEP 647)은 컨텍스트 매니저 방식으로 더 읽기 쉽다. asyncio.wait_for()는 3.11 이전 코드와 호환된다.

asyncio.Queue: 프로듀서-컨슈머 패턴

파이프라인에서 생산 속도와 소비 속도가 다를 때 큐로 버퍼링한다.

import asyncio

async def producer(queue: asyncio.Queue[int], n: int) -> None:
    for i in range(n):
        await queue.put(i)
        print(f"  생산: {i}")
        await asyncio.sleep(0.1)
    await queue.put(None)       # 종료 신호

async def consumer(queue: asyncio.Queue[int]) -> None:
    while True:
        item = await queue.get()
        if item is None:
            break
        print(f"소비: {item}")
        await asyncio.sleep(0.3)  # 소비가 더 느림
        queue.task_done()

async def main() -> None:
    queue: asyncio.Queue[int] = asyncio.Queue(maxsize=5)  # 버퍼 크기 제한
    await asyncio.gather(producer(queue, 10), consumer(queue))

asyncio.run(main())

maxsize가 0이면 무한 버퍼, 양수이면 생산자가 put()에서 블록된다.

CPU-bound 코드와 asyncio 통합: run_in_executor

asyncio는 단일 스레드이므로 CPU-bound 작업이 이벤트 루프를 막는다. loop.run_in_executor()로 ThreadPoolExecutor 또는 ProcessPoolExecutor에 위임한다.

import asyncio
from concurrent.futures import ProcessPoolExecutor
import math

def cpu_heavy(n: int) -> bool:
    """소수 판별 — CPU를 오래 씀."""
    if n < 2:
        return False
    return all(n % i != 0 for i in range(2, int(math.sqrt(n)) + 1))

async def main() -> None:
    loop = asyncio.get_running_loop()
    with ProcessPoolExecutor() as pool:
        # 이벤트 루프를 막지 않고 별도 프로세스에서 실행
        result = await loop.run_in_executor(pool, cpu_heavy, 999_999_937)
    print(result)   # True

asyncio.run(main())

순수 I/O라면 ThreadPoolExecutor로도 충분하다. run_in_executor(None, fn, ...) — None은 기본 스레드 풀이다.

asynccontextmanager: 비동기 컨텍스트 매니저

from contextlib import asynccontextmanager
from typing import AsyncIterator
import asyncpg

@asynccontextmanager
async def db_connection(dsn: str) -> AsyncIterator[asyncpg.Connection]:
    conn = await asyncpg.connect(dsn)
    try:
        yield conn
    finally:
        await conn.close()

async def main() -> None:
    async with db_connection("postgresql://localhost/app") as conn:
        rows = await conn.fetch("SELECT id FROM users LIMIT 10")

__aenter__/__aexit__를 직접 구현해도 되지만, @asynccontextmanager가 훨씬 간결하다.

실전 패턴 요약

흔한 실수와 주의점

1. 코루틴을 await 없이 호출

async def bad() -> None:
    fetch_data("url")  # await 빠짐 → 코루틴 객체 생성만 하고 버림
    # RuntimeWarning: coroutine 'fetch_data' was never awaited

2. 동기 블로킹 코드를 이벤트 루프 안에서 직접 호출

import time, requests

async def bad() -> None:
    time.sleep(5)            # 이벤트 루프 전체를 5초 동안 막음
    requests.get("...")      # 동기 HTTP — 같은 문제

asyncio.sleep()과 aiohttp처럼 async-native 라이브러리를 써야 한다. 어쩔 수 없으면 run_in_executor로 위임한다.

3. 여러 스레드에서 이벤트 루프 공유

asyncio 객체(Task, Lock, Queue)는 생성된 이벤트 루프에 묶인다. 별도 스레드에서 접근하려면 asyncio.run_coroutine_threadsafe()를 사용한다.

정리

1. asyncio는 단일 스레드 협력적 멀티태스킹 — I/O-bound 코드에 적합.
2. 코루틴은 async def로 정의, await으로 일시 중단하고 제어권을 이벤트 루프에 넘긴다.
3. 이벤트 루프는 ① Ready 실행 → ② I/O poll → ③ 예약 콜백 확인을 반복한다.
4. create_task() 는 즉시 스케줄링; gather() 는 여러 코루틴을 동시 실행하고 결과를 모은다.
5. Semaphore로 동시 실행 수를 제한, Queue로 생산·소비 속도 차이를 버퍼링한다.
6. CPU-bound 작업은 run_in_executor() + ProcessPoolExecutor로 이벤트 루프 블로킹을 방지한다.
7. asyncio.timeout()(3.11+) 또는 wait_for()로 시간 제한을 건다.

References

• Python Docs, asyncio — Coroutines and Tasks: https://docs.python.org/3/library/asyncio-task.html
• Python Docs, asyncio — Event Loop: https://docs.python.org/3/library/asyncio-eventloop.html
• Python Docs, asyncio — Synchronization Primitives: https://docs.python.org/3/library/asyncio-sync.html
• Python Docs, asyncio — Developing with asyncio (pitfalls): https://docs.python.org/3/library/asyncio-dev.html
• Python Docs, A Conceptual Overview of asyncio: https://docs.python.org/3/howto/asyncio-conceptual-overview.html
• Real Python, "Python's asyncio: A Hands-On Walkthrough": https://realpython.com/async-io-python/
• SuperFastPython, "Python Asyncio: The Complete Guide": https://superfastpython.com/python-asyncio/
• DEV Community, "Asyncio Architecture in Python: Event Loops, Tasks, and Futures Explained": https://dev.to/imsushant12/asyncio-architecture-in-python-event-loops-tasks-and-futures-explained-4pn3