← 파이썬 강의 목록으로
🚀
고급 (Advanced)
GIL · ThreadPoolExecutor · ProcessPoolExecutor · Lock · async/await · gather · Queue · timeout · pytest · parametrize · fixture · mock

4~6주차 — 동시성 스레딩과 멀티프로세싱 · 동시성 asyncio · 테스트

I/O 바운드 작업은 ThreadPoolExecutor 로, CPU 바운드는 ProcessPoolExecutor 로 — GIL 의 의미와 동시성 도구의 선택 기준을 익힙니다. async/await 문법, asyncio.gather 로 동시 실행, Queue 기반 producer/consumer, 그리고 wait_for 타임아웃까지 — 비동기 프로그래밍의 핵심을 익힙니다. pytest 기본 사용법, parametrize 로 다양한 케이스, fixture 로 공통 setup, 그리고 mock 으로 외부 의존성을 격리하는 방법까지 다룹니다.

threadingmultiprocessingGILconcurrent.futuresasyncioasync/awaitgatherqueuepytestfixtureparametrizemock
소요 시간
⏱ 2시간 + 2시간 + 2시간
난이도
📊 고급
선수 조건
🎯 고급 3주차
결과물
I/O·CPU 작업에 맞는 동시성 도구 선택 / 비동기 동시 실행과 Queue 패턴 / pytest 로 자동화된 테스트 작성

이 강의에서 배우는 것

  • 1I/O 바운드 vs CPU 바운드 작업을 구분한다
  • 2ThreadPoolExecutor 로 I/O 작업을 가속한다
  • 3ProcessPoolExecutor 로 CPU 작업을 가속한다
  • 4threading.Lock 으로 동기화한다
  • 5async/await 문법을 사용한다
  • 6asyncio.run, asyncio.gather 로 동시 실행한다
  • 7비동기 HTTP 호출(httpx, aiohttp)을 한다
  • 8producer/consumer 패턴을 구현한다
  • 9pytest 기본 사용법을 익힌다
  • 10parametrize 로 다양한 케이스를 테스트한다
  • 11fixture 로 공통 setup 을 제공한다
  • 12mock 으로 외부 의존성을 격리한다

4주차 — 동시성 (1) 스레딩과 멀티프로세싱

I/O 바운드 작업은 ThreadPoolExecutor 로, CPU 바운드는 ProcessPoolExecutor 로 — GIL 의 의미와 동시성 도구의 선택 기준을 익힙니다.

1. GIL (Global Interpreter Lock)

CPython은 한 번에 하나의 스레드만 파이썬 바이트코드를 실행 → CPU 바운드는 스레딩으로 가속 X. I/O 작업(네트워크, 디스크) 에서는 효과 있음.

작업 종류도구
I/O 바운드 (네트워크, 파일)threading, asyncio
CPU 바운드 (계산, 인코딩)multiprocessing

2. ThreadPoolExecutor

python
from concurrent.futures import ThreadPoolExecutor
import requests

urls = ["https://example.com"] * 10

def fetch(url):
    return len(requests.get(url).text)

with ThreadPoolExecutor(max_workers=5) as ex:
    results = list(ex.map(fetch, urls))

순차 실행 대비 5배 정도 빨라집니다(네트워크 대기 동안 다른 작업).

3. ProcessPoolExecutor

python
from concurrent.futures import ProcessPoolExecutor

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

if __name__ == "__main__":   # Windows는 필수
    with ProcessPoolExecutor() as ex:
        print(list(ex.map(heavy, [1_000_000] * 4)))

4. submit / as_completed

python
from concurrent.futures import ThreadPoolExecutor, as_completed

with ThreadPoolExecutor() as ex:
    futures = [ex.submit(fetch, url) for url in urls]
    for fut in as_completed(futures):
        print(fut.result())

5. threading.Lock

여러 스레드가 같은 자원을 변경할 때 데이터 깨짐 방지.

python
import threading

counter = 0
lock = threading.Lock()

def inc():
    global counter
    for _ in range(100_000):
        with lock:
            counter += 1

ts = [threading.Thread(target=inc) for _ in range(4)]
for t in ts: t.start()
for t in ts: t.join()
print(counter)   # 400000

