4~6주차 — 동시성 스레딩과 멀티프로세싱 · 동시성 asyncio · 테스트
I/O 바운드 작업은 ThreadPoolExecutor 로, CPU 바운드는 ProcessPoolExecutor 로 — GIL 의 의미와 동시성 도구의 선택 기준을 익힙니다. async/await 문법, asyncio.gather 로 동시 실행, Queue 기반 producer/consumer, 그리고 wait_for 타임아웃까지 — 비동기 프로그래밍의 핵심을 익힙니다. pytest 기본 사용법, parametrize 로 다양한 케이스, fixture 로 공통 setup, 그리고 mock 으로 외부 의존성을 격리하는 방법까지 다룹니다.
이 강의에서 배우는 것
- 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
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
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
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
여러 스레드가 같은 자원을 변경할 때 데이터 깨짐 방지.
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) # 400000Lock 없으면 결과가 작거나 가변적입니다.
자주 하는 실수
- CPU 작업에 thread 사용 — GIL 때문에 느려질 수도
- __main__ 가드 누락(Windows + multiprocessing) — 무한 루프
- 공유 자원에 Lock 누락 — 결과 비결정적
- 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 로 정의된 함수는 코루틴. 호출해도 즉시 실행 안 되고 객체 반환.
import asyncio
async def hello():
print("hello")
await asyncio.sleep(1)
print("world")
asyncio.run(hello())2. asyncio.gather
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
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
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. 타임아웃
try:
result = await asyncio.wait_for(slow_task(), timeout=1.0)
except asyncio.TimeoutError:
print("시간 초과")6. 동기 vs 비동기
| 스레딩 | asyncio | |
|---|---|---|
| 단위 | OS 스레드 | 코루틴 |
| 컨텍스트 스위칭 | OS 결정 | await 에서만 |
| 메모리 | 무거움 | 가벼움 |
| 라이브러리 | 대부분 호환 | async 지원 필요 |
자주 하는 실수
- async def 결과를 await 안 함 — 코루틴 객체만 반환
- 동기 함수 안에서 await — async def 안에서만
- 블로킹 함수 호출 — time.sleep 대신 await asyncio.sleep
- 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 기본
pip install pytest테스트 파일은 test_*.py 또는 *_test.py 입니다.
# test_calc.py
def add(a, b): return a + b
def test_add():
assert add(2, 3) == 5
assert add(-1, 1) == 0pytest # 전체 실행
pytest test_calc.py # 한 파일
pytest -v # verbose
pytest -k "add" # 이름 매칭3. parametrize
같은 테스트를 여러 입력으로.
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) == expected4. fixture
@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):
@pytest.fixture(scope="module")
def expensive():
print("한 번만 setup")
return ...5. 예외 검증
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)을 가짜로.
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. 커버리지
pip install pytest-cov
pytest --cov=mypackage --cov-report=term-missing자주 하는 실수
- 테스트 함수가 test_ 로 시작 안 함 — pytest가 못 찾음
- 테스트끼리 의존 — 실행 순서가 바뀌면 깨짐. 독립적으로
- 전역 상태 변경 — fixture로 setup/teardown
- 외부 API 진짜로 호출 — 느리고 불안정. mock 사용
FAQ
Q1. unittest vs pytest? — 표준은 unittest. 신규는 pytest 압도적 권장.
Q2. TDD 를 꼭? — 아니지만 권장. 핵심 기능은 테스트.
Q3. UI 테스트는? — Selenium, Playwright 등 별도.
💻 예제 (examples)
실제로 실행해 결과를 확인할 수 있는 예제 코드입니다.
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]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]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)400000import 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.50simport asyncio
async def hello():
print("hello")
await asyncio.sleep(0.1)
print("world")
asyncio.run(hello())hello
worldimport 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.0simport 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
소비: 2import 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())시간 초과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 -v01_basic_test.py::test_add_simple PASSED
01_basic_test.py::test_add_negative PASSED
2 passed in 0.01simport 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 -vtest_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.02simport 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 -vtest_first PASSED
test_count PASSED
2 passed in 0.01sfrom 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 -vtest_get_temp PASSED
1 passed in 0.05s📝 과제 (exercises)
직접 풀어보고, 막힐 때 정답을 펼쳐 비교해보세요.
동시 다운로드 시간 비교
목표: 여러 URL 다운로드를 순차/스레드로 비교한다.
- ThreadPoolExecutor 사용
- 순차 시간과 스레드 시간 출력
time.sleep 으로 네트워크 흉내
max_workers=5
순차: 5.0s
스레드: 1.1s- · 스레드 가속 확인
- · 시간 출력
▶정답 코드 펼치기 / 접기
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.0sCPU 작업 프로세스 풀
목표: 큰 합 계산을 ProcessPoolExecutor 로 가속한다.
- if __name__ == '__main__': 가드
- 결과 출력
heavy 함수 정의를 모듈 최상위에
[333328333350000, 333328333350000, 333328333350000]- · __main__ 가드
- · 결과 list
▶정답 코드 펼치기 / 접기
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]Lock 적용 전후
목표: 공유 카운터에 Lock 적용 전후를 비교한다.
- Lock 없이 / Lock 있이 두 번 실행
- 최종 값 출력
증가량이 정확하면 Lock 적용 성공
Lock 없음: 알 수 없음
Lock 있음: 400000- · 두 번 실행
- · 차이 관찰
▶정답 코드 펼치기 / 접기
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동시 fetch 시간 측정
목표: 여러 가짜 fetch 를 동시 실행해 시간을 측정한다.
- fetch(i) 가 1초 sleep
- 5개를 gather 로 동시 실행
- 총 시간 출력
time.perf_counter()
[0, 1, 2, 3, 4]
총 1.0s- · gather 사용
- · 1초 안팎
▶정답 코드 펼치기 / 접기
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.0sQueue 작업 분배
목표: 1개 producer, 2개 consumer 로 작업을 나눠 처리한다.
- producer 가 5개 작업 put
- consumer 2개가 동시에 get/처리
- 종료 신호 처리
q.put(None) 을 consumer 수만큼
소비1: 0
소비2: 1
소비1: 2
... (순서는 다를 수 있음)- · Queue 사용
- · 두 consumer 동작
▶정답 코드 펼치기 / 접기
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타임아웃 fetch
목표: 느린 fetch 에 1초 타임아웃을 걸어 처리한다.
- asyncio.wait_for 사용
- TimeoutError 처리
except asyncio.TimeoutError
시간 초과- · timeout 동작
- · 예외 처리
▶정답 코드 펼치기 / 접기
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())시간 초과단위 테스트 5개
목표: 간단한 함수에 대해 테스트 5개를 작성한다.
- is_even(n) -> bool 에 대해 짝수/홀수/0 케이스
- sum_list(xs) 에 대해 빈/일반 케이스
pytest 명령어로 실행
5 passed- · test_ 함수
- · 5개 PASS
▶정답 코드 펼치기 / 접기
# 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 -v5 passed in 0.02sparametrize 테이블
목표: parametrize 로 6개 케이스 테스트.
- max_of(a, b) 에 대해 6개 (a, b, expected) 케이스
@pytest.mark.parametrize
6 passed- · 6개 케이스
- · 모두 통과
▶정답 코드 펼치기 / 접기
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 -v6 passed in 0.02smock 으로 API 테스트
목표: 외부 호출을 mock 으로 대체한다.
- fetch_user(uid) 가 requests.get 사용
- patch 로 결과 가짜
mock_get.return_value.json.return_value
1 passed- · patch 사용
- · 테스트 PASS
▶정답 코드 펼치기 / 접기
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 -v1 passed in 0.05s댓글 0
아직 댓글이 없습니다. 첫 댓글을 남겨 주세요.
댓글은 회원만 쓸 수 있습니다.
광고·욕설·비방·개인정보·불법 내용이 담긴 댓글은 숨기거나 지웁니다. 댓글은 닉네임과 함께 공개됩니다. 개인정보처리방침