Go并发模式：安全有效地合并多个通道

package main

import (
    "fmt"
    "math/big"
    "sync" // 最终解决方案会用到
    "time" // 用于模拟生产数据
)

// Mux 函数：尝试将多个输入通道合并为一个输出通道 (初始版本 - 存在问题)
func Mux(channels []chan big.Int) chan big.Int {
    // n 用于计数已关闭的通道数量，当 n 归零时关闭输出通道。
    n := len(channels)
    // ch 是最终的输出通道，缓冲区大小设置为输入通道的数量。
    ch := make(chan big.Int, n)

    // 为每个输入通道启动一个 goroutine
    for _, c := range channels {
        go func() { // 问题根源之一：闭包变量捕获
            // 从输入通道 c 读取数据并发送到输出通道 ch
            for x := range c {
                ch <- x
            }
            // 输入通道 c 关闭后，递减 n
            n -= 1 // 问题根源之二：竞态条件
            // 如果所有输入通道都已关闭，则关闭输出通道
            if n == 0 {
                close(ch)
            }
        }()
    }
    return ch
}

// fromTo 函数：生成一个从 f 到 t-1 的 big.Int 序列并发送到通道
func fromTo(f, t int) chan big.Int {
    ch := make(chan big.Int)
    go func() {
        for i := f; i < t; i++ {
            // fmt.Println("Feed:", i) // 调试输出
            ch <- *big.NewInt(int64(i))
        }
        close(ch)
    }()
    return ch
}

// testMux 函数：测试 Mux 功能
func testMux() {
    r := make([]chan big.Int, 10)
    for i := 0; i < 10; i++ {
        r[i] = fromTo(i*10, i*10+10) // 创建10个输入通道，每个通道生成10个数字
    }
    all := Mux(r) // 调用 Mux 合并通道
    // 从合并后的通道读取并打印所有数据
    for l := range all {
        fmt.Println(l)
    }
}

// func main() {
//     testMux()
// }

for _, c := range channels {
    go func() { // 这里
        for x := range c { // 这里的 c
            ch <- x
        }
        // ...
    }()
}

for _, c := range channels {
    go func(inputChan <-chan big.Int) { // 将 c 作为参数 inputChan 传入
        for x := range inputChan {
            ch <- x
        }
        // ...
    }(c) // 立即执行并传入当前的 c 值
}

package main

import (
    "fmt"
    "math/big"
    "sync"
    "time"
)

/*
  Mux 函数：将多个输入通道的数据合并到一个输出通道。
  使用 sync.WaitGroup 安全地等待所有输入通道关闭。
*/
func Mux(channels []chan big.Int) chan big.Int {
    // wg 用于等待所有处理输入通道的 goroutine 完成。
    var wg sync.WaitGroup
    wg.Add(len(channels)) // 初始化 WaitGroup 计数器为输入通道的数量。

    // ch 是最终的输出通道，缓冲区大小设置为输入通道的数量，
    // 以便在所有输入通道关闭前，可以缓冲一些数据。
    ch := make(chan big.Int, len(channels))

    // 为每个输入通道启动一个 goroutine 来泵送数据。
    for _, c := range channels {
        // 关键：将循环变量 c 作为参数传入匿名函数，避免闭包捕获问题。
        go func(inputChan <-chan big.Int) {
            defer wg.Done() // 确保在 goroutine 退出时递减 WaitGroup 计数器。
            // 从输入通道读取所有数据并发送到输出通道。
            for x := range inputChan {
                ch <- x
            }
        }(c) // 立即执行匿名函数并传入当前的 c 值。
    }

    // 启动一个独立的 goroutine 来等待所有输入通道处理完成，然后关闭输出通道。
    go func() {
        wg.Wait() // 阻塞直到所有 goroutine 都调用了 wg.Done()。
        close(ch) // 所有输入通道都已关闭，此时可以安全关闭输出通道。
    }()

    return ch // 返回合并后的输出通道。
}

// fromTo 函数：生成一个从 f 到 t-1 的 big.Int 序列并发送到通道
func fromTo(f, t int) chan big.Int {
    ch := make(chan big.Int)
    go func() {
        for i := f; i < t; i++ {
            fmt.Println("Feed:", i) // 调试输出，观察数据生产顺序
            ch <- *big.NewInt(int64(i))
        }
        close(ch)
    }()
    return ch
}

// testMux 函数：测试 Mux 功能
func testMux() {
    r := make([]chan big.Int, 10)
    for i := 0; i < 10; i++ {
        r[i] = fromTo(i*10, i*10+10) // 创建10个输入通道，每个通道生成10个数字
    }
    all := Mux(r) // 调用 Mux 合并通道
    // 从合并后的通道读取并打印所有数据
    for l := range all {
        fmt.Println("Received:", l) // 调试输出，观察数据接收顺序
    }
    fmt.Println("All data received and processed.")
}

func main() {
    testMux()
    // 给予一些时间确保所有 goroutine 都完成，尽管 WaitGroup 已经处理了大部分同步。
    // time.Sleep(time.Second)
}