Goroutines、Channels 和死锁

Question

我试图更多地了解 go 的通道和 goroutines，所以我决定制作一个小程序来计算文件中的单词，由bufio.NewScanner对象读取：nCPUs := flag.Int("cpu", 2, "number of CPUs to use")flag.Parse()runtime.GOMAXPROCS(*nCPUs)    scanner := bufio.NewScanner(file)lines := make(chan string)results := make(chan int)for i := 0; i < *nCPUs; i++ {    go func() {        for line := range lines {            fmt.Printf("%s\n", line)            results <- len(strings.Split(line, " "))        }    }()}for scanner.Scan(){    lines <- scanner.Text()}close(lines)acc := 0for i := range results {      acc += i }fmt.Printf("%d\n", acc)现在，在我迄今为止发现的大多数示例中，lines和results通道都将被缓冲，例如make(chan int, NUMBER_OF_LINES_IN_FILE). 尽管如此，在运行此代码后，我的程序存在并显示fatal error: all goroutines are asleep - deadlock!错误消息。基本上我的想法是我需要两个通道：一个将文件中的行与 goroutine 通信（因为它可以是任何大小，我不喜欢认为我需要在make(chan)函数调用中通知大小。其他通道将收集来自 goroutine 的结果，在主函数中我将使用它来计算累积结果。使用 goroutine 和通道以这种方式进行编程的最佳选择应该是什么？任何帮助深表感谢。

Answer 1

正如@AndrewN 指出的那样，问题是每个 goroutine 都到达了它试图发送到results通道的地步，但是这些发送将阻塞，因为results通道没有缓冲并且在for i := range results循环之前没有从它们读取任何内容。你永远不会进入那个循环，因为你首先需要完成for scanner.Scan()循环，它试图将所有的lines发送到lines通道，这是被阻塞的，因为 goroutines 永远不会循环回 ，range lines因为它们被卡在发送到results.

您可能会尝试做的第一件事是将这些scanner.Scan()东西放在一个 goroutine 中，以便可以立即开始读取results通道。但是，您将遇到的下一个问题是知道何时结束for i := range results循环。您想要关闭results通道，但只有在原始 goroutine 完成读取lines通道之后。您可以在关闭results频道后立即关闭lines频道，但是我认为这可能会引入潜在的竞争，因此最安全的做法是在关闭results频道之前等待原始两个 goroutine 完成：（操场链接）：

package main

import "fmt"

import "runtime"

import "bufio"

import "strings"

import "sync"

func main() {

    runtime.GOMAXPROCS(2)

    scanner := bufio.NewScanner(strings.NewReader(`

hi mom

hi dad

hi sister

goodbye`))

    lines := make(chan string)

    results := make(chan int)

    wg := sync.WaitGroup{}

    for i := 0; i < 2; i++ {

        wg.Add(1)

        go func() {

            for line := range lines {

                fmt.Printf("%s\n", line)

                results <- len(strings.Split(line, " "))

            }

            wg.Done()

        }()

    }

    go func() {

        for scanner.Scan() {

            lines <- scanner.Text()

        }

        close(lines)

        wg.Wait()

        close(results)

    }()

    acc := 0

    for i := range results {

        acc += i

    }

    fmt.Printf("%d\n", acc)

}

Answer 2

默认情况下，go 中的通道是无缓冲的，这意味着您生成的匿名 goroutine 都不能发送到结果通道，直到您开始尝试从该通道接收。这不会在主程序中开始执行，直到scanr.Scan()完成填充线路通道......它被阻止执行直到您的匿名函数可以发送到结果通道并重新启动它们的循环。僵局。

您的代码中的另一个问题，即使通过缓冲通道来微不足道地修复上述问题，对于 i := range 结果也将在没有更多结果输入时死锁，因为通道尚未关闭。

编辑：如果您想避免缓冲通道，这是一种潜在的解决方案。基本上，第一个问题是通过一个新的 goroutine执行发送到结果通道来避免的，允许行循环完成。第二个问题（不知道何时停止读取通道）可以通过在创建 goroutine 时对其进行计数并在考虑到每个 goroutine 时明确关闭通道来避免。用等待组做类似的事情可能会更好，但这只是展示如何无缓冲地执行此操作的一种非常快速的方法。