Goroutine、渠道和僵局

Question

我试图更多地了解Go的通道和goroutine，所以我决定制作一个小程序，从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 := 0
for i := range results {
      acc += i
 }

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

现在，在我发现的大多数示例中，行和结果通道都将被缓冲，比如make(chan int,NUMBER_OF_LINES_IN_FILE)。但是，在运行这段代码之后，我的程序仍然存在一个致命错误：所有goroutines都是睡眠-死锁！错误消息。

基本上，我认为我需要两个通道：一个通道将文件中的行与goroutine通信（因为它可以是任意大小的，所以我不认为我需要在make(chan)函数调用中通知大小）；另一个通道将从goroutine收集结果，并在main函数中使用它来计算累积结果。

用goroutines和channels以这种方式编程的最佳选择应该是什么？任何帮助都是非常感谢的。

Answer 1

正如@Andrewn所指出的，问题是每个goroutine都试图发送到results通道，但是这些发送将会阻塞，因为results通道没有缓冲，并且在for i:=range results循环之前不会从它们中读取任何内容。您永远不会到达那个循环，因为首先需要完成scanner.scan()循环，该循环试图将所有行发送到行通道，该通道被阻塞，因为goroutines永远不会循环回范围行，因为它们无法发送到结果。

解决这个问题的第一件事是将scanner.scan()放在goroutine中，这样就可以立即读取results通道。但是，您将面临的下一个问题是知道何时结束i:=range results循环的。您希望关闭results通道，但只有在原始goroutines读取lines通道之后。您可以在关闭行通道后立即关闭Results通道，但是我认为这可能会引入潜在的竞争，因此最安全的做法也是在关闭Results通道之前等待原始的两个goroutines完成：（playground链接）：

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)
}