asyncio vs threading? 3가지 질문으로 정답을 고르자

1. 개요

Python 동시성을 배우는 많은 사람이 “asyncio가 좋을까, threading이 좋을까?“라고 묻습니다 — 이 질문 자체가 처음부터 잘못되었습니다. 두 도구는 서로 다른 문제를 해결하기 때문입니다. 이 글에서는 3가지 구체적인 질문으로 직접 답하도록 돕겠습니다. 실행 가능한 코드 예시도 함께 제공합니다. 이 글을 읽고 나면 threading을 언제 써야 하는지, asyncio를 언제 써야 하는지, 둘 다 소용없어서 multiprocessing을 써야 하는지를 정확히 알 수 있을 겁니다.

2. “어느 게 빠를까”라는 질문이 잘못된 이유

Python의 Global Interpreter Lock(GIL) 때문에 오직 한 개의 Python thread만 한 번에 bytecode를 실행합니다. 머신에 CPU 코어가 몇 개나 있든 상관없습니다. threading과 asyncio 모두 GIL을 깨뜨릴 수 없습니다. 둘 다 I/O 대기(파일 읽기, 네트워크 호출, DB 쿼리) 중에 CPU가 “정체”되지 않도록 할 뿐입니다 — 계산을 더 빠르게 해주지는 않습니다.

import time

def cpu_bound_task():
    total = 0
    for i in range(10_000_000):
        total += i * i
    return total

start = time.perf_counter()
cpu_bound_task()
cpu_bound_task()
print(f"Sequential: {time.perf_counter() - start:.2f}s")

Sequential: 1.85s

cpu_bound_task()를 threading.Thread나 asyncio.gather로 두 번 돌려도 총 소요 시간은 감소하지 않습니다. GIL이 한 번에 하나의 thread만 Python code를 실행하게 하기 때문입니다. 아래에 있는 첫 번째 질문이 가장 중요한 이유가 바로 이것입니다.

주의: 작업이 CPU-bound(무거운 산술 계산, 이미지 처리, 암호화 등)라면 threading과 asyncio 모두 도움이 되지 않습니다. multiprocessing을 사용하거나 무거운 부분을 C/Rust 확장으로 작성해야 합니다(NumPy, Cython 등).

3. 질문 1: 당신의 작업이 I/O-bound인가 CPU-bound인가?

이 질문이 “대기형” 동시성(I/O-bound)을 사용해야 하는지를 결정합니다.

3.1. I/O-bound: threading과 asyncio 모두 효율적

import time
import requests
from concurrent.futures import ThreadPoolExecutor

URLS = ["https://httpbin.org/delay/1"] * 5

def fetch(url: str) -> int:
    return requests.get(url).status_code

start = time.perf_counter()
with ThreadPoolExecutor(max_workers=5) as pool:
    results = list(pool.map(fetch, URLS))
print(f"Threading: {time.perf_counter() - start:.2f}s, results={results}")

Threading: 1.12s, results=[200, 200, 200, 200, 200]

ThreadPoolExecutor는 5개의 OS thread로 5개의 요청을 동시에 실행합니다. requests.get은 네트워크를 기다리는 blocking call이므로, 이 대기 시간에 GIL이 C 레이어에서 해제됩니다 — 5개의 요청이 실제로 병렬로 실행되어 총 소요 시간이 5초 대신 약 1초입니다.

import asyncio
import aiohttp
import time

URLS = ["https://httpbin.org/delay/1"] * 5

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

async def main():
    async with aiohttp.ClientSession() as session:
        tasks = [fetch(session, url) for url in URLS]
        return await asyncio.gather(*tasks)

start = time.perf_counter()
results = asyncio.run(main())
print(f"Asyncio: {time.perf_counter() - start:.2f}s, results={results}")

Asyncio: 1.08s, results=[200, 200, 200, 200, 200]

결과는 threading과 유사하지만 메커니즘이 다릅니다: 단 하나의 thread만 사용하고, event loop가 I/O 대기 시 주도적으로 await하며 제어를 양보합니다. 새로운 OS thread를 만들 필요가 없습니다.

3.2. CPU-bound: 둘 다 multiprocessing에 밀림

from concurrent.futures import ProcessPoolExecutor

def cpu_bound_task():
    return sum(i * i for i in range(10_000_000))

