Orc’s WIKI

Orc Hwang

한눈에 보기
4단계의 락은 “공유 상태를 보호”하는 방식이었습니다. 이번 단계는 다른 철학인 “메모리를 공유하지 말고, 메시지를 주고받자(Communication)” 를 다룹니다. 작업을 메시지로 나눠 주고받으면, 공유 상태를 거의 없앨 수 있어 락·레이스·데드락이 줄어듭니다.
파이썬의 queue.Queue, multiprocessing.Queue, Pipe를 익힙니다.
Go의 채널(Channel) 과 select 를 익힙니다.
두 언어 모두로 Fan-out / Fan-in 파이프라인을 구현해 비교합니다.
Go의 격언: “메모리를 공유함으로써 통신하지 말고, 통신함으로써 메모리를 공유하라.” 이 단계의 모토입니다.
Python 통신 기본기
queue.Queue — 스레드 간 안전한 큐
queue.Queue는 여러 스레드가 동시에 안전하게 작업을 넣고 빼는 큐입니다. 내부적으로 락을 알아서 처리하므로, 우리가 직접 락을 관리할 필요가 없습니다.
import queue, threading, time
q = queue.Queue()
def producer():
    for i in range(10):
        q.put(i)                 # 작업 넣기
        time.sleep(0.05)
    q.put(None)                  # 종료 신호 (sentinel)
def worker(name):
    while True:
        item = q.get()           # 작업 꺼내기 (비면 대기)
        if item is None:         # 종료 신호 감지
            q.put(None)          # 다음 워커를 위해 재전파
            break
        print(f"[{name}] 처리 {item}")
        q.task_done()            # 처리 완료 통지
threading.Thread(target=producer, daemon=True).start()
for n in ("A", "B"):
    threading.Thread(target=worker, args=(n,), daemon=True).start()
q.join()   # 모든 task_done()까지 대기
핵심 메서드: put()(넣기), get()(꺼내기, 비면 블로킹), task_done()/join()(처리 완료 추적). 작업의 끝을 알리는 None 같은 sentinel을 넣는 패턴이 일반적입니다.
multiprocessing.Queue — 프로세스 간 안전한 큐
프로세스 간에는 메모리를 공유하지 않으므로, 별도의 IPC 전용 큐인 multiprocessing.Queue를 씁니다. 내부적으로 데이터를 직렬화(pickle)해 전송합니다.
from multiprocessing import Process, Queue
def producer(q):
    for i in range(5):
        q.put(i*10)
    q.put(None)
def worker(q):
    while True:
        item = q.get()
        if item is None:
            break
        print("받음:", item)
q = Queue()
p1 = Process(target=producer, args=(q,))
p2 = Process(target=worker, args=(q,))
p1.start(); p2.start(); p1.join(); p2.join()
프로세스 풀과 함께 쓰면 CPU 병렬 작업을 나누는 강력한 도구가 됩니다. multiprocessing.Manager()는 큐·딕셔너리·리스트 등을 프로세스 간 공유 상태로 만들어주는 또 다른 방법입니다.
Pipe — 두 지점 간 양방향 통신
Pipe()는 두 프로세스(또는 스레드) 사이의 점대점(point-to-point) 통신입니다. 한쪽 끝이 다른 쪽과 직접 연결됩니다.
from multiprocessing import Process, Pipe
def child(conn):
    msg = conn.recv()          # 부모가 보낸 메시지 수신
    conn.send(f"hello {msg}")  # 응답 송신
