# 파이썬 asyncio: async/await, TaskGroup, 스레드, multiprocessing과 GIL

URL: https://skillaria.top/ko/guides/python/python-asyncio

파이썬 asyncio는 async def로 만든 코루틴을 await에서 잠시 멈추게 하고, 그동안 이벤트 루프가 다른 작업을 실행하게 합니다. 덕분에 한 스레드에서도 수백 개의 네트워크 요청을 동시에 처리할 수 있습니다. 대기 작업에는 asyncio나 threading을, 무거운 계산에는 multiprocessing을 씁니다. CPython의 GIL 때문에 일반 스레드는 계산을 병렬로 처리하지 못하기 때문입니다.

파이썬 asyncio는 비동기 코드를 위한 표준 라이브러리입니다. `async def`로 정의한 함수는 `await`에서 잠시 멈출 수 있습니다. 한 작업이 네트워크 응답을 기다리는 동안 이벤트 루프가 다른 작업을 실행합니다. 그래서 한 스레드에서도 수백 개의 요청을 동시에 보낼 수 있습니다. 하지만 asyncio는 세 가지 도구 중 하나일 뿐입니다. 스레드(`threading`)와 프로세스(`multiprocessing`)도 있습니다. 이 글에서는 asyncio가 어떻게 동작하는지, 어떤 작업에 무엇을 고를지, 그리고 GIL이 어떤 역할을 하는지 살펴봅니다.

## 기다리는 일과 계산하는 일은 다릅니다

이 차이를 이해하지 않으면 나머지가 모두 헷갈립니다. 프로그램이 느린 이유는 두 가지 중 하나입니다.

- **I/O 바운드** — 입출력을 기다립니다. API 응답, 데이터베이스, 디스크 읽기가 여기에 해당하며, 이때 프로세서는 놀고 있습니다.
- **CPU 바운드** — 계산합니다. 이미지 처리나 수백만 건의 반복 계산이 해당하며, 프로세서가 바쁩니다.

기다리는 작업에는 asyncio나 스레드가 맞습니다. 하나가 기다리는 동안 다른 작업을 실행하면 됩니다. 계산 작업은 여러 코어에서 실제로 동시에 돌아야 하므로 프로세스가 필요합니다. 이 둘을 헷갈리는 것이 가장 흔한 실수입니다. 순수 파이썬에서 스레드를 늘려도 계산은 빨라지지 않습니다. 반대로 HTTP 요청 세 개를 보내려고 프로세스를 여러 개 띄우는 것은 과합니다.

## asyncio란: 코루틴과 이벤트 루프

비동기 코드는 두 단어로 이루어집니다. `async def`는 코루틴을 선언합니다. 중간에 멈출 수 있는 함수입니다. 호출하면 본문이 실행되지 않고 코루틴 객체가 만들어집니다. 본문 안의 `await`는 '여기서 기다릴 테니, 그동안 다른 작업을 진행하라'는 뜻입니다. 전체를 관리하는 것이 이벤트 루프이고, `asyncio.run()`으로 시작합니다.

```
import asyncio

async def fetch_rate(currency):
    await asyncio.sleep(1)          # API 요청을 흉내 낸 대기
    return {"USD": 92.5, "EUR": 100.1, "CNY": 12.7}[currency]

async def main():
    rate = await fetch_rate("USD")
    print(rate)

asyncio.run(main())  # 92.5
```

`asyncio.sleep()`은 비동기 대기입니다. 이벤트 루프에 제어권을 넘깁니다. 코루틴 안에서 일반 `time.sleep()`을 쓰면 루프 전체가 멈추고, 다른 작업도 모두 기다리게 됩니다.

### 순서대로 또는 동시에

`await`를 반복문 안에서 쓰면 요청이 하나씩 차례로 나갑니다. 통화 세 개를 조회하면 3초가 걸립니다. 동시에 보내려면 작업으로 등록해야 합니다.

```
import asyncio, time

async def main():
    start = time.perf_counter()
    for cur in ("USD", "EUR", "CNY"):
        await fetch_rate(cur)                       # 차례대로
    print(f"{time.perf_counter() - start:.1f}초")   # 3.0초

asyncio.run(main())
```

## TaskGroup: 함께 시작하고 모두 기다리기

현대적인 방법은 Python 3.11부터 쓸 수 있는 작업 그룹입니다. `async with` 블록 안에서 `tg.create_task()`를 호출하면 작업이 바로 시작됩니다. 블록을 빠져나올 때는 모든 작업이 끝나 있습니다.