start = time.perf_counter()
with ProcessPoolExecutor(max_workers=2) as pool:
    results = list(pool.map(cpu_bound_task, [None, None]))
print(f"Multiprocessing: {time.perf_counter() - start:.2f}s")

Multiprocessing: 1.02s

ProcessPoolExecutor는 각각 고유한 GIL을 가진 별도의 프로세스를 생성하므로 실제 여러 CPU 코어를 활용할 수 있습니다. CPU-bound 작업에서 “threading인가 asyncio인가”를 고민 중이라면 정답은 둘 다 틀렸습니다 — multiprocessing을 사용하세요.

4. 질문 2: 당신이 사용하는 에코시스템이 async를 지원하는가?

이것이 실제 코드베이스에서 작업할 때 가장 실용적인 질문입니다. 모든 걸 처음부터 작성하는 게 아니기 때문입니다.

import psycopg2  # 동기 드라이버, blocking

def get_user(user_id: int) -> dict:
    conn = psycopg2.connect("dbname=app")
    cur = conn.cursor()
    cur.execute("SELECT id, name FROM users WHERE id = %s", (user_id,))
    row = cur.fetchone()
    conn.close()
    return {"id": row[0], "name": row[1]}

psycopg2는 blocking 라이브러리입니다 — await 가능한 버전이 없습니다. 이 드라이버로 여러 쿼리를 병렬로 실행하려면 threading(또는 asyncio의 run_in_executor)이 자연스러운 선택입니다. 다시 작성할 필요가 없기 때문입니다.

import asyncpg  # async-native 드라이버

async def get_user(conn: asyncpg.Connection, user_id: int) -> dict:
    row = await conn.fetchrow(
        "SELECT id, name FROM users WHERE id = $1", user_id
    )
    return {"id": row["id"], "name": row["name"]}

에코시스템이 이미 async-native 드라이버(asyncpg, aiohttp, motor for MongoDB 등)를 제공한다면, asyncio는 최대 장점을 발휘합니다: 하나의 thread에서 수천 개의 동시 연결을 처리하면서 OS thread 생성 비용을 피합니다.

주의: blocking code를 coroutine에 제대로 await 없이 섞으면(예: coroutine 내에서 requests.get을 직접 호출) event loop 전체가 차단됩니다 — 관련 없는 다른 task들도 기다려야 합니다. 이것이 threading에서 asyncio로 전환할 때 가장 흔한 실수입니다.

5. 질문 3: 동시에 처리해야 할 task는 몇 개인가?

thread와 coroutine 사이의 리소스 비용 차이는 task 수가 증가하면서 매우 커집니다.

import threading
import resource

def noop():
    threading.Event().wait(2)

threads = [threading.Thread(target=noop) for _ in range(2000)]
for t in threads:
    t.start()

peak_kb = resource.getrusage(resource.RUSAGE_SELF).ru_maxrss
print(f"Peak RSS with 2000 threads: {peak_kb / 1024:.1f} MB")

for t in threads:
    t.join()

Peak RSS with 2000 threads: 1850.3 MB

각 OS thread는 기본적으로 수백 KB에서 수 MB의 stack을 차지하므로, 2000개 thread는 쉽게 2GB 이상의 RAM을 사용하고 OS scheduler에 큰 context-switching 부담을 줍니다.

import asyncio
import resource

async def noop():
    await asyncio.sleep(2)

async def main():
    tasks = [asyncio.create_task(noop()) for _ in range(2000)]
    await asyncio.gather(*tasks)
    peak_kb = resource.getrusage(resource.RUSAGE_SELF).ru_maxrss
    print(f"Peak RSS with 2000 coroutines: {peak_kb / 1024:.1f} MB")

asyncio.run(main())

Peak RSS with 2000 coroutines: 95.7 MB

Coroutine은 순수 Python 객체이며 별도의 OS thread가 필요 없습니다 — 2000개 task를 만드는 비용은 단지 수십 MB입니다. 수천 개의 동시 연결을 처리해야 한다면(웹 서버, 채팅 서버, 대규모 크롤러), asyncio는 거의 필수 선택입니다. threading은 대역폭이 남기 전에 리소스가 떨어질 겁니다.

6. 빠른 결정 테이블