parent_conn, child_conn = Pipe()
p = Process(target=child, args=(child_conn,))
p.start()
parent_conn.send("world")      # 자식으로 송신
print(parent_conn.recv())      # 자식 응답 수신 → "hello world"
p.join()
큐(여러 소비자) vs 파이프(점대점 1:1)를 용도에 따라 고릅니다.
Go Channel — 통신의 일급 시민
기본: chan과 <-
Go의 채널은 고루틴 사이에 값을 주고받는 타입 있는 파이프입니다. 선언은 chan T(타입)이고, make로 만들며, <-로 보내고 받습니다.
package main
import "fmt"
func main() {
    ch := make(chan int)   // 정수를 주고받는 채널
    go func() {
        ch <- 42            // 보내기
    }()
    v := <-ch               // 받기 (보낼 때까지 블로킹)
    fmt.Println(v)          // 42
}
언버퍼드(unbuffered) 채널: 기본값. 보내는 측은 받는 측이 받을 때까지 블로킹되어, 양쪽이 동기화(핸드셰이크)됩니다. 자연스러운 동기화 수단입니다.
버퍼 채널: make(chan int, 3)처럼 크기를 주면, 버퍼가 찰 때까지 블로킹 없이 보낼 수 있습니다.
ch := make(chan int, 3)  // 3개까지 버퍼
ch <- 1
ch <- 2
ch <- 3
// 아직 차지 않아 블로킹 안 함
Fan-out / Fan-in — 여러 워커로 분배 후 수집
이것이 이 단계의 핵심 실습입니다. 작업 채널을 만들어 여러 워커 고루틴이 나눠 처리하고(Fan-out), 결과 채널로 다시 모읍니다(Fan-in).
package main
import (
    "fmt"
    "sync"
    "time"
)
func worker(id int, jobs <-chan int, results chan<- int) {
    for j := range jobs {              // 작업 채널을 모두 소진할 때까지
        time.Sleep(50 * time.Millisecond)
        results <- j * 2               // 결과 채널로 보내기 (Fan-in)
    }
}
func main() {
    const jobs = 10
    jobsCh := make(chan int, jobs)
    results := make(chan int, jobs)
    // 워커 4개 시작 (Fan-out)
    var wg sync.WaitGroup
    for w := 1; w <= 4; w++ {
        wg.Add(1)
        go func(id int) {
            defer wg.Done()
            worker(id, jobsCh, results)
        }(w)
    }
    // 작업 넣기
    for j := 1; j <= jobs; j++ {
        jobsCh <- j
    }
    close(jobsCh)            // 작업 채널 닫기 → 워커가 for range로 종료
    // 결과 수집 (Fan-in) — 워커가 끝나면 결과 채널 닫기
    go func() {
        wg.Wait()
        close(results)
    }()
    for r := range results { // 결과 채널이 닫힐 때까지 수신
        fmt.Println(r)
    }
}
포인트:
sem <- struct{}{} (2단계의 채널 세마포어)와 함께 쓰면 동시 실행 수를 제한하며 작업을 나눌 수 있습니다.
close(ch) 로 채널을 닫으면, for range 하는 수신자는 모든 값을 받고 깔끔하게 종료됩니다.
select — 여러 채널에서 동시 대기
select는 여러 채널 중 하나라도 준비되면 그 쪽을 실행하는 Go의 강력한 문법입니다. 타임아웃, 취소, 우선순위 라우팅에 유용합니다.
package main
import (
    "fmt"
    "time"
)
func main() {
    fast := make(chan string)
    slow := make(chan string)
    go func() { time.Sleep(200 * time.Millisecond); fast <- "빠른 응답" }()
    go func() { time.Sleep(500 * time.Millisecond); slow <- "느린 응답" }()
    select {
    case msg := <-fast:
        fmt.Println("수신:", msg)   // 먼저 준비된 쪽을 실행
    case msg := <-slow:
        fmt.Println("수신:", msg)
    case <-time.After(100 * time.Millisecond):  // 타임아웃
        fmt.Println("타임아웃!")
    }
}
select는 타임아웃(time.After), 취소 처리(context.Done()), 그리고 여러 작업 소스를 한 번에 기다릴 때 유용합니다. 4단계의 데드락 예방에서 언급한 “일정 시간만 기다리고 포기”도 이 select로 구현합니다.
실습 — 파이프라인: URL 텍스트 길이 합산
같은 문제를 두 언어로 풀어봅니다. 여러 URL에서 본문을 받아 글자 수를 합산하는 파이프라인을 Fan-out/Fan-in으로 구현합니다. (네트워크 I/O + CPU가 섞인 워크로드로, 2~3단계 지식이 함께 적용됩니다.)
Python — ThreadPoolExecutor + 부분 결과 합산:
from concurrent.futures import ThreadPoolExecutor
import urllib.request
def body_len(url):
    try:
        return len(urllib.request.urlopen(url, timeout=3).read())
    except Exception:
        return 0
