併發模式 | 程式語言精通

Goroutine + Channel 組合出來的幾個經典模式。實務上幾乎所有併發程式都是這幾種模式的變形。

Worker Pool#

固定數量的 worker 從同一個 jobs channel 拉工作、結果寫到 results channel。

flowchart LR
    J[Jobs Channel] --> W1[Worker 1]
    J --> W2[Worker 2]
    J --> W3[Worker 3]
    W1 --> R[Results Channel]
    W2 --> R
    W3 --> R

func workerPool(numWorkers int, jobs <-chan int, results chan<- int) {
    var wg sync.WaitGroup

    for i := 0; i < numWorkers; i++ {
        wg.Add(1)
        go func(id int) {
            defer wg.Done()
            for job := range jobs {
                results <- process(job)
            }
        }(i)
    }

    wg.Wait()
    close(results)
}

func main() {
    jobs := make(chan int, 100)
    results := make(chan int, 100)

    // 啟動 worker pool
    go workerPool(5, jobs, results)

    // 傳送工作
    for i := 0; i < 50; i++ {
        jobs <- i
    }
    close(jobs)

    // 收集結果
    for result := range results {
        fmt.Println(result)
    }
}

Pipeline#

每個階段是獨立的 goroutine，用 channel 把輸出串到下一個階段的輸入。

flowchart LR
    A[generator<br>產生資料] -->|chan int| B[square<br>平方運算]
    B -->|chan int| C[consumer<br>消費結果]

func generator(nums ...int) <-chan int {
    out := make(chan int)
    go func() {
        for _, n := range nums {
            out <- n
        }
        close(out)
    }()
    return out
}

func square(in <-chan int) <-chan int {
    out := make(chan int)
    go func() {
        for n := range in {
            out <- n * n
        }
        close(out)
    }()
    return out
}

func main() {
    // 建立 pipeline
    nums := generator(1, 2, 3, 4, 5)
    squared := square(nums)

    // 消費結果
    for result := range squared {
        fmt.Println(result)
    }
}

Fan-out / Fan-in#

把工作從一個 channel 分散給多個 goroutine（fan-out），再把多個結果 channel 合併回一個（fan-in）。

flowchart TB
    subgraph Fan-out
        IN[Input Channel] --> W1[Worker 1]
        IN --> W2[Worker 2]
        IN --> W3[Worker 3]
    end

    subgraph Fan-in
        W1 --> OUT[Output Channel]
        W2 --> OUT
        W3 --> OUT
    end

// Fan-out：多個 goroutine 讀取同一 channel
func fanOut(in <-chan int, n int) []<-chan int {
    outs := make([]<-chan int, n)
    for i := 0; i < n; i++ {
        outs[i] = worker(in)
    }
    return outs
}

// Fan-in：合併多個 channel 到單一 channel
func fanIn(channels ...<-chan int) <-chan int {
    var wg sync.WaitGroup
    out := make(chan int)

    for _, ch := range channels {
        wg.Add(1)
        go func(c <-chan int) {
            defer wg.Done()
            for v := range c {
                out <- v
            }
        }(ch)
    }

    go func() {
        wg.Wait()
        close(out)
    }()

    return out
}

Fan-out 適合 CPU bound 工作（用 worker 數 ≈ CPU 核心數）；fan-in 用一個額外 goroutine + WaitGroup 確保所有來源關閉後才 close output。