Go语言并发编程：构建健壮的通道复用器

package main

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

// Mux 函数尝试将多个 big.Int 类型的通道合并为一个
func Mux(channels []chan big.Int) chan big.Int {
    // n 用于计数，当所有输入通道关闭时，关闭输出通道
    n := len(channels)
    // 创建带缓冲的输出通道，缓冲区大小为输入通道的数量
    ch := make(chan big.Int, n)

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

// fromTo 辅助函数，生成一个包含指定范围整数的通道
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()
}

package main

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

/*
  Multiplex a number of channels into one.
*/
func Mux(channels []chan big.Int) chan big.Int {
    // 创建一个 WaitGroup，用于等待所有输入通道的 goroutine 完成
    var wg sync.WaitGroup
    // WaitGroup 的计数器设置为输入通道的数量
    wg.Add(len(channels))

    // 创建带缓冲的输出通道。缓冲区大小可以根据需求调整，这里使用输入通道的数量。
    ch := make(chan big.Int, len(channels))

    // 为每个输入通道启动一个 goroutine
    for _, c := range channels {
        // 关键改进：将循环变量 c 作为参数传递给匿名函数
        // 这样每个 goroutine 都会拥有 c 的一个局部副本，避免了闭包捕获问题。
        go func(c <-chan big.Int) { // 使用只读通道类型更安全
            // 从输入通道读取数据并发送到输出通道
            for x := range c {
                ch <- x
            }
            // 当一个输入通道的 goroutine 完成任务后，调用 wg.Done() 减少计数器
            wg.Done()
        }(c) // 将当前的 c 值传递给 goroutine
    }

    // 启动一个独立的 goroutine，负责在所有输入通道的 goroutine 完成后关闭输出通道
    go func() {
        // 阻塞，直到 WaitGroup 的计数器归零（即所有输入通道的 goroutine 都已完成）
        wg.Wait()
        // 所有 goroutine 完成后，安全地关闭输出通道
        close(ch)
    }()

    return ch
}

// fromTo 辅助函数，生成一个包含指定范围整数的通道
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 个整数
    }

    start := time.Now()
    all := Mux(r) // 调用 Mux 进行复用

    count := 0
    // 从复用后的通道中读取并打印数据
    for l := range all {
        fmt.Println(l)
        count++
    }
    elapsed := time.Since(start)
    fmt.Printf("Total items received: %d\n", count)
    fmt.Printf("Time taken: %s\n", elapsed)
}

func main() {
    testMux()
}