关闭完成通道后 goroutine 缺少打印

Question

当我注意到并非管道中的所有关闭打印都被打印时，我正在根据“Go 中的并发”一书中的示例运行以下代码。
看到“完成倍增！”不见了。
另一方面，NumGoroutine() 只显示 main func 正在运行。
以下代码有什么问题？ (https://play.golang.org/p/tkFgvKboVgS)

package main

import (
    "fmt"
    "runtime"
    "time"
)

func main() {
    generator := func(done <-chan struct{}) <-chan int {
        intStream := make(chan int)
        i:=0
        go func() {
            defer close(intStream)
            for {
                select {
                case <-done:
                    fmt.Println("done generator!")
                    return
                case intStream <- i:
                    time.Sleep(1 * time.Second)
                    i++
                }
                fmt.Println("generator after select")
            }
        }()
        return intStream
    }

    multiply := func(
        done <-chan struct{},
        intStream <-chan int,
        multiplier int,
    ) <-chan int {
        multipliedStream := make(chan int)
        go func() {
            defer close(multipliedStream)
            for i := range intStream {
                select {
                case <-done:
                    fmt.Println("done multiply !")
                    return
                case multipliedStream <- i * multiplier:
                }
                fmt.Println("multiply after select")
            }
        }()
        return multipliedStream
    }
    add := func(
        done <-chan struct{},
        intStream <-chan int,
        additive int,
    ) <-chan int {
        addedStream := make(chan int)
        go func() {
            defer close(addedStream)
            for i := range intStream {
                select {
                case <-done:
                    fmt.Println("done add !")
                    return
                case addedStream <- i + additive:
                }
                fmt.Println("add after select")
            }
        }()
        return addedStream
    }

    done := make(chan struct{})

    intStream := generator(done)
    pipeline := add(done, multiply(done, intStream, 2), 2)
    go func() {
        time.Sleep(3 * time.Second)
        close(done)
        fmt.Println("Closed done")
    }()
    for v := range pipeline {
        fmt.Println(v)
    }
    fmt.Println("finished iterating pipeline")
    time.Sleep(10 * time.Second)
    fmt.Println("ramaining goroutines:", runtime.NumGoroutine())
    fmt.Println("finished!")
}

输出：

add after select
2
multiply after select
generator after select
multiply after select
add after select
4
generator after select
multiply after select
add after select
6
generator after select
Closed done
multiply after select
done add !
finished iterating pipeline
generator after select
done generator!
ramaining goroutines: 1
finished!

Answer 1

有些代码路径不会打印某些done 消息。调度程序碰巧选择了一个不打印multiply 的那个。如果您稍微更改一下代码（例如，在与现在不同的实例上登录），您会发现它也可能会错过 add done 消息。 (https://play.golang.org/p/meEPM5GR9Rr)。原因如下：

如果done 消息在生成器将数字写入通道并且乘法器读取它之后立即到达，那么乘法器会看到done 可用并选择它。当multiplier 打印done 消息时就是这种情况。如果 done 消息在 multiplier 在 for 循环中等待时到达，则 multiplier 将接收输入通道（而不是 done 通道）的关闭，导致 for 循环终止而不打印 done 消息。

出现问题是因为您正在 for 循环中从通道读取，然后进行选择。在等待 for 循环从通道中读取数据时，不会评估任何与 select 相关的事件。

解决这个问题的更好方法是不使用 for 循环从通道中读取。例如：

for {
     select {
        case <-done:
           return
        case i, ok:= <-intstream:
           if !ok {
              return
           }
           select {
               case <- done:
                    return
               case addedStream <- i + additive:
           }
     }
}

Answer 2

您的add 和multiply 例程不是永远循环，而是for ... range 循环。因此，在每个循环的顶部，它们等待下一个整数，而不是在select 中等待，后者要么从done 接收关闭消息，要么将结果发送到它们的流。这不是一个问题，但它意味着如果他们的输入流是关闭，他们将返回而不进入循环本身。

如果我add fmt.Println calls to expose the point at which they exit due to reaching the end of their input stream，行为会发生轻微变化（可能是由于时间的原因；我没有费心去解释它，Burak Serdar posted his answer 在我打字的时候已经）并且输出变为：

add after select
2
multiply after select
generator after select
multiply after select
add after select
4
generator after select
multiply after select
add after select
6
generator after select
Closed done
done multiply !
add got end of stream - done!
finished iterating pipeline
generator after select
done generator!
ramaining goroutines: 1
finished!

通常更合理的做法是仅生成器本身接收done 信号，并使流水线函数始终写入其所有结果，这使得它们更可预测。当然，读取每个管道的人必须读到最后——但你已经在主 goroutine 中这样做了，所以我们只是在整个过程中传播它。 Here 是您的代码的简化版本，它就是这样做的；它输出：

2
generator after select
4
generator after select
6
generator after select
Closed done
8
generator after select
done generator!
multiply got end of stream - done!
add got end of stream - done!
finished iterating pipeline
remaining goroutines: 1

请注意，这一次，我们从最终生成的值 (3) 中得到最终的计算值 (8)。