Go Channel 并发模式实战：Pipeline、Fan-Out/In、Or-Done 与选路策略

// 无缓冲：发送与接收同步配对，一方等待另一方
ch := make(chan int)

// 有缓冲：缓冲区未满时发送不阻塞，未空时接收不阻塞
ch := make(chan int, 10)

// 只读 / 只写（方向约束，常用于参数签名防误用）
var in <-chan int  // 只读
var out chan<- int // 只写

// 判断通道是否已关闭的标准姿势
v, ok := <-ch
if !ok {
    // ch 已关闭且无残留数据
    return
}

// 阶段1：生成整数
func gen(nums ...int) <-chan int {
    out := make(chan int)
    go func() {
        defer close(out)
        for _, n := range nums {
            out <- n
        }
    }()
    return out
}

// 阶段2：平方
func sq(in <-chan int) <-chan int {
    out := make(chan int)
    go func() {
        defer close(out)
        for n := range in {
            out <- n * n
        }
    }()
    return out
}

func main() {
    // 连接成管道：gen → sq → sq
    out := sq(sq(gen(1, 2, 3, 4)))
    for n := range out {
        fmt.Println(n) // 1, 16, 81, 256
    }
}

□ 每个阶段独立成 goroutine，天然并行
□ 阶段间用 channel 解耦，改一处不影响全局
□ 数据流单向，错误定位清晰
□ 可任意组合（中间插入新阶段）

关键点：
- 每阶段记得 close(out)，让下游 range 正常退出
- 数据流是"拉"式的：下游就绪才推进，天然背压
- 单个慢阶段会拖慢整条链 → 慢阶段考虑 Fan-Out 扩容

// 一个输入 channel 分发到 N 个 worker，充分利用多核
func fanOut(in <-chan int, workers int) {
    var wg sync.WaitGroup
    for i := 0; i < workers; i++ {
        wg.Add(1)
        go func(id int) {
            defer wg.Done()
            for n := range in {   // 每个 worker 独立消费
                process(id, n)
            }
        }(i)
    }
    wg.Wait()
}

// 多个来源 channel 合并到一个输出 channel
func fanIn(channels ...<-chan int) <-chan int {
    out := make(chan int)
    var wg sync.WaitGroup
    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 适用：
  - 任务可并行拆分（无依赖的独立任务）
  - 单 worker 是瓶颈（CPU 密集/IO 并行）
  - 数据量大，单 goroutine 处理不过来

Fan-In 适用：
  - 多路查询结果聚合（多数据源合并）
  - 多 worker 完成后结果汇流

边界与风险：
  - 不是所有任务都能并行（有顺序依赖的不能拆）
  - worker 数不要超过实际并行能力（过度并发反降性能）
  - 上游关闭要能及时传导，避免 worker 永久阻塞

func orDone(done <-chan struct{}, in <-chan int) <-chan int {
    out := make(chan int)
    go func() {
        defer close(out)
        for {
            select {
            case <-done:
                return            // 收到取消信号，立即退出
            case v, ok := <-in:
                if !ok {
                    return        // 数据源关闭
                }
                select {
                case out <- v:
                case <-done:
                    return
                }
            }
        }
    }()
    return out
}

// 调用方：带取消地消费
for v := range orDone(ctx.Done(), source) {
    // 处理 v，随时可被 ctx 取消
}

为什么发送时还要再 select 一次？
  接收端或Done可能阻塞在 out <- v 上
  如果此时 done 触发了，goroutine 就卡死泄漏
  → 发送前再 select 一次 done，双保险退出

Or-Done 的价值：
  □ 消费方只需面对一个 channel，取消逻辑被封装
  □ 不漏数据：done 触发时未完成的数据被丢弃是调用方决策
  □ goroutine 不泄漏：任何路径都能退出

select {
case v := <-ch1:
    handle(v)
case ch2 <- v:
    sendOK()
case <-time.After(3 * time.Second):
    timeout()
default:
    // 所有 case 都不可立即执行时的兜底（非阻塞 select）
}

select 的三个特性：
  1. 同时就绪的多个 case 随机选择 → 天然公平，避免饿死某个 channel
  2. 空 select（无 case）会永久阻塞 → 常用于主 goroutine 挂起
  3. default 让 select 变成非阻塞 → 实现轮询/尝试发送接收

// 定时器驱动的心跳 + 数据 + 超时三路合一
ticker := time.NewTicker(500 * time.Millisecond)
defer ticker.Stop()
for {
    select {
    case <-ticker.C:
        sendHeartbeat()
    case v, ok := <-workCh:
        if !ok {
            return
        }
        process(v)
    case <-ctx.Done():
        return          // 全局取消优先
    }
}