我做了一个简单的代码示例来理解pipeline的用法，就在这里。

package main

import (
    "fmt"
    "sync"
    "time"
)

func main() {
    ch1 := make(chan int, 10) // Use buffered channel so as to avoid clogging
    ch2 := make(chan string, 10)
    var wg sync.WaitGroup
    for i := 0; i < 3; i++ {
        wg.Add(1)
        go func1(i, ch1, &wg)
        go func2(ch1, ch2)
    }
    wg.Wait()
    close(ch1)
    for val := range ch2 {
        fmt.Println(val)
    }
}

func func1(seconds int, ch chan<- int, wg *sync.WaitGroup) {
    defer wg.Done()
    time.Sleep(time.Duration(seconds) * time.Second)
    fmt.Println(seconds)
    ch <- seconds
}

func func2(ch1 chan int, ch2 chan string) {
    for range ch1 {
        ch2 <- "hello"
    }
    close(ch2)
}

现在，问题是我没有得到一致的输出(我知道这是一些并发问题，我还没有完全理解)。

输出

> go run pipeline-loop.go 
0
1
2
hello
hello

> go run pipeline-loop.go 
0
1
2
hello
hello
hello

> go run pipeline-loop.go 
0
1
2
hello
hello

> go run pipeline-loop.go 
0
1
2
hello
hello

> go run pipeline-loop.go 
0
1
2
hello
hello
panic: close of closed channel

goroutine 6 [running]:
main.func2(0xc00006c000, 0xc000056180)
    /home/projects/go-tuts/pipeline-loop.go:36 +0x72
created by main.main
    /home/projects/go-tuts/pipeline-loop.go:16 +0x10f
exit status 2

另一个人更改了代码(并且它正在运行)并将 func2 放在循环之外，但我希望 func1 的每次迭代都使用 func2 >.

问题

所以，我想了解应该在哪里使用 WaitGroup 和 close(ch) ？

谢谢。
暂时的
(一个 golang 菜鸟)

更新

根据用户的回答，我更改了代码，现在我得到了预期的输出(但不是这个问题的解决方案)，但仍然存在死锁。 https://play.golang.org/p/O_rp_FLvNh8

package main

import (
    "fmt"
    "sync"
    "time"
)

func main() {
    ch1 := make(chan int, 10) // Use buffered channel so as to avoid clogging
    ch2 := make(chan string, 10)
    var wg1 sync.WaitGroup
    // var wg2 sync.WaitGroup
    for i := 0; i < 3; i++ {
        wg1.Add(1)
        go func1(i, ch1)
        go func2(ch1, ch2, &wg1)
    }
    for val := range ch2 {
        fmt.Println(val)
    }
    wg1.Wait()
    close(ch1)
    close(ch2)
}

// func func1(seconds int, ch chan<- int, wg *sync.WaitGroup) {
func func1(seconds int, ch chan<- int) {
    // defer wg.Done()
    time.Sleep(time.Duration(seconds) * time.Second)
    fmt.Println(seconds)
    ch <- seconds
}

func func2(ch1 chan int, ch2 chan string, wg *sync.WaitGroup) {
    defer wg.Done()
    for range ch1 {
        ch2 <- "hello"
    }
}

最佳答案

您的代码中存在多个问题。

在循环中，您生成了多个运行 func2 的 goroutine，在 func2 中，您将数据发送到 ch2 并且调用 close(ch2)。这是个问题。当一个 goroutine 将数据发送到 ch2 时，另一个 goroutine 可能会关闭该 channel ，这会导致:

panic: close of closed channel

goroutine 6 [running]:
main.func2(0xc00006c000, 0xc000056180)
    /home/projects/go-tuts/pipeline-loop.go:36 +0x72
created by main.main
    /home/projects/go-tuts/pipeline-loop.go:16 +0x10f
exit status 2

一般来说，您不需要多次关闭 channel - 您只需要在它们全部完成后关闭它们。为此，您需要另一个 WaitGroup；您需要将这两个函数都传递给 WaitGroup。

进一步阅读:https://blog.golang.org/pipelines

更新:

我个人使用一种“工作”模式，将数据生成到同一个 channel 中，并且在所有工作完成后需要关闭该 channel :

for something {
    wg.Add(1)
    go func(i int) {
        work(ch)
        wg.Done()
    }
}

go func() {
    wg.Wait()
    close()
}()

我认为让 WorkGroup 中的 API 保持干净是个好主意，因为 WorkGroup 是关于如何同步工作而不是如何完成工作。

我已将您的代码更改为这种模式:https://play.golang.org/p/vdCNsxWhgyQ

关于go - 我应该在哪个函数中传递 WaitGroup？，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/54624442/