이펙티브 파이썬 7장 - 동시성과 병렬성

동시성 vs 병렬성

구분	동시성 (Concurrency)	병렬성 (Parallelism)
의미	여러 작업을 번갈아가며 처리 (겉으로는 동시에)	여러 작업을 실제로 동시에 처리
처리 방식	하나의 코어가 작업을 빠르게 전환하며 처리	여러 코어가 각 작업을 동시에 실행
예시	카페에서 혼자 주문 받고, 계산하고, 음료 만들기	카페에서 세 명이 각각 주문, 계산, 음료 만들기
목적	응답성 향상, I/O 대기 시간 활용	처리 속도 향상
파이썬	스레드, 코루틴 (asyncio)	하위 프로세스, C 확장, 시스템 콜
자바스크립트	이벤트 루프, 프로미스, async/await	Web Workers, Node.js의 Worker Threads

 

GIL (Global Interupt Lock)

• 파이썬에는 GIL이라는 전역 인터프리터 락이 존재함
• 하나의 스레드만 파이썬 바이트코드를 실행할 수 있음
• -> 멀티스레드는 동시성은 가능하지만 병렬성이 제한적
• 병렬 처리를 원하면 하위 프로세스(자식 프로세스), C 확장, 시스템 콜을 활용할 수 있다.

파이썬에서의 멀티쓰레딩 작업

 

 

 

Better way 52 자식 프로세스를 관리하기 위해 subprocess를 사용하라

• subprocess 모듈을 사용해 자식 프로세스를 실행하고 입력과 출력 스트림을 관리 할 수 있다.
• 자식 프로세스는 파이썬 인터프리터와 병렬로 실행되기에 CPU 코어를 활용할 수 있다.
• subprocess.run()은 단순 실행
• subprocess.Popen()은 파이프라인 연결 등 고급 제어

import subprocess

# 단순한 명령 실행 - subprocess.run()
result = subprocess.run(['ls', '-l'], capture_output=True, text=True)

print("stdout:", result.stdout)
print("stderr:", result.stderr)
print("returncode:", result.returncode)

 

import subprocess

# ls와 grep을 파이프로 연결
ls_proc = subprocess.Popen(['ls'], stdout=subprocess.PIPE)
grep_proc = subprocess.Popen(['grep', 'py'], stdin=ls_proc.stdout, stdout=subprocess.PIPE)

ls_proc.stdout.close()  # 파이프 종료 신호
output, _ = grep_proc.communicate()

print(output.decode())

 

 

Better way 53 블로킹 I/O의 경우 스레드를 사용하고 병렬성을 피하라

import threading
import requests

def download(url):
    response = requests.get(url)
    print(f"{url}: {len(response.content)} bytes")

thread = threading.Thread(target=download, args=("https://example.com",))
thread.start()

 

Better way 54 스레드에서 데이터 경합을 피하기 위해 Lock을 사용하라

GIL (Global Interpreter Lock) 이 존재해서, 한 시점에 하나의 스레드만 바이트코드를 실행할 수 있다.

하지만 여전히 공유 자원에 대한 Race condition은 발생할 수 있다. 

class Counter:
    def __init__(self):
        self.counter = 0

    def increment(self, offset): # 여기에서 race condition이 발생
        value = self.counter
        time.sleep(0.00001)  # 딜레이 추가
        self.counter = value + offset


def worker(sensor_index, how_many, counter):
    for _ in range(how_many):
        counter.increment(1)


how_many = 1000  # 반복 적게
counter = Counter()

threads = []
for i in range(5):
    thread = threading.Thread(target=worker, args=(i, how_many, counter))
    threads.append(thread)
    thread.start()

for thread in threads:
    thread.join()

print(f"Counter should be {5 * how_many}, found {counter.counter}")

# 실제 출력값 - Counter should be 5000, found 1001

 

import threading
import time

class Counter:
    def __init__(self):
        self.counter = 0
        self.lock = threading.Lock() # Lock 사용

    def increment(self, offset):
        with self.lock:  # 공유 자원 접근 시 Lock으로 보호
            value = self.counter
            time.sleep(0.00001)  # 딜레이 있어도 안전
            self.counter = value + offset


def worker(sensor_index, how_many, counter):
    for _ in range(how_many):
        counter.increment(1)


how_many = 1000
counter = Counter()

threads = []
for i in range(5):
    thread = threading.Thread(target=worker, args=(i, how_many, counter))
    threads.append(thread)
    thread.start()

for thread in threads:
    thread.join()

print(f"Counter should be {5 * how_many}, found {counter.counter}")

# 실제 출력값 - Counter should be 5000, found 5000

 

 

 

 

 

 

Better Way 55 ~ 59 : 파이썬 내부의 동시성 패키지

팬인 (Fan-in)

• 여러 생산자(source)들이 데이터를 모아서 하나의 소비자(consumer)에게 전달하는 구조
• 여러 센서 데이터 → 하나의 처리기

팬아웃 (Fan-in)

• 하나의 생산자가 여러 소비자에게 작업을 분배하는 구조
• 크롤러가 URL 목록을 받아 여러 스레드/작업으로 분배 처리

 

요약

• 55 Queue를 사용해 스레드 사이의 작업을 조율하라
  • queue.Queue 사용해 스레드 간 안전하게 데이터 통신을 할 수 있다.
• 56 언제 동시성이 필요할지 인식하는 방법을 알아두라 
  • I/O vs CPU-bound
• 57 요구에 따라 팬아웃을 진행하려면 새로운 스레드를 생성하지 말라
  • 요청마다 스레드 생성하지 말고, ThreadPoolExecutor로 제한하라
