Goroutines 节流示例

Question

我正在通过 Udemy 课程学习基本的围棋。在 goroutines 部分，有一个节流的例子，它让我理解了等待组是如何工作的。package mainimport (    "fmt"    "math/rand"    "sync"    "time")func main() {    c1 := make(chan int)    c2 := make(chan int)    go populate(c1)    go fanOutIn(c1, c2)    for v := range c2 {        fmt.Println(v)    }    fmt.Println("about to exit")}func populate(c chan int) {    for i := 0; i < 100; i++ {        c <- i    }    close(c)}func fanOutIn(c1, c2 chan int) {    var wg sync.WaitGroup    const goroutines = 10    wg.Add(goroutines)     for i := 0; i < goroutines; i++ {        go func() {            for v := range c1 {                func(v2 int) {                    c2 <- timeConsumingWork(v2)                }(v)            }            wg.Done()        }()    }    wg.Wait()    close(c2)}func timeConsumingWork(n int) int {    time.Sleep(time.Microsecond * time.Duration(rand.Intn(500)))    return n + rand.Intn(1000)}不符合我理解的部分是在fanOutIn我们设置WaitGroup, 和Add(10).为什么我打印出 100 个值？只能将单个值 ( i := 0) 放入c1，并且该值永远不会从通道中显式删除。然后代码命中wg.Done()，等待组队列减少到 9，依此类推。根据我目前的理解，我希望看到 10 个值0 + rand.Intn(1000)。

Answer 1

分拆出来的函数如下所示（包括go前面的 和调用它的括号）：

go func() {

    for v := range c1 {

        func(v2 int) {

            c2 <- timeConsumingWork(v2)

        }(v)

    }

    wg.Done()

}()

这段代码有点奇怪和离奇。让我们进一步缩小它，丢弃wg.Done并只保留for循环本身：

for v := range c1 {

    func(v2 int) {

        c2 <- timeConsumingWork(v2)

    }(v)

}

有一个内部未命名的函数在这里非常没用；我们可以在不改变程序行为的情况下丢弃它，得到：

for v := range c1 {

    c2 <- timeConsumingWork(v)

}

这最后是一个简单的循环。现在的一个关键问题是：您期望从这个循环中进行多少次迭代？ 注意：它不一定是任何常数。或许更好的方式来表达这个问题是：这个循环什么时候结束？

for循环读取一个通道。这种循环在从通道读取指示没有更多数据时结束，即通道已关闭且其队列为空。（请参阅循环的 Go 规范部分for。）

所以这个最里面的循环，for v := range c1直到通道c1关闭并且队列中没有更多数据时才会终止。该频道是使用以下内容创建的：

c1 := make(chan int)

所以它没有队列，所以我们甚至不需要考虑这一点：它在 aclose(c1)关闭它后终止。 您现在应该寻找一个closecloses c1。

我们的近在哪里？

这是关闭的地方c1：

func populate(c chan int) {

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

        c <- i

    }

    close(c)

}

我们称之为 withc1作为它的参数，所以它的 final close(c)closes c1。现在你可以问：我们什么时候接到这个close电话？ 答案很明显：i >= 100在循环之后，即，在我们将 100 个值（分别从 0 到 99）发送到 channel 之后c1。

什么fanOutIn是剥离 10 个 goroutine。10 个 goroutine 中的每一个都运行我上面引用的第一个匿名函数。该匿名函数有一个循环，运行次数不定，一直重复直到通道c1关闭。循环中的每次行程都会获取通道的值，因此最初，如果十个 goroutine 都设法在有任何可用值之前启动，那么所有十个 goroutine 都将等待值。

当生产者函数将一个值放入通道时，十个等待的 goroutine 中的一个将获取它并开始使用它。如果该 goroutine 需要很长时间才能回到自己for循环的顶部，则另一个 goroutine 将获取下一个生成的值。所以这里发生的情况是，多达 10 个生成的值通过通道传播到多达 10 个 goroutine。1 这些（最多 10 个）goroutine 中的每一个都花费了一些重要的时间来使用它的值，然后将最终产品值发送到通道c2并返回到它自己的无限for循环的顶部。

只有当生产者关闭了它的通道c（这里是我们的c1）时，十个 goroutine 才会看到一个关闭的通道空队列，从而允许它们退出for循环。当他们退出他们的for循环时，他们每个人都会调用wg.Done()（一次）并终止。

因此，一旦close(c1)发生（通过close(c)in populate），最终所有十个匿名 goroutine 都将调用wg.Done(). 届时，wg.Wait()infanOutIn将返回。这将调用close(c2)并从 中返回fanOutIn，同时终止该goroutine。

同时，在 中main，我们使用for v := range c2从通道中读取c2。当十个 goroutine 中的任何一个for写入值时，此循环将运行。c2只有当它自己关闭时它才会退出c2（它的队列也必须是空的，但又c2是一个零长度队列）。所以在关闭之前main不会继续通过for循环，直到返回才会发生，直到发生十次调用才会发生，直到通道关闭才会发生。c2wg.Wait()wg.Done()c1

这意味着在调用之前main无法通过自己的for循环，并且只有在生成恰好 100 个值之后才会发生这种情况。populateclose(c)