推迟关闭多个 Goroutines？

Question

我有以下代码：

package main

import (
    "fmt"
    "time"
)

func main() {
    t := time.Now()
    stuff := fanIn(
        generator(4, 5, 6, 7),
        generator(1, 2, 6, 3, 7),
        generator(12, 15, 33, 40, 10),
        generator(18, 13, 20, 40, 15),
        generator(100, 200, 64000, 3121, 1237),
    )

    for v := range stuff {
        fmt.Println(v)
    }

    fmt.Println(t.Sub(time.Now()))
}

func generator(nums ...int) <-chan int {
    out := make(chan int, 10)
    go func() {
        defer close(out)
        for _, v := range nums {
            out <- v
        }
    }()
    return out
}

func fanIn(in ...<-chan int) <-chan int {
    out := make(chan int, 10)

    for _, v := range in {
        go func(ch <-chan int) {
            for val := range ch {
                go func(c int) { out <- c }(val)
            }
        }(v)
    }

    return out
}

它导致第 18 行出现死锁：

for v := range stuff {...}

问题（我认为）是我没有推迟关闭返回只读通道的 fanIn 函数。我不知道什么时候推迟它，因为它必须等待多个 goroutine 结束才能完成。

解决这个僵局的惯用方法是什么？这段代码甚至是惯用的吗？

谢谢！

去玩

Answer 1

fanIn您关于未关闭通道的错误原因是正确的。您可以使用 async.WaitGroup来解决该问题：

func fanIn(in ...<-chan int) <-chan int {
    // use a WaitGroup here
    var wg sync.WaitGroup
    out := make(chan int, 10)

    for _, v := range in {
        wg.Add(1)
        go func(ch <-chan int) {
            defer wg.Done()
            for val := range ch {
                out <- val
            }
        }(v)
    }

    // wait for wait groups to finish in another goroutine
    go func() {
        wg.Wait()
        close(out)
    }()
    return out
}

工作代码。