为什么这个 golang 脚本让我陷入僵局？ + 几个问题

Question

我从 github 上的某个人那里得到了这段代码，我正在尝试使用它来理解并发性。

package main

import (
    "bufio"
    "fmt"
    "os"
    "sync"
    "time"
)

var wg sync.WaitGroup

func sad(url string) string {
    fmt.Printf("gonna sleep a bit\n")
    time.Sleep(2 * time.Second)
    return url + " added stuff"
}

func main() {
    sc := bufio.NewScanner(os.Stdin)
    urls := make(chan string)
    results := make(chan string)

    for i := 0; i < 20; i++ {
        wg.Add(1)
        go func() {
            defer wg.Done()
            for url := range urls {
                n := sad(url)
                results <- n
            }
        }()
    }

    for sc.Scan() {
        url := sc.Text()
        urls <- url
    }

    for result := range results {
        fmt.Printf("%s arrived\n", result)
    }

    wg.Wait()
    close(urls)
    close(results)
}

我有几个问题：

1. 为什么这段代码让我陷入僵局？
2. for 循环如何存在在从用户接收输入的操作之前，go 例程是否等到任何内容通过 urls 通道然后开始工作？我不明白这一点，因为它不是顺序的，例如为什么从用户那里获取输入然后将每个输入都放在 urls 通道中然后运行 ​​go 例程被认为是错误的？
3. 在 for 循环中，我有 另一个循环，它在 urls 通道上进行迭代，每个 go 例程是否只处理一行输入？还是一个 goroutine 一次处理多行？这是如何工作的？
4. 我在这里收集的输出是否正确？

Answer 1

大多数情况下，您都在正确地做事，但有些事情有点乱。 for sc.Scan() 循环将一直持续到 Scanner 完成，而 for result := range results 循环将永远不会运行，因此没有 go 例程（在本例中为“main”）将能够从 results 接收。在运行您的示例时，我在 for sc.Scan() 之前启动了 for result := range results 循环，并且还在它自己的 go 例程中启动了 - 否则将永远无法到达 for sc.Scan()。

go func() {
    for result := range results {
        fmt.Printf("%s arrived\n", result)
    }
}()

for sc.Scan() {
    url := sc.Text()
    urls <- url
}

另外，因为你在 close(urls) 之前运行 wg.Wait()，所以主 goroutine 被阻塞，等待 20 个 sad() goroutine 完成。但是直到调用close(urls) 才能完成。因此，在等待等待组之前关闭该通道。

close(urls)
wg.Wait()
close(results)

Answer 2

我对以前的答案不太满意，所以这里有一个基于go tour、go doc、specifications 中记录的行为的解决方案。

package main

import (
    "bufio"
    "fmt"
    "strings"
    "sync"
    "time"
)

var wg sync.WaitGroup

func sad(url string) string {
    fmt.Printf("gonna sleep a bit\n")
    time.Sleep(2 * time.Millisecond)
    return url + " added stuff"
}

func main() {
    // sc := bufio.NewScanner(os.Stdin)
    sc := bufio.NewScanner(strings.NewReader(strings.Repeat("blah blah\n", 15)))
    urls := make(chan string)
    results := make(chan string)

    for i := 0; i < 20; i++ {
        wg.Add(1)
        go func() {
            defer wg.Done()
            for url := range urls {
                n := sad(url)
                results <- n
            }
        }()
    }
    // results is consumed by so many goroutines
    // we must wait for them to finish before closing results
    // but we dont want to block here, so put that into a routine.
    go func() {
        wg.Wait()
        close(results)
    }()

    go func() {
        for sc.Scan() {
            url := sc.Text()
            urls <- url
        }
        close(urls) // done consuming a channel, close it, right away.
    }()

    for result := range results {
        fmt.Printf("%s arrived\n", result)
    } // the program will finish when it gets out of this loop.
    // It will get out of this loop because you have made sure the results channel is closed.

}

Answer 3

for 循环创建 20 个 goroutine，所有等待来自 urls 通道的输入。当有人向这个通道写入数据时，其中一个 goroutine 会拾取它并在其中工作。这是一个典型的工作池实现。

然后，扫描器逐行读取输入，并将其发送到urls 通道，其中一个goroutines 将接收它并将响应写入results 通道。此时，没有其他 goroutine 从 results 通道读取，所以这将阻塞。

当扫描器读取 URL 时，所有其他 goroutine 将拾取并阻止它们。所以如果扫描器读取的 URL 超过 20 个，它就会死锁，因为所有的 goroutine 都在等待一个读取器。

如果 URL 少于 20 个，扫描器 for 循环将结束，并读取结果。然而这最终也会死锁，因为当通道关闭时for循环将终止，并且没有人关闭通道。

要解决此问题，首先，在阅读完毕后立即关闭urls 频道。这将释放 goroutine 中的所有 for 循环。然后你应该把从results 通道读取的for循环放入一个goroutine，这样你就可以在处理结果时调用wg.Wait。在wg.Wait之后，可以关闭results频道。

这并不能保证results 频道中的所有项目都会被读取。程序可能会在所有消息被处理之前终止，因此请使用您在 goroutine 结束时关闭的第三个通道，该通道从 results 通道读取。那就是：

done:=make(chan struct{})
go func() {
  defer close(done)
  for result := range results {
        fmt.Printf("%s arrived\n", result)
    }
}()
wg.Wait()
close(results)
<-done