Lock 없으면 결과가 작거나 가변적입니다.

자주 하는 실수

  1. CPU 작업에 thread 사용 — GIL 때문에 느려질 수도
  2. __main__ 가드 누락(Windows + multiprocessing) — 무한 루프
  3. 공유 자원에 Lock 누락 — 결과 비결정적
  4. Future 결과를 안 받음 — 예외가 묻힘

FAQ

Q1. asyncio 와 비교? — asyncio는 단일 스레드 협력적. threading은 OS 스케줄링. I/O 가벼우면 asyncio.

Q2. multiprocessing 단점? — 프로세스 시작 비용. 작은 작업은 오히려 느림. 객체 직렬화(pickle) 필요.

5주차 — 동시성 (2) asyncio

async/await 문법, asyncio.gather 로 동시 실행, Queue 기반 producer/consumer, 그리고 wait_for 타임아웃까지 — 비동기 프로그래밍의 핵심을 익힙니다.

1. 코루틴

async def 로 정의된 함수는 코루틴. 호출해도 즉시 실행 안 되고 객체 반환.

python
import asyncio

async def hello():
    print("hello")
    await asyncio.sleep(1)
    print("world")

asyncio.run(hello())

2. asyncio.gather

python
async def task(name, delay):
    await asyncio.sleep(delay)
    print(f"{name} 완료")

async def main():
    await asyncio.gather(
        task("A", 1),
        task("B", 2),
        task("C", 1),
    )

asyncio.run(main())
# 약 2초 소요 (순차면 4초)

3. 비동기 HTTP

python
import asyncio
import httpx

async def fetch(client, url):
    r = await client.get(url)
    return len(r.text)

async def main():
    async with httpx.AsyncClient() as client:
        tasks = [fetch(client, "https://example.com") for _ in range(10)]
        results = await asyncio.gather(*tasks)
    print(results)

asyncio.run(main())

requests 는 동기, httpx 와 aiohttp 가 비동기.

4. producer / consumer

python
async def producer(q):
    for i in range(5):
        await q.put(i)
        print(f"생산: {i}")
        await asyncio.sleep(0.1)
    await q.put(None)

async def consumer(q):
    while True:
        item = await q.get()
        if item is None:
            break
        print(f"소비: {item}")

async def main():
    q = asyncio.Queue()
    await asyncio.gather(producer(q), consumer(q))

5. 타임아웃

python
try:
    result = await asyncio.wait_for(slow_task(), timeout=1.0)
except asyncio.TimeoutError:
    print("시간 초과")

6. 동기 vs 비동기

스레딩asyncio
단위OS 스레드코루틴
컨텍스트 스위칭OS 결정await 에서만
메모리무거움가벼움
라이브러리대부분 호환async 지원 필요

자주 하는 실수

  1. async def 결과를 await 안 함 — 코루틴 객체만 반환
  2. 동기 함수 안에서 await — async def 안에서만
  3. 블로킹 함수 호출 — time.sleep 대신 await asyncio.sleep
  4. asyncio.run 중첩 — Jupyter 등에서는 await main() 사용

FAQ

Q1. requests 를 asyncio 안에서? — 동작은 함. 하지만 블로킹이라 동시성 효과 없음. httpx 사용.

Q2. gather vs wait? — gather 는 결과를 순서대로, 실패 시 전파. wait 는 더 유연.

Q3. async 디버깅? — traceback 이 다소 복잡. 디버거 도움.

6주차 — 테스트 (unittest, pytest)

pytest 기본 사용법, parametrize 로 다양한 케이스, fixture 로 공통 setup, 그리고 mock 으로 외부 의존성을 격리하는 방법까지 다룹니다.

1. 왜 테스트?

  • 회귀 방지 — 코드 변경 시 동작 보장
  • 명세 역할 — 의도 문서화
  • 리팩토링 안전망

2. pytest 기본

bash
pip install pytest

테스트 파일은 test_*.py 또는 *_test.py 입니다.

python
# test_calc.py
def add(a, b): return a + b

def test_add():
    assert add(2, 3) == 5
    assert add(-1, 1) == 0