질문	답변 → 선택
1. I/O-bound인가 CPU-bound인가?	CPU-bound → multiprocessing. I/O-bound → 질문 2로 넘어가기
2. 당신이 사용하는 라이브러리가 async-native인가?	아니오 (구형 blocking 드라이버) → threading. 네 (aiohttp, asyncpg 등) → asyncio
3. 동시 task가 몇 개 필요한가?	수십 개 → threading으로 충분히 간단. 수백/수천 개 → asyncio

실제로는 많은 시스템이 둘 다 사용합니다: 대량의 네트워크 I/O-bound에는 asyncio를 쓰고, blocking code(구형 라이브러리 등)는 run_in_executor로 별도의 thread pool에서 실행해 event loop을 차단하지 않습니다.

import asyncio

def legacy_blocking_call(x: int) -> int:
    import time
    time.sleep(1)  # 구형 라이브러리 시뮬레이션
    return x * 2

async def main():
    loop = asyncio.get_running_loop()
    result = await loop.run_in_executor(None, legacy_blocking_call, 21)
    print(result)

asyncio.run(main())

42

run_in_executor는 blocking 함수를 기본 thread pool의 thread로 보내고, event loop은 그동안 다른 coroutine들을 처리할 수 있습니다.

7. pytest로 동작 검증하기

asyncio.gather가 task들을 실제로 병렬로 실행하는지(순차적이 아니라) 확인하는 테스트 예시:

import asyncio
import time
import pytest

async def sleepy_task(seconds: float) -> float:
    await asyncio.sleep(seconds)
    return seconds

@pytest.mark.asyncio
async def test_gather_runs_concurrently():
    start = time.perf_counter()
    results = await asyncio.gather(
        sleepy_task(0.5), sleepy_task(0.5), sleepy_task(0.5)
    )
    elapsed = time.perf_counter() - start

    assert results == [0.5, 0.5, 0.5]
    # 순차 실행이면 ~1.5초, 병렬 실행이면 ~0.5초
    assert elapsed < 0.7

필요한 의존성:

pytest
pytest-asyncio

이 테스트는 asyncio의 본질을 확인합니다: 세 개의 sleep(0.5) task가 하나의 thread에서 동시에 실행되어 1.5초가 아닌 약 0.5초 걸립니다 — elapsed < 0.7 assertion이 실패하면 코드가 순차적으로 await하고 있다는 뜻입니다.

8. 흔한 실수들

• Coroutine에서 time.sleep() 사용: await asyncio.sleep()을 써야 합니다. time.sleep()은 blocking call이며 event loop 전체를 정지시킵니다.
• await 빠뜨리기: fetch(session, url)을 await 없이 호출하면 coroutine object를 만들 뿐 실행되지 않습니다 — Python이 RuntimeWarning: coroutine was never awaited 경고를 표시합니다.
• Thread 간 state 공유 시 lock 없이: threading은 실제 race condition이 있습니다(여러 thread이 동일 변수에 쓰기), threading.Lock이 필요합니다. asyncio는 단일 thread이므로 이 종류의 race condition은 없지만, await 중간에 발생할 수 있는 로직 버그는 여전히 가능합니다.
• CPU-bound 작업에 ThreadPoolExecutor 사용 후 실망: GIL이 여전히 차단하므로 더 빨라지지 않습니다. ProcessPoolExecutor를 사용하세요.

주의: 팀이 threading에 익숙하고 코드베이스가 blocking 라이브러리로 가득하다면, “더 새로워 보인다”는 이유만으로 asyncio로 리팩토링하지 마세요. 드라이버를 전부 다시 작성하고, exception 처리를 다르게 하고, coroutine stack trace를 디버깅하는 비용이 동시 task 수가 많지 않은 경우 얻을 수 있는 이점보다 클 수 있습니다.

9. 결론

asyncio와 threading 중 선택하는 건 “뭐가 더 빨까”가 아니라 3가지 구체적인 질문입니다: 작업이 I/O-bound인가 CPU-bound인가, 라이브러리 에코시스템이 async-native인가, 동시성 규모가 얼마나 큰가 하는 것입니다. CPU-bound는 항상 multiprocessing이 필요합니다. I/O-bound이면서 구형 라이브러리와 작은 동시성 규모라면 threading이 간단하고 통합하기 쉽습니다. I/O-bound이면서 큰 규모이고 async-native 라이브러리가 있다면 asyncio가 성능과 리소스 측면에서 올바른 선택입니다.