Python asyncio 완전 정복 — 이벤트 루프, 코루틴, 동시성 패턴 실전 가이드

이벤트 루프 기초 — asyncio.run()이 정답
import asyncio
import time

async def fetch_data(name: str, delay: float) -> str:
    print(f"[{name}] 시작")
    await asyncio.sleep(delay)  # I/O 대기 시뮬레이션
    print(f"[{name}] 완료 ({delay}s)")
    return f"{name} 결과"

async def main():
    start = time.perf_counter()

    # 순차 실행: 총 3초 소요
    result1 = await fetch_data("A", 1.0)
    result2 = await fetch_data("B", 2.0)

    elapsed = time.perf_counter() - start
    print(f"순차 실행: {elapsed:.2f}s")  # 약 3.0s

asyncio.run(main())

# asyncio.run()은 내부적으로:
# 1. 새 이벤트 루프 생성
# 2. main() 코루틴 실행
# 3. 루프 종료 및 정리 (pending tasks 취소, 소켓 닫기 등)

코루틴 vs Task — 실행 시점의 차이
import asyncio

async def work(name: str) -> str:
    print(f"{name}: 실행 시작")
    await asyncio.sleep(1)
    return f"{name}: 완료"

async def main():
    # 코루틴 객체 생성 (아직 실행 안 됨)
    coro = work("코루틴")
    print("코루틴 객체 생성됨 (실행 전)")

    # Task 생성: 이벤트 루프에 즉시 등록, 백그라운드에서 시작
    task = asyncio.create_task(work("Task"))
    print("Task 등록됨 (백그라운드 실행 시작)")

    # 다른 작업 수행 중에도 task는 실행됨
    await asyncio.sleep(0)  # 이벤트 루프에 제어권 양보 → task가 시작됨

    # 코루틴은 여기서 처음 실행됨
    result_coro = await coro
    result_task = await task

    print(result_coro)
    print(result_task)

asyncio.run(main())

gather vs wait vs TaskGroup 비교
import asyncio
from typing import Any

async def fetch(url: str, delay: float) -> str:
    await asyncio.sleep(delay)
    if "error" in url:
        raise ValueError(f"fetch 실패: {url}")
    return f"응답: {url}"

# ── 1. asyncio.gather ──────────────────────────────────────
async def use_gather():
    results = await asyncio.gather(
        fetch("api/users", 0.5),
        fetch("api/posts", 1.0),
        fetch("api/error", 0.3),
        return_exceptions=True  # 예외도 결과 리스트에 포함
    )
    for r in results:
        if isinstance(r, Exception):
            print(f"에러: {r}")
        else:
            print(r)

# ── 2. asyncio.wait ────────────────────────────────────────
async def use_wait():
    tasks = {
        asyncio.create_task(fetch("api/fast", 0.2)),
        asyncio.create_task(fetch("api/slow", 2.0)),
    }
    done, pending = await asyncio.wait(
        tasks,
        timeout=0.5,
        return_when=asyncio.FIRST_COMPLETED
    )
    for t in done:
        print(f"완료: {t.result()}")
    for t in pending:
        t.cancel()  # 타임아웃된 Task 직접 취소 필요
        print("타임아웃으로 취소됨")

# ── 3. TaskGroup (Python 3.11+, 권장) ──────────────────────
async def use_taskgroup():
    results: list[str] = []
    try:
        async with asyncio.TaskGroup() as tg:
            t1 = tg.create_task(fetch("api/users", 0.5))
            t2 = tg.create_task(fetch("api/posts", 1.0))
            # 블록 종료 시 모든 task 완료 대기
            # 예외 발생 시 나머지 task 자동 취소
        results = [t1.result(), t2.result()]
    except* ValueError as eg:  # ExceptionGroup 처리
        for e in eg.exceptions:
            print(f"TaskGroup 내 예외: {e}")
    return results

aiohttp — 동시 HTTP 요청 + 연결 풀 재사용
import asyncio
import aiohttp
from typing import Any

async def fetch_json(
    session: aiohttp.ClientSession,
    url: str
) -> dict[str, Any]:
    async with session.get(url, timeout=aiohttp.ClientTimeout(total=10)) as resp:
        resp.raise_for_status()
        return await resp.json()

async def fetch_all(urls: list[str]) -> list[dict]:
    # ClientSession은 한 번만 생성 — 커넥션 풀 재사용
    connector = aiohttp.TCPConnector(limit=50)  # 동시 연결 상한
    async with aiohttp.ClientSession(connector=connector) as session:
        tasks = [fetch_json(session, url) for url in urls]
        results = await asyncio.gather(*tasks, return_exceptions=True)

    successful = []
    for url, result in zip(urls, results):
        if isinstance(result, Exception):
            print(f"실패 [{url}]: {result}")
        else:
            successful.append(result)
    return successful

# 100개 URL 동시 요청
urls = [f"https://jsonplaceholder.typicode.com/posts/{i}" for i in range(1, 101)]
results = asyncio.run(fetch_all(urls))
print(f"성공: {len(results)}개")

asyncpg — 연결 풀 + 트랜잭션
import asyncio
import asyncpg

async def setup_pool() -> asyncpg.Pool:
    return await asyncpg.create_pool(
        dsn="postgresql://user:password@localhost:5432/mydb",
        min_size=5,
        max_size=20,
        command_timeout=60
    )