bash
pytest                  # 전체 실행
pytest test_calc.py     # 한 파일
pytest -v               # verbose
pytest -k "add"         # 이름 매칭

3. parametrize

같은 테스트를 여러 입력으로.

python
import pytest

@pytest.mark.parametrize("a,b,expected", [
    (1, 2, 3),
    (-1, 1, 0),
    (0, 0, 0),
    (100, 200, 300),
])
def test_add(a, b, expected):
    assert add(a, b) == expected

4. fixture

python
@pytest.fixture
def sample_users():
    return [{"name": "Alice"}, {"name": "Bob"}]

def test_first(sample_users):
    assert sample_users[0]["name"] == "Alice"

def test_count(sample_users):
    assert len(sample_users) == 2

스코프(session, module, function):

python
@pytest.fixture(scope="module")
def expensive():
    print("한 번만 setup")
    return ...

5. 예외 검증

python
def divide(a, b):
    if b == 0:
        raise ValueError("0 division")
    return a / b

def test_divide_by_zero():
    with pytest.raises(ValueError, match="0 division"):
        divide(10, 0)

6. mock

외부 의존성(API, DB)을 가짜로.

python
from unittest.mock import patch

def get_weather(city):
    import requests
    return requests.get(f"https://api/{city}").json()

def test_get_weather():
    with patch("requests.get") as mock_get:
        mock_get.return_value.json.return_value = {"temp": 20}
        assert get_weather("seoul") == {"temp": 20}

7. 커버리지

bash
pip install pytest-cov
pytest --cov=mypackage --cov-report=term-missing

자주 하는 실수

  1. 테스트 함수가 test_ 로 시작 안 함 — pytest가 못 찾음
  2. 테스트끼리 의존 — 실행 순서가 바뀌면 깨짐. 독립적으로
  3. 전역 상태 변경 — fixture로 setup/teardown
  4. 외부 API 진짜로 호출 — 느리고 불안정. mock 사용

FAQ

Q1. unittest vs pytest? — 표준은 unittest. 신규는 pytest 압도적 권장.

Q2. TDD 를 꼭? — 아니지만 권장. 핵심 기능은 테스트.

Q3. UI 테스트는? — Selenium, Playwright 등 별도.

💻 예제 (examples)

실제로 실행해 결과를 확인할 수 있는 예제 코드입니다.

01_thread_pool.py— ThreadPoolExecutor (I/O)
CODE
from concurrent.futures import ThreadPoolExecutor
import time

def fake_io(i):
    time.sleep(0.5)
    return i * 10

with ThreadPoolExecutor(max_workers=5) as ex:
    results = list(ex.map(fake_io, range(5)))
print(results)
▶ 실행 결과
[0, 10, 20, 30, 40]
02_process_pool.py— ProcessPoolExecutor (CPU)
CODE
from concurrent.futures import ProcessPoolExecutor

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

if __name__ == "__main__":
    with ProcessPoolExecutor() as ex:
        results = list(ex.map(heavy, [100_000, 200_000, 300_000]))
    print(results)
▶ 실행 결과
[333328333350000, 2666646666700000, 8999955000050000]
03_lock.py— threading.Lock
CODE
import threading

counter = 0
lock = threading.Lock()

def inc():
    global counter
    for _ in range(100_000):
        with lock:
            counter += 1

ts = [threading.Thread(target=inc) for _ in range(4)]
for t in ts: t.start()
for t in ts: t.join()
print(counter)
▶ 실행 결과
400000
04_compare.py— 순차 vs 스레드 시간 비교
CODE
import time
from concurrent.futures import ThreadPoolExecutor

def task(i):
    time.sleep(0.5)
    return i

start = time.perf_counter()
[task(i) for i in range(5)]
print(f"순차: {time.perf_counter() - start:.2f}s")

start = time.perf_counter()
with ThreadPoolExecutor(max_workers=5) as ex:
    list(ex.map(task, range(5)))
print(f"스레드: {time.perf_counter() - start:.2f}s")
▶ 실행 결과
순차: 2.50s
스레드: 0.50s
01_basic.py— async/await 기본
CODE
import asyncio

