带有 sync.WaitGroup 的 Goroutine 在最后一个 wg.Done() 之前结束

Question

我有一个示例代码（您可以在Go Playground 上找到它）：

package main

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

func main() {
    messages := make(chan int)
    var wg sync.WaitGroup
    var result []int

    // you can also add these one at 
    // a time if you need to 

    wg.Add(1)
    go func() {
        defer wg.Done()
        time.Sleep(time.Second * 1)
        messages <- 1
    }()
    wg.Add(1)
    go func() {
        defer wg.Done()
        time.Sleep(time.Second * 1)
        messages <- 2
    }() 
    wg.Add(1)
    go func() {
        defer wg.Done()
        time.Sleep(time.Second * 1)
        messages <- 3
    }()
    go func() {
        for i := range messages {
            fmt.Println(i)
        result = append(result, i)
        }

    }()

    wg.Wait()
    fmt.Println(result)
}

我得到了这个输出：

2
1
[2 1]

我想我知道它为什么会发生，但我无法解决它。 WaitGroup 中有 3 个项目，我的意思是三个 goroutine，第 4 个 groutine 使用来自通道的数据。当最后一个 groutine 说 wg.Done() 时，程序结束了，因为 wg.Wait() 表示每个 goroutine 都完成了，最后一个 goroutine 结果第四个 goroutine 不能消费，因为程序结束了。我尝试在第四个函数中使用 wg.Add(1) 和 wg.Done() 加一，但在这种情况下我遇到了死锁。

Answer 1

关闭通道是一种惯用的 Go 信号模式，如果您关闭缓冲通道，消费者可以读取所有排队的数据然后停止。

这段代码可以正常工作：

func main() {
    messages := make(chan int)
    var wg sync.WaitGroup
    var result []int

    // you can also add these one at
    // a time if you need to

    wg.Add(1)
    go func() {
        defer wg.Done()
        time.Sleep(time.Second * 1)
        messages <- 1
    }()
    wg.Add(1)
    go func() {
        defer wg.Done()
        time.Sleep(time.Second * 1)
        messages <- 2
    }()
    wg.Add(1)
    go func() {
        defer wg.Done()
        time.Sleep(time.Second * 1)
        messages <- 3
    }()

    // this goroutine added to signal end of data stream
    // by closing messages channel
    go func() {
        wg.Wait()
        close(messages)
    }()

    // if you need this to happen inside a go routine,
    // this channel is used for signalling end of the work,
    // also another sync.WaitGroup could be used, but for just one
    // goroutine, a single channel as a signal makes sense (there is no
    // groups)
    done := make(chan struct{})
    go func() {
        defer close(done)
        for i := range messages {
            fmt.Println(i)
            result = append(result, i)
        }
    }()

    <-done
    fmt.Println(result)
}

如您所见，我们刚刚添加了另一个 goroutine，它在所有生产者完成后关闭 messages 通道。

Answer 2

kostix的回答是正确的，直到他们提到

一个简单的解决方法是为该 goroutine 添加 wg.Add(1) 并在其中延迟 wg.Done()。

如果没有关闭消息通道，这将导致您的循环永远不会结束！因此，主 goroutine 将在您最后一个“收集”goroutine 完成之前再次完成。如果将一个 goroutine 绑定到您的 wg WaitGroup ，您还会收到一个错误，该 goroutine 永远不会发送 Done() 信号。

当他们提到时又一次

更好的解决方案是在wg.Wait() 之后和从切片读取之前close() 消息通道

他们建议的展示位置将再次给您同样的错误，因为您将在同一个 WaitGroup wg 上等待。虽然您的最后一个“收集”goroutine 将继续在您的 messages 频道中寻找更多消息，并且永远不会到达延迟的 wg.Done()

然后Alex Yu 的评论通过在完全阅读结果之前等待修复它，这是一个很好的修复。但是，如果您希望您的收集 goroutine 立即开始，而不是等待所有以前的 goroutine（写入 messages 频道）完成，然后再开始从所述频道读取，我建议以下...

创建一个结果 WaitGroup，Add(1)，然后再开始你的最后一个“收集”goroutine，defer wgResult.Done() 在你最后一个“收集”goroutine 中，然后在最后，在你的wg.Wait() 和你的fmt.Println(result) 之间，你应该@ 987654336@和wgResult.Wait()。

这允许您的所有 goroutine 尽快启动，并且仅在需要时等待写入 goroutine 以及读取 goroutine。

这是建议解决方案的 GoPlayground 链接

https://play.golang.org/p/na0JS1HTwNP

Answer 3

你生成的最后一个 goroutine——那个用来收集结果的 goroutine——没有被main() 等待，所以wg.Wait() 在那里返回，main() 退出并收获剩余的 goroutine。 据说那时只有一个收集的 goroutine 仍然存在，但它无法更新切片。

还请注意，由于同样的原因，您的程序中存在数据竞争：当main() 读取结果片段时，它不知道读取它是否安全—也就是说，作者是否写完。

一个简单的解决方法是为该 goroutine 添加 wg.Add(1) 并在其中添加 defer wg.Done()。

更好的解决方案是close()messageswg.Wait() 之后的频道 在从切片中读取之前。这将使收集 goroutine 的 range 循环终止，这还将在该 goroutine 和 main() 之间创建一个适当的同步点。