urls = [f"https://example.com?p={i}" for i in range(20)]
with ThreadPoolExecutor(max_workers=5) as ex:
    lengths = list(ex.map(body_len, urls))   # Fan-out 처리 후 순서대로 수집
print("총 글자 수:", sum(lengths))
Go — 작업 채널 + 결과 채널 Fan-out/Fan-in:
package main
import (
    "fmt"
    "io"
    "net/http"
    "sync"
)
func main() {
    urls := make(chan string, 20)   // 작업 채널
    results := make(chan int)       // 결과 채널
    // 워커 5개 (Fan-out)
    var wg sync.WaitGroup
    for w := 0; w < 5; w++ {
        wg.Add(1)
        go func() {
            defer wg.Done()
            for u := range urls {
                resp, err := http.Get(u)
                if err != nil {
                    results <- 0
                    continue
                }
                body, _ := io.ReadAll(resp.Body)
                resp.Body.Close()
                results <- len(body)
            }
        }()
    }
    // 작업 넣고 닫기
    go func() {
        for i := 0; i < 20; i++ {
            urls <- fmt.Sprintf("https://example.com?p=%d", i)
        }
        close(urls)
    }()
    // 결과 수집 (Fan-in)
    go func() {
        wg.Wait()
        close(results)
    }()
    total := 0
    for n := range results { total += n }
    fmt.Println("총 글자 수:", total)
}
두 언어 모두 “작업을 나눠 처리(Fan-out) → 결과를 모아 합산(Fan-in)”이라는 동일한 구조로 풀었습니다. 파이썬은 ThreadPoolExecutor.map이 순서 보장과 수집을 알아서 하고, Go는 채널을 직접 열고 닫으며 제어합니다.
Python vs Go 통신 비교
기능
      Python
      Go
    
스레드 간 큐
      queue.Queue
      채널 (chan)
    
프로세스 간 큐
      multiprocessing.Queue/Manager
      (고루틴으로 충분)
    
점대점 통신
      multiprocessing.Pipe
      채널 (보통 언버퍼드)
    
버퍼 크기
      Queue(maxsize=...)
      make(chan T, K)
    
채널 닫기
      sentinel(None) 패턴
      내장 close(ch)
    
다중 대기
      (직접 구현)
      select 문법
    
동시성 제한
      ThreadPoolExecutor(max_workers)
      버퍼 채널 세마포어
    
결과 순서
      map()이 순서 보장
      Fan-in은 순서 보장 안 함(직접 처리)
    
Summary
통신(메시지 전달) 은 락보다 우아한 동시성 방식입니다. 공유 상태를 줄이면 레이스와 데드락도 줄어듭니다.
Python은 queue.Queue(스레드), multiprocessing.Queue/Pipe(프로세스)를 제공합니다.
Go의 채널은 타입 있는 파이프로, select 와 함께 다중 대기·타임아웃·취소를 다룹니다.
Fan-out/Fan-in 파이프라인 — 작업 채널로 분배하고 결과 채널로 수집하는 패턴 — 이 통신의 가장 대표적인 실무 활용입니다.
close(ch)/sentinel을 써서 작업의 끝을 알리는 것이 중요합니다.
다음 학습 (Next Learning)
Stage 6 · 실전 통합 — 여기서 익힌 채널·큐·select를 하나의 실제 웹 크롤러 프로젝트로 통합합니다.
Go by Example — Channels & Select — Go 채널 심화 예제 모음.
Asyncio Eventloop Optimization — 파이썬의 비동기(이벤트 루프) 관점 통신.

Stage 5 · 통신 — Queue · Pipe · Channel · select

댓글