async def hello():
    print("hello")
    await asyncio.sleep(0.1)
    print("world")

asyncio.run(hello())
▶ 실행 결과
hello
world
02_gather.py— gather 동시 실행
CODE
import asyncio
import time

async def task(name, delay):
    await asyncio.sleep(delay)
    return f"{name}({delay}s)"

async def main():
    start = time.perf_counter()
    results = await asyncio.gather(
        task("A", 1),
        task("B", 2),
        task("C", 1),
    )
    print(results)
    print(f"총 {time.perf_counter() - start:.1f}s")

asyncio.run(main())
▶ 실행 결과
['A(1s)', 'B(2s)', 'C(1s)']
총 2.0s
03_queue.py— Queue producer/consumer
CODE
import asyncio

async def producer(q):
    for i in range(3):
        await q.put(i)
        print(f"생산: {i}")
    await q.put(None)

async def consumer(q):
    while True:
        item = await q.get()
        if item is None:
            break
        print(f"소비: {item}")

async def main():
    q = asyncio.Queue()
    await asyncio.gather(producer(q), consumer(q))

asyncio.run(main())
▶ 실행 결과
생산: 0
생산: 1
생산: 2
소비: 0
소비: 1
소비: 2
04_timeout.py— wait_for 타임아웃
CODE
import asyncio

async def slow():
    await asyncio.sleep(2)
    return "완료"

async def main():
    try:
        result = await asyncio.wait_for(slow(), timeout=1.0)
        print(result)
    except asyncio.TimeoutError:
        print("시간 초과")

asyncio.run(main())
▶ 실행 결과
시간 초과
01_basic_test.py— 기본 assert
CODE
def add(a, b):
    return a + b

def test_add_simple():
    assert add(2, 3) == 5

def test_add_negative():
    assert add(-1, 1) == 0

# 실행: pytest 01_basic_test.py -v
▶ 실행 결과
01_basic_test.py::test_add_simple PASSED
01_basic_test.py::test_add_negative PASSED
2 passed in 0.01s
02_parametrize.py— parametrize
CODE
import pytest

def add(a, b):
    return a + b

@pytest.mark.parametrize("a,b,expected", [
    (1, 2, 3),
    (-1, 1, 0),
    (0, 0, 0),
    (100, 200, 300),
])
def test_add(a, b, expected):
    assert add(a, b) == expected

# 실행: pytest 02_parametrize.py -v
▶ 실행 결과
test_add[1-2-3] PASSED
test_add[-1-1-0] PASSED
test_add[0-0-0] PASSED
test_add[100-200-300] PASSED
4 passed in 0.02s
03_fixture.py— fixture
CODE
import pytest

@pytest.fixture
def sample_users():
    return [{"name": "Alice"}, {"name": "Bob"}]

def test_first(sample_users):
    assert sample_users[0]["name"] == "Alice"

def test_count(sample_users):
    assert len(sample_users) == 2

# 실행: pytest 03_fixture.py -v
▶ 실행 결과
test_first PASSED
test_count PASSED
2 passed in 0.01s
04_mock.py— unittest.mock
CODE
from unittest.mock import patch

def get_temp(city):
    import requests
    r = requests.get(f"https://api/{city}")
    return r.json()["temp"]

def test_get_temp():
    with patch("requests.get") as mock_get:
        mock_get.return_value.json.return_value = {"temp": 20}
        assert get_temp("seoul") == 20

# 실행: pytest 04_mock.py -v
▶ 실행 결과
test_get_temp PASSED
1 passed in 0.05s

📝 과제 (exercises)

직접 풀어보고, 막힐 때 정답을 펼쳐 비교해보세요.

과제 1

동시 다운로드 시간 비교

목표: 여러 URL 다운로드를 순차/스레드로 비교한다.

요구사항
  • ThreadPoolExecutor 사용
  • 순차 시간과 스레드 시간 출력
💡 힌트

time.sleep 으로 네트워크 흉내

max_workers=5

입출력 예시
순차: 5.0s
스레드: 1.1s
채점
  • · 스레드 가속 확인
  • · 시간 출력
