Go语言多通道并发读取与聚合策略

func addnum(num1, num2, sum chan int) {
    done := make(chan bool)
    go func() {
        n1 := <- num1 // 尝试读取num1
        done <- true
    }()
    n2 := <- num2 // 尝试读取num2
    <- done       // 等待n1读取完成
    sum <- n1 + n2 // 求和
}

package main

import (
    "fmt"
    "time"
)

// numgen 模拟一个数字生成器，向通道发送指定数量的数字
func numgen(id int, count int, out chan<- int) {
    for i := 1; i <= count; i++ {
        time.Sleep(time.Millisecond * 50) // 模拟生产延迟
        out <- i * id
        fmt.Printf("Generator %d sent %d\n", id, i*id)
    }
    close(out) // 完成发送后关闭通道
    fmt.Printf("Generator %d finished.\n", id)
}

// aggregator 负责从两个输入通道读取并求和
func aggregator(in1, in2 <-chan int, out chan<- int) {
    defer close(out) // 确保输出通道在聚合器退出时关闭

    // 用于跟踪通道是否已关闭
    in1Open, in2Open := true, true
    var n1, n2 int // 存储从通道读取的值
    var sum int

    for in1Open || in2Open { // 只要有一个输入通道还开着，就继续循环
        select {
        case val, ok := <-in1:
            if !ok { // in1 已关闭
                in1Open = false
                fmt.Println("Channel in1 closed.")
                // 如果in1关闭，但in2还开着，我们需要继续处理in2的剩余数据
                // 但对于“zip”操作，如果一个通道关闭，通常意味着无法再配对
                // 在此示例中，我们假设一旦一个通道关闭，就无法再进行配对求和
                // 实际应用中，这里可能需要更复杂的逻辑来处理不平衡的情况
                if in2Open { // 如果另一个通道还开着，等待它关闭
                    for range in2 {
                        // 消费掉剩余数据，或者根据业务逻辑决定如何处理
                    }
                    in2Open = false
                    fmt.Println("Channel in2 consumed after in1 closed.")
                }
                return // 终止聚合器
            }
            n1 = val

            // 尝试立即从in2读取以完成配对
            select {
            case val2, ok2 := <-in2:
                if !ok2 { // in2 在等待时关闭
                    fmt.Println("Channel in2 closed while waiting for pairing.")
                    return // 无法配对，终止
                }
                n2 = val2
            default:
                // 如果in2没有立即准备好，这表示数据可能不平衡
                // 对于严格的“zip”操作，这可能是一个错误或需要等待
                // 这里我们简化处理，认为如果不能立即配对，就等待下一个循环
                // 实际生产中，可能需要一个缓冲区或更复杂的同步
                fmt.Println("Warning: in2 not immediately available for pairing with in1. Re-evaluating in next select cycle.")
                continue // 跳过当前循环，重新进入select
            }

            sum = n1 + n2
            out <- sum
            fmt.Printf("Aggregated %d + %d = %d\n", n1, n2, sum)

        case val, ok := <-in2:
            if !ok { // in2 已关闭
                in2Open = false
                fmt.Println("Channel in2 closed.")
                // 同样，处理in1的剩余数据或终止
                if in1Open {
                    for range in1 {
                        // 消费掉剩余数据
                    }
                    in1Open = false
                    fmt.Println("Channel in1 consumed after in2 closed.")
                }
                return // 终止聚合器
            }
            n2 = val

            // 尝试立即从in1读取以完成配对
            select {
            case val1, ok1 := <-in1:
                if !ok1 { // in1 在等待时关闭
                    fmt.Println("Channel in1 closed while waiting for pairing.")
                    return // 无法配对，终止
                }
                n1 = val1
            default:
                fmt.Println("Warning: in1 not immediately available for pairing with in2. Re-evaluating in next select cycle.")
                continue // 跳过当前循环，重新进入select
            }

            sum = n1 + n2
            out <- sum
            fmt.Printf("Aggregated %d + %d = %d\n", n1, n2, sum)
        }
    }
    fmt.Println("Aggregator finished processing all channels.")
}

func main() {
    c1 := make(chan int)
    c2 := make(chan int)
    out := make(chan int)

    go numgen(10, 3, c1) // 生成器10，发送3个数字 (10, 20, 30)
    go numgen(1, 3, c2)  // 生成器1，发送3个数字 (1, 2, 3)

    go aggregator(c1, c2, out)

    // 从输出通道读取结果
    for res := range out {
        fmt.Printf("Received sum: %d\n", res)
    }
    fmt.Println("Main goroutine finished.")
}

func aggregator(in1, in2 <-chan int, out chan<- int)