• 58 동시성과 Queue를 사용하기 위해 코드를 어떻게 리팩터링해야 하는지 이해하라
  • Queue를 활용한 작업 분리 구조 리팩터링
• 59 동시성을 위해 스레드가 필요한 경우에는 ThreadpoolExecutor를 사용하라
  • 간단한 동시성 처리에는 ThreadPoolExecutor가 효과적

 

queue.Queue 를 활용하여 팬인/팬아웃 구현

import threading
from queue import Queue
import time

def worker(name, q):
    while True:
        task = q.get()
        if task is None:
            break
        print(f"{name} processing {task}")
        time.sleep(0.1)
        q.task_done()

q = Queue()

# 팬아웃: 하나의 큐에 여러 소비자
threads = []
for i in range(3):
    t = threading.Thread(target=worker, args=(f"Worker-{i}", q))
    t.start()
    threads.append(t)

# 팬인: 여러 생산자가 큐에 작업 전달
for i in range(10):
    q.put(f"task-{i}")

q.join()

for _ in range(3):
    q.put(None)
for t in threads:
    t.join()

 

ThreadPoolExecutor 를 활용하여 팬인/팬아웃 구현

from concurrent.futures import ThreadPoolExecutor

def task(n):
    print(f"Processing {n}")
    return n * 2

with ThreadPoolExecutor(max_workers=4) as executor:
    results = list(executor.map(task, range(10)))

 

Better Way 60 ~ 63 : asyncio (async - await) 코루틴을 사용해라

이벤트 루프

 

• asyncio는 단일 스레드 기반 이벤트 루프에서 코루틴을 순서대로 실행하며 I/O 타이밍을 효율적으로 처리함
• "비동기 함수 → 작업 등록 → 루프가 타이밍 맞춰 실행" 구조
• async def, await, asyncio.run()은 모두 이 이벤트 루프 기반에서 작동함

 

 

 

 

Better way 60 I/O를 할 때는 코루틴을 사용해 동시성을 높여라

import aiohttp
import asyncio

async def fetch(url):
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as resp:
            return await resp.text()

async def main():
    html = await fetch("https://example.com")
    print(html)

asyncio.run(main())

 

 

 

 

Better way 61 스레드를 사용한 I/O를 어떻게 asyncio로 포팅할 수 있는지 알아두라

Better way 62 asyncio로 쉽게 옮겨갈 수 있도록 스레드와 코루틴을 함께 사용하라

Better way 63 응답성을 최대로 높이려면 asyncio 이벤트 루프를 블록하지 말라

 

import asyncio
import time
from concurrent.futures import ThreadPoolExecutor

# 기존의 블로킹 I/O 함수 (예: 파일 읽기, DB 접근 등)
def legacy_blocking_io(task_id):
    print(f"[Blocking] Task {task_id} start")
    time.sleep(2)  # ❌ 블로킹 함수 (event loop 막지 않게 executor에서 실행해야 함)
    print(f"[Blocking] Task {task_id} done")
    return f"result-{task_id}"

# asyncio 기반의 비동기 작업 (예: I/O 모듈과 함께 쓰이는 함수)
async def async_io_task(task_id):
    print(f"[Async I/O] Task {task_id} start")
    await asyncio.sleep(1)  # ✅ 논블로킹 대기
    print(f"[Async I/O] Task {task_id} done")
    return f"async-result-{task_id}"

# 블로킹 함수를 코루틴처럼 실행 (Better way 62)
async def wrapped_blocking_io(task_id, executor):
    loop = asyncio.get_running_loop()
    return await loop.run_in_executor(executor, legacy_blocking_io, task_id)

async def main():
    executor = ThreadPoolExecutor(max_workers=3)

    tasks = []

    # 혼합된 작업 리스트 (블로킹 + 비동기)
    for i in range(5):
        if i % 2 == 0:
            # 블로킹 함수는 run_in_executor로 실행
            tasks.append(wrapped_blocking_io(i, executor))
        else:
            # 비동기 함수는 그대로 await
            tasks.append(async_io_task(i))

    results = await asyncio.gather(*tasks)

    print("\nFinal Results:")
    for res in results:
        print(res)

asyncio.run(main())

 

[Blocking] Task 0 start
[Async I/O] Task 1 start
[Blocking] Task 2 start
[Async I/O] Task 3 start
[Blocking] Task 4 start
[Async I/O] Task 1 done
[Async I/O] Task 3 done
[Blocking] Task 0 done
[Blocking] Task 2 done
[Blocking] Task 4 done

Final Results:
result-0
async-result-1
result-2
async-result-3
result-4

 

Better way 64 진정한 병렬성을 살리려면 concurrent.futures를 사용하라

import time
from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor

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

numbers = [10**7] * 8  # 8개의 작업

def run_with_executor(executor_class, label):
    start = time.time()
    with executor_class() as executor:
        results = list(executor.map(cpu_bound_task, numbers))
    duration = time.time() - start
    print(f"{label} took {duration:.2f} seconds")

if __name__ == "__main__":
    run_with_executor(ThreadPoolExecutor, "ThreadPoolExecutor")
    run_with_executor(ProcessPoolExecutor, "ProcessPoolExecutor")

# 출력 결과
# ThreadPoolExecutor took 2.71 seconds
# ProcessPoolExecutor took 0.72 seconds

 

항목	ThreadPoolExecutor	ProcessPoolExecutor
병렬성 방식	스레드 (GIL 제한 있음)	프로세스 (멀티코어 병렬 실행 가능)
CPU-bound 작업 처리	❌ 느림	✅ 빠름 (진정한 병렬성)
I/O-bound 작업 처리	✅ 효율적	❌ 오히려 느림 (프로세스 관리 비용 큼)