▶정답 코드 펼치기 / 접기
SOLUTION
import time
from concurrent.futures import ThreadPoolExecutor

def fetch(i):
    time.sleep(1)
    return i

urls = list(range(5))

start = time.perf_counter()
[fetch(u) for u in urls]
print(f"순차: {time.perf_counter() - start:.1f}s")

start = time.perf_counter()
with ThreadPoolExecutor(max_workers=5) as ex:
    list(ex.map(fetch, urls))
print(f"스레드: {time.perf_counter() - start:.1f}s")
▶ 실행 결과
순차: 5.0s
스레드: 1.0s
과제 2

CPU 작업 프로세스 풀

목표: 큰 합 계산을 ProcessPoolExecutor 로 가속한다.

요구사항
  • if __name__ == '__main__': 가드
  • 결과 출력
💡 힌트

heavy 함수 정의를 모듈 최상위에

입출력 예시
[333328333350000, 333328333350000, 333328333350000]
채점
  • · __main__ 가드
  • · 결과 list
▶정답 코드 펼치기 / 접기
SOLUTION
from concurrent.futures import ProcessPoolExecutor

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

if __name__ == "__main__":
    with ProcessPoolExecutor() as ex:
        results = list(ex.map(heavy, [100_000] * 3))
    print(results)
▶ 실행 결과
[333328333350000, 333328333350000, 333328333350000]
과제 3

Lock 적용 전후

목표: 공유 카운터에 Lock 적용 전후를 비교한다.

요구사항
  • Lock 없이 / Lock 있이 두 번 실행
  • 최종 값 출력
💡 힌트

증가량이 정확하면 Lock 적용 성공

입출력 예시
Lock 없음: 알 수 없음
Lock 있음: 400000
채점
  • · 두 번 실행
  • · 차이 관찰
▶정답 코드 펼치기 / 접기
SOLUTION
import threading

def run(use_lock):
    counter = 0
    lock = threading.Lock()
    def inc():
        nonlocal counter
        for _ in range(100_000):
            if use_lock:
                with lock:
                    counter += 1
            else:
                counter += 1
    ts = [threading.Thread(target=inc) for _ in range(4)]
    for t in ts: t.start()
    for t in ts: t.join()
    return counter

print("Lock 없음:", run(False))
print("Lock 있음:", run(True))
▶ 실행 결과
Lock 없음: 218394
Lock 있음: 400000
과제 4

동시 fetch 시간 측정

목표: 여러 가짜 fetch 를 동시 실행해 시간을 측정한다.

요구사항
  • fetch(i) 가 1초 sleep
  • 5개를 gather 로 동시 실행
  • 총 시간 출력
💡 힌트

time.perf_counter()

입출력 예시
[0, 1, 2, 3, 4]
총 1.0s
채점
  • · gather 사용
  • · 1초 안팎
▶정답 코드 펼치기 / 접기
SOLUTION
import asyncio
import time

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

async def main():
    start = time.perf_counter()
    results = await asyncio.gather(*(fetch(i) for i in range(5)))
    print(results)
    print(f"총 {time.perf_counter() - start:.1f}s")

asyncio.run(main())
▶ 실행 결과
[0, 1, 2, 3, 4]
총 1.0s
과제 5

Queue 작업 분배

목표: 1개 producer, 2개 consumer 로 작업을 나눠 처리한다.

요구사항
  • producer 가 5개 작업 put
  • consumer 2개가 동시에 get/처리
  • 종료 신호 처리
💡 힌트

q.put(None) 을 consumer 수만큼

입출력 예시
소비1: 0
소비2: 1
소비1: 2
... (순서는 다를 수 있음)
채점
  • · Queue 사용
  • · 두 consumer 동작
▶정답 코드 펼치기 / 접기
SOLUTION
import asyncio

async def producer(q, n):
    for i in range(n):
        await q.put(i)
    for _ in range(2):
        await q.put(None)

async def consumer(q, name):
    while True:
        item = await q.get()
        if item is None:
            break
        print(f"{name}: {item}")
        await asyncio.sleep(0.05)

async def main():
    q = asyncio.Queue()
    await asyncio.gather(
        producer(q, 5),
        consumer(q, "소비1"),
        consumer(q, "소비2"),
    )

