package main

import (
    "fmt"
)

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

// Stage 2: 處理器（求平方）
func square(in <-chan int) <-chan int {
    out := make(chan int)
    go func() {
        for n := range in {
            out <- n * n
        }
        close(out)
    }()
    return out
}

// Stage 3: 輸出階段
func printResults(in <-chan int) {
    for n := range in {
        fmt.Println("結(jié)果:", n)
    }
}

func main() {
    // 構(gòu)建流水線
    gen := generator(1, 2, 3, 4, 5)
    sq := square(gen)
    printResults(sq)
}

結(jié)果: 1
結(jié)果: 4
結(jié)果: 9
結(jié)果: 16
結(jié)果: 25

[Generator] --> [Square] --> [Print]
     |              |          |
    goroutine     goroutine   主線程

func merge(cs ...<-chan int) <-chan int {
    out := make(chan int)
    var wg sync.WaitGroup

    output := func(c <-chan int) {
        for n := range c {
            out <- n
        }
        wg.Done()
    }

    wg.Add(len(cs))
    for _, c := range cs {
        go output(c)
    }

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

in := generator(1, 2, 3, 4, 5, 6)
c1 := square(in)
c2 := square(in) // 注意不能重復(fù)消費(fèi)同一個(gè)channel
// 正確方式是廣播in的內(nèi)容到多個(gè)square協(xié)程

// 這里只是示意，如果需要并發(fā)執(zhí)行 square，需用 fan-out + fan-in 模式