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

통신의 핵심 패턴 **Fan-out / Fan-in** — 생산자가 작업을 채널/큐에 넣고, 여러 워커가 나눠 처리(Fan-out)한 뒤, 결과를 다시 수집자로 모아(Fan-in) 최종 결과를 만듭니다. Python의 Queue/Pipe와 Go의 Channel 모두 이 흐름을 지원합니다.

한눈에 보기

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 — 파이썬의 비동기(이벤트 루프) 관점 통신.