asyncio.run(main())
▶ 실행 결과
소비1: 0
소비2: 1
소비1: 2
소비2: 3
소비1: 4
과제 6

타임아웃 fetch

목표: 느린 fetch 에 1초 타임아웃을 걸어 처리한다.

요구사항
  • asyncio.wait_for 사용
  • TimeoutError 처리
💡 힌트

except asyncio.TimeoutError

입출력 예시
시간 초과
채점
  • · timeout 동작
  • · 예외 처리
▶정답 코드 펼치기 / 접기
SOLUTION
import asyncio

async def slow_fetch():
    await asyncio.sleep(3)
    return "데이터"

async def main():
    try:
        result = await asyncio.wait_for(slow_fetch(), timeout=1)
        print(result)
    except asyncio.TimeoutError:
        print("시간 초과")

asyncio.run(main())
▶ 실행 결과
시간 초과
과제 7

단위 테스트 5개

목표: 간단한 함수에 대해 테스트 5개를 작성한다.

요구사항
  • is_even(n) -> bool 에 대해 짝수/홀수/0 케이스
  • sum_list(xs) 에 대해 빈/일반 케이스
💡 힌트

pytest 명령어로 실행

입출력 예시
5 passed
채점
  • · test_ 함수
  • · 5개 PASS
▶정답 코드 펼치기 / 접기
SOLUTION
# test_basics.py
def is_even(n):
    return n % 2 == 0

def sum_list(xs):
    return sum(xs)

def test_is_even_true():
    assert is_even(4) is True

def test_is_even_false():
    assert is_even(7) is False

def test_is_even_zero():
    assert is_even(0) is True

def test_sum_empty():
    assert sum_list([]) == 0

def test_sum_basic():
    assert sum_list([1, 2, 3]) == 6

# pytest test_basics.py -v
▶ 실행 결과
5 passed in 0.02s
과제 8

parametrize 테이블

목표: parametrize 로 6개 케이스 테스트.

요구사항
  • max_of(a, b) 에 대해 6개 (a, b, expected) 케이스
💡 힌트

@pytest.mark.parametrize

입출력 예시
6 passed
채점
  • · 6개 케이스
  • · 모두 통과
▶정답 코드 펼치기 / 접기
SOLUTION
import pytest

def max_of(a, b):
    return a if a > b else b

@pytest.mark.parametrize("a,b,expected", [
    (1, 2, 2),
    (5, 3, 5),
    (-1, -2, -1),
    (0, 0, 0),
    (100, 99, 100),
    (-5, 0, 0),
])
def test_max_of(a, b, expected):
    assert max_of(a, b) == expected

# pytest test_max.py -v
▶ 실행 결과
6 passed in 0.02s
과제 9

mock 으로 API 테스트

목표: 외부 호출을 mock 으로 대체한다.

요구사항
  • fetch_user(uid) 가 requests.get 사용
  • patch 로 결과 가짜
💡 힌트

mock_get.return_value.json.return_value

입출력 예시
1 passed
채점
  • · patch 사용
  • · 테스트 PASS
▶정답 코드 펼치기 / 접기
SOLUTION
from unittest.mock import patch

def fetch_user(uid):
    import requests
    return requests.get(f"https://api/users/{uid}").json()

def test_fetch_user():
    with patch("requests.get") as mock_get:
        mock_get.return_value.json.return_value = {"id": 1, "name": "Alice"}
        result = fetch_user(1)
    assert result == {"id": 1, "name": "Alice"}

# pytest test_user.py -v
▶ 실행 결과
1 passed in 0.05s
예제 코드 / 강의 자료

전체 강의 자료와 예제 코드는 GitHub에서 자유롭게 받아볼 수 있습니다.

GitHub에서 보기 ↗

댓글 0

아직 댓글이 없습니다. 첫 댓글을 남겨 주세요.

댓글은 회원만 쓸 수 있습니다.

광고·욕설·비방·개인정보·불법 내용이 담긴 댓글은 숨기거나 지웁니다. 댓글은 닉네임과 함께 공개됩니다. 개인정보처리방침