```
async def main():
    start = time.perf_counter()
    async with asyncio.TaskGroup() as tg:
        tasks = {cur: tg.create_task(fetch_rate(cur)) for cur in ("USD", "EUR", "CNY")}
    rates = {cur: t.result() for cur, t in tasks.items()}
    print(rates, f"{time.perf_counter() - start:.1f}초")   # ... 1.0초

asyncio.run(main())
```

TaskGroup은 엄격하게 동작합니다. 작업 하나가 실패하면 나머지는 취소되고, 모든 오류는 `ExceptionGroup`으로 묶입니다. 이를 `except*`로 받습니다.

```
async def main():
    try:
        async with asyncio.TaskGroup() as tg:
            tg.create_task(fetch_rate("USD"))
            tg.create_task(fetch_rate("XXX"))   # KeyError
    except* KeyError as eg:
        print("알 수 없는 통화:", eg.exceptions)
```

### asyncio.gather와 create_task

TaskGroup이 나오기 전의 표준은 `asyncio.gather()`였습니다. 코루틴을 동시에 실행하고, 결과를 입력 순서대로 돌려줍니다. 기존 코드에는 아직 많습니다.

```
async def main():
    usd, eur, cny = await asyncio.gather(
        fetch_rate("USD"), fetch_rate("EUR"), fetch_rate("CNY")
    )
    # return_exceptions=True를 주면 오류도 결과 목록에 담기고 전체가 멈추지 않습니다
```

`asyncio.create_task()`는 그룹 없이 코루틴을 백그라운드에서 실행합니다. 작업 객체의 참조를 반드시 보관해야 합니다. 그렇지 않으면 가비지 컬렉터가 작업을 지울 수 있고, 오류도 아무도 보지 못합니다. 그래서 새 프로젝트에서는 TaskGroup이 더 안전합니다.

### 작업 취소와 시간 제한

```
async def main():
    task = asyncio.create_task(fetch_rate("USD"))
    task.cancel()                     # 작업 내부에서 CancelledError 발생
    try:
        await task
    except asyncio.CancelledError:
        print("취소됨")

    try:
        async with asyncio.timeout(0.5):   # Python 3.11 이상
            await fetch_rate("EUR")
    except TimeoutError:
        print("시간 초과")
```

이전 버전에서는 `asyncio.timeout()` 대신 `asyncio.wait_for(coro, timeout=0.5)`를 씁니다. `asyncio.wait()`도 있습니다. 완료된 작업과 미완료 작업의 집합을 돌려주고, `return_when=asyncio.FIRST_COMPLETED`로 가장 먼저 끝나는 작업을 기다릴 수 있습니다.

강의에서 소개하듯, Python 3.15의 TaskGroup에는 `tg.cancel()` 메서드가 생겼습니다. 그룹의 모든 작업을 한 번에 취소합니다. '누가 먼저 응답하나' 패턴이 쉬워집니다. 미러 서버 세 곳에 물어보고 가장 먼저 온 응답을 쓰고, 나머지는 취소하는 식입니다. 이전에는 별도의 신호용 예외를 던지거나 작업을 하나씩 취소해야 했습니다.

## 동기화: Semaphore, Lock, Event, Queue

asyncio에는 `await`와 함께 동작하는 자체 동기화 도구가 있습니다. 가장 유용한 것은 `Semaphore`입니다. 동시에 실행되는 작업 수를 제한하므로, 남의 API에 수천 개의 요청을 한꺼번에 보내는 사고를 막을 수 있습니다.

```
sem = asyncio.Semaphore(10)

async def limited(url):
    async with sem:              # 동시에 최대 10개
        await asyncio.sleep(0.1) # 실제라면 요청이 들어갈 자리
        return url
```

- `asyncio.Lock` — 임계 구역에 한 번에 하나의 코루틴만 들어갑니다.
- `asyncio.Event` — 한 코루틴은 `await event.wait()`로 기다리고, 다른 코루틴은 `event.set()`으로 신호를 보냅니다.
- `asyncio.Queue` — 생산자와 소비자를 잇는 큐입니다. `await q.put()`, `await q.get()`, `q.task_done()`을 씁니다.

## 함정: await를 빼먹음

가장 흔한 실수는 `await` 없이 코루틴을 호출하는 것입니다.

```
async def main():
    rate = fetch_rate("USD")    # await를 빠뜨림
    print(rate * 2)             # TypeError
# RuntimeWarning: coroutine 'fetch_rate' was never awaited
```