async def get_users(pool: asyncpg.Pool, limit: int = 100) -> list[dict]:
    async with pool.acquire() as conn:
        rows = await conn.fetch(
            "SELECT id, email, created_at FROM users ORDER BY created_at DESC LIMIT $1",
            limit
        )
        return [dict(row) for row in rows]

async def create_user_with_profile(
    pool: asyncpg.Pool,
    email: str,
    name: str
) -> int:
    async with pool.acquire() as conn:
        async with conn.transaction():  # 트랜잭션 — 실패 시 자동 롤백
            user_id = await conn.fetchval(
                "INSERT INTO users (email) VALUES ($1) RETURNING id",
                email
            )
            await conn.execute(
                "INSERT INTO profiles (user_id, name) VALUES ($1, $2)",
                user_id, name
            )
            return user_id

async def main():
    pool = await setup_pool()
    try:
        users = await get_users(pool, limit=10)
        print(f"유저 {len(users)}명 조회")

        # 여러 유저 동시 생성
        tasks = [
            create_user_with_profile(pool, f"user{i}@example.com", f"유저{i}")
            for i in range(10)
        ]
        ids = await asyncio.gather(*tasks)
        print(f"생성된 ID: {ids}")
    finally:
        await pool.close()

asyncio.run(main())

Semaphore — 동시 요청 수 제한
import asyncio
import aiohttp

async def fetch_with_limit(
    session: aiohttp.ClientSession,
    sem: asyncio.Semaphore,
    url: str
) -> str:
    async with sem:  # 세마포어 획득 — 동시 실행 수 제한
        async with session.get(url) as resp:
            return await resp.text()

async def crawl(urls: list[str], concurrency: int = 50) -> list[str]:
    sem = asyncio.Semaphore(concurrency)
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_with_limit(session, sem, url) for url in urls]
        return await asyncio.gather(*tasks, return_exceptions=True)

# ── Queue를 활용한 워커 풀 패턴 ────────────────────────────
async def producer(queue: asyncio.Queue, items: list[str]):
    for item in items:
        await queue.put(item)
    # sentinel 값으로 워커 종료 신호
    for _ in range(5):  # 워커 수만큼
        await queue.put(None)

async def worker(worker_id: int, queue: asyncio.Queue, results: list):
    while True:
        item = await queue.get()
        if item is None:  # 종료 신호
            queue.task_done()
            break
        try:
            # 실제 처리 로직
            await asyncio.sleep(0.1)
            results.append(f"worker{worker_id}: {item} 처리 완료")
        finally:
            queue.task_done()

async def process_with_worker_pool(items: list[str]) -> list[str]:
    queue: asyncio.Queue = asyncio.Queue(maxsize=100)
    results: list[str] = []

    workers = [asyncio.create_task(worker(i, queue, results)) for i in range(5)]
    await producer(queue, items)
    await asyncio.gather(*workers)
    return results

타임아웃, 블로킹 코드 오프로드, 에러 처리
import asyncio
import time
import requests  # 블로킹 라이브러리
from concurrent.futures import ThreadPoolExecutor

executor = ThreadPoolExecutor(max_workers=10)

# ── 타임아웃 처리 ──────────────────────────────────────────
async def with_timeout():
    # asyncio.wait_for: Python 3.8+
    try:
        result = await asyncio.wait_for(
            some_slow_coroutine(),
            timeout=5.0
        )
    except asyncio.TimeoutError:
        print("5초 초과, 취소됨")

    # asyncio.timeout: Python 3.11+ 권장
    try:
        async with asyncio.timeout(5.0):
            result = await some_slow_coroutine()
    except TimeoutError:
        print("타임아웃")

async def some_slow_coroutine():
    await asyncio.sleep(10)
    return "완료"

# ── 블로킹 코드 오프로드 ───────────────────────────────────
async def run_blocking_io(url: str) -> str:
    loop = asyncio.get_running_loop()
    # requests.get은 블로킹 — 스레드풀에서 실행
    response = await loop.run_in_executor(
        executor,
        lambda: requests.get(url, timeout=10)
    )
    return response.text

# CPU 바운드 작업도 동일하게 오프로드
async def run_cpu_intensive(data: list[int]) -> int:
    loop = asyncio.get_running_loop()
    return await loop.run_in_executor(executor, sum, data)

# ── Task 예외 처리 ─────────────────────────────────────────
async def may_fail():
    await asyncio.sleep(1)
    raise RuntimeError("예상치 못한 에러")

async def safe_task_usage():
    # 나쁜 예: Task 생성 후 결과를 await하지 않으면 예외가 사라짐
    # task = asyncio.create_task(may_fail())  # 예외 무시됨!

    # 좋은 예 1: 나중에 await
    task = asyncio.create_task(may_fail())
    try:
        await task
    except RuntimeError as e:
        print(f"Task 에러 처리: {e}")

    # 좋은 예 2: done callback으로 에러 로깅
    def handle_task_result(task: asyncio.Task):
        if not task.cancelled() and task.exception():
            print(f"[ERROR] Task 실패: {task.exception()}")

    task2 = asyncio.create_task(may_fail())
    task2.add_done_callback(handle_task_result)
    await asyncio.sleep(2)  # task2가 완료될 때까지 대기