为什么我的 "done" 频道随机关闭？

Question

我构建了以下 go 代码。

想法是构建一个done通道和一个生成int通道的生成器。 Link 它们在 2 级管道中 chanNumbers := pipeb(done, pipea(done, gen(done)))

几秒钟后，取消已完成的频道。我希望看到生成器和管道的两个阶段取消和 return，但是文本 "PipeX is terminating now" 只是随机出现，我真的不明白为什么。有人有想法吗？

package main
import (
    "fmt"
    "time"
)

func gen(done <-chan interface{}) <-chan int {
    ret := make(chan int)
    cx := 0

    go func() {
        for {
            select {
            case <-done:
                fmt.Println("**Generator Terminates now")
                time.Sleep(2 * time.Second)
                fmt.Println("**Generator has terminated now")
                close(ret)
                return
            case ret <- cx:
                fmt.Printf("Gen : we push %d \n", cx)
                cx = cx + 1
            }
        }
    }()
    fmt.Println("Generator has created and returned its channel")
    return ret
}

func pipea(done <-chan interface{}, in <-chan int) <-chan int {
    ret := make(chan int)

    go func() {
        for {
            select {
            case <-done:
                fmt.Println("**pipeA terminates")
                time.Sleep(2 * time.Second)
                fmt.Println("**pipeA has terminated now")
                close(ret)
                return
            case tmp, ok := (<-in):
                if ok {
                    fmt.Printf("pipeA : we push %d \n", tmp)
                    ret <- tmp
                } else {
                    in = nil
                }
            }
        }
    }()
    return ret
}

func pipeb(done <-chan interface{}, in <-chan int) <-chan int {
    ret := make(chan int)

    go func() {
        for {
            select {
            case <-done:
                fmt.Println("**pipeB terminates")
                time.Sleep(2 * time.Second)
                fmt.Println("**pipeB has terminated now")
                close(ret)
                return
            case tmp, ok := (<-in):
                if ok {
                    fmt.Printf("pipeB : we push %d \n", tmp)
                    ret <- tmp
                } else {
                    in = nil
                }
            }
        }
    }()
    return ret
}

func main() {
    done := make(chan interface{})

    chanNumbers := pipeb(done, pipea(done, gen(done)))
    go func() {
        time.Sleep(2 * time.Second)
        close(done)
    }()

forloop:
    for {
        select {
        case n := <-chanNumbers:
            fmt.Printf("Received in main element : %d\n", n)
        case <-done:
            break forloop
        }
    }

    //end of the main program
    fmt.Println("Sleeping some seconds before termiating")
    time.Sleep(8 * time.Second)
    fmt.Println("exit...")
}

Answer 1

你有四个 go-routines 运行:

gen，您的生成器，正在写入无缓冲输出通道，直到 done
pipeA，从 gen 读取，写入无缓冲输出通道，直到 done
pipeB，从 pipeA 读取，写入无缓冲输出通道，直到 done
main，从 pipeB 读到 done

现在，当您关闭 done 时，这完全取决于 go-routines 看到它的顺序。

如果main第一个看到done关闭，它会打破for循环并停止从pipeB消费。但如果 pipeB 仍在尝试写入输出通道 (ret <- tmp)，它将阻塞在那里；所以它永远不会到达 <- done 部分。

有两个选项可以解决此问题：

在你的生成器中只听 done，让其他 go-routines 使用 for n := range in { }。
将您的发送逻辑也放在 select 中，这样您的生成器和管道就可以检测到 done 何时关闭。

或者，您可能想使用缓冲输出通道，但即使这样，这个问题仍然会发生。

为什么我的 "done" 频道随机关闭？

Why is my "done" channel closing randomly?

channel

go