이 경우 파이썬은 함수를 실행하지 않고, 코루틴 객체만 만들었습니다. 로그에 'was never awaited'가 보이면 빠진 `await`를 찾으세요. 두 번째로 흔한 실수는 코루틴 안에서 블로킹 호출(`time.sleep`, `requests.get`)을 하는 것입니다. HTTP에는 httpx나 aiohttp 같은 비동기 라이브러리를 쓰고, 어쩔 수 없이 동기 코드를 써야 한다면 `await asyncio.to_thread(함수, 인자)`로 스레드에 넘기세요.

## 파이썬 멀티스레딩: threading 모듈

스레드는 하나의 프로세스 안에서 독립적으로 실행되는 흐름입니다. 스레드들은 메모리를 공유합니다. 네트워크 대기에는 잘 맞고, 코드를 async로 바꿀 필요도 없습니다.

```
import threading, time

def download(name):
    time.sleep(1)               # 네트워크 대기
    print(name, "완료")

threads = [threading.Thread(target=download, args=(n,)) for n in ("a", "b", "c")]
for t in threads:
    t.start()
for t in threads:
    t.join()                    # 모두 약 1초 안에 끝납니다
```

메모리를 공유한다는 것은 장점이자 위험입니다. 두 스레드가 같은 변수를 바꾸면 `threading.Lock`이 필요합니다. 대부분의 경우 스레드를 직접 다루기보다 풀을 쓰는 편이 낫습니다.

### ThreadPoolExecutor

`concurrent.futures.ThreadPoolExecutor`는 스레드 풀을 만들고 작업을 나눠 줍니다. `map`은 결과를 원래 순서대로 돌려줍니다.

```
from concurrent.futures import ThreadPoolExecutor
import time

def fetch(url):
    time.sleep(1)
    return f"{url}: 200"

urls = [f"https://example.com/{i}" for i in range(5)]
with ThreadPoolExecutor(max_workers=5) as pool:
    for result in pool.map(fetch, urls):
        print(result)           # 전부 약 1초
```

## GIL이란, 그리고 스레드가 계산을 빠르게 하지 못하는 이유

GIL(Global Interpreter Lock, 전역 인터프리터 락)은 CPython의 전역 잠금입니다. 한 순간에 파이썬 바이트코드를 실행하는 스레드는 하나뿐입니다. 스레드가 입출력을 기다릴 때는 GIL을 놓으므로, 네트워크 작업에서는 스레드가 효과를 냅니다. 하지만 순수 계산에서는 스레드 네 개가 한 개보다 빠르지 않습니다. 잠금을 번갈아 넘겨받을 뿐입니다.

```
from concurrent.futures import ThreadPoolExecutor

def heavy(n):
    return sum(i * i for i in range(n))

with ThreadPoolExecutor(4) as pool:
    list(pool.map(heavy, [5_000_000] * 4))   # 순차 실행과 거의 같은 시간
```

GIL을 놓는 C 기반 라이브러리(NumPy 등)는 무거운 연산 중에 잠금을 풀기 때문에, 이런 라이브러리에서는 스레드가 실제로 빨라집니다.

## multiprocessing: 진짜 병렬 처리

각 프로세스는 자신의 인터프리터와 GIL을 가집니다. 그래서 여러 코어에서 동시에 계산할 수 있습니다. 대신 메모리는 따로 쓰고, 프로세스 간 데이터는 직렬화(pickle)로 전달하며, 프로세스 시작은 스레드보다 비쌉니다.

```
from concurrent.futures import ProcessPoolExecutor

def heavy(n):
    return sum(i * i for i in range(n))

if __name__ == "__main__":      # Windows와 macOS에서는 필수
    with ProcessPoolExecutor() as pool:
        print(list(pool.map(heavy, [5_000_000] * 4)))
```

`if __name__ == "__main__"` 검사가 없으면 자식 프로세스가 모듈을 임포트하면서 또 프로세스를 만들려고 합니다. 저수준 모듈 `multiprocessing`에는 `Process`, `Pool`, `Queue`가 있지만, `ProcessPoolExecutor`는 같은 인터페이스를 가진 스레드 풀과 똑같이 쓸 수 있어서 더 간단합니다.

## 무엇을 고를까: asyncio, threading, multiprocessing

| 작업 | 도구 | 이유 |
| --- | --- | --- |

| 200개 페이지 내려받기, 많은 API 요청 | asyncio (httpx/aiohttp) 또는 ThreadPoolExecutor | 네트워크 대기, 프로세서는 한가함 |

| 이미 async인 프로그램 안의 여러 요청 | TaskGroup | 코드가 이미 비동기 |

| async 버전이 없는 동기 라이브러리 | 스레드, `asyncio.to_thread` | 코드를 다시 쓸 필요가 없음 |

| 순수 파이썬으로 하는 무거운 계산 | ProcessPoolExecutor | GIL 우회, 모든 코어 사용 |

## GIL 없는 파이썬과 서브인터프리터: 새로운 변화

Python 3.13부터 GIL이 없는 빌드(free-threading, PEP 703)가 있습니다. 이름에 t가 붙습니다. 예를 들어 `python3.15t`입니다. 3.13에서는 실험적이었고, 3.14부터는 공식 지원됩니다(PEP 779). 이 빌드에서는 스레드가 여러 코어에서 실제로 병렬 계산합니다. 하지만 기본 설치는 여전히 GIL이 있는 파이썬입니다. 자유 스레드 빌드에서는 단일 스레드 코드가 조금 느릴 수 있고, 모든 라이브러리가 아직 지원하지는 않습니다.

```
import sys
print(sys._is_gil_enabled())   # 자유 스레드 빌드에서는 False
```

병렬 계산을 위한 또 다른 길은 Python 3.14(PEP 734)에서 추가된 서브인터프리터입니다. `concurrent.interpreters` 모듈은 한 프로세스 안에서 격리된 인터프리터를 여러 개 만들며, 각각 자기 GIL을 가집니다. `InterpreterPoolExecutor`를 쓰면 스레드 풀과 똑같은 방식으로 사용할 수 있습니다. 과정 강의의 측정에서는 무거운 작업 네 개가 1개 스레드에서 1.03초, 이 방식으로는 0.39초가 걸렸습니다.

## 정리

파이썬 asyncio는 기다리는 작업을 위한 도구입니다. `async def`로 코루틴을 만들고, `await`에서 멈추게 하며, TaskGroup이나 `gather`로 동시에 실행합니다. 스레드는 코드를 async로 바꾸지 않고 같은 일을 할 수 있게 해 줍니다. 무거운 계산에는 프로세스가 필요합니다. GIL 때문에 일반 CPython의 스레드는 계산을 병렬로 처리하지 못하기 때문입니다. 자유 스레드 빌드와 서브인터프리터가 이 그림을 조금씩 바꾸고 있지만, '기다리면 asyncio나 스레드, 계산하면 프로세스'라는 기본 규칙은 아직 유효합니다. 이 주제는 과정 강의에서 실습합니다. 트래커가 환율을 병렬로 받아 1초 만에 끝내도록 만듭니다.

## FAQ

**비동기 함수를 일반 코드에서 호출하려면 어떻게 하나요?**

asyncio.run(코루틴())을 쓰세요. 이벤트 루프를 만들어 코루틴을 실행하고 루프를 닫습니다. 이미 실행 중인 루프 안에서는 asyncio.run을 부를 수 없고, await를 써야 합니다.

**asyncio는 여러 스레드에서 동작하나요?**

아닙니다. 이벤트 루프는 모든 코루틴을 한 스레드에서 실행하고, await 지점에서만 전환합니다. 그래서 데이터 경합이 스레드보다 적지만, 블로킹 호출 하나가 모든 작업을 멈춥니다.

**asyncio를 pip로 따로 설치해야 하나요?**

아닙니다. asyncio는 Python 3 표준 라이브러리에 포함되어 있으므로 import asyncio만 쓰면 됩니다. PyPI의 asyncio 패키지는 오래되었으니 설치할 필요가 없습니다.

**무거운 함수를 asyncio에서 실행하면서 루프를 막지 않으려면?**

계산에는 loop.run_in_executor와 ProcessPoolExecutor를, 블로킹 입출력에는 asyncio.to_thread를 쓰세요. 그러면 이벤트 루프가 다른 작업을 계속 처리합니다.

**코루틴과 태스크(Task)는 어떻게 다른가요?**

코루틴은 누군가 await하기 전까지 아무 일도 하지 않는 객체입니다. 태스크는 코루틴을 감싸서 이벤트 루프의 스케줄에 바로 등록하므로, 다른 코드와 동시에 실행됩니다.


---
Source: Skillaria.Top (https://skillaria.top)