关闭未知长度的通道

Question

当不知道它的
长度时,我无法关闭频道

package main

import (
    "fmt"
    "time"
)

func gen(ch chan int) {
    var i int
    for {
        time.Sleep(time.Millisecond * 10)
        ch <- i
        i++
        // when no more data (e.g. from db, or event stream)
        if i > 100 {
            break
        }
    }

    // hot to close it properly?
    close(ch)
}

func receiver(ch chan int) {
    for i := range ch {
        fmt.Println("received:", i)
    }
}

func main() {
    ch := make(chan int)

    for i := 0; i < 10; i++ {
        go gen(ch)
    }

    receiver(ch)
}

它给了我错误

panic: send on closed channel

goroutine 8 [running]:
main.gen(0xc82001a0c0)
    /home/exu/src/github.com/exu/go-workshops/100-concurrency-channels/16-close-problem.go:12 +0x57
created by main.main
    /home/exu/src/github.com/exu/go-workshops/100-concurrency-channels/16-close-problem.go:35 +0xbd

goroutine 1 [panicwait]:
runtime.gopark(0x0, 0x0, 0x50b8e0, 0x9, 0x10, 0x1)
    /usr/lib/go/src/runtime/proc.go:185 +0x163
runtime.main()
    /usr/lib/go/src/runtime/proc.go:121 +0x2f4
runtime.goexit()
    /usr/lib/go/src/runtime/asm_amd64.s:1696 +0x1

goroutine 6 [sleep]:
time.Sleep(0x989680)
    /usr/lib/go/src/runtime/time.go:59 +0xf9
main.gen(0xc82001a0c0)
    /home/exu/src/github.com/exu/go-workshops/100-concurrency-channels/16-close-problem.go:11 +0x29
created by main.main
    /home/exu/src/github.com/exu/go-workshops/100-concurrency-channels/16-close-problem.go:33 +0x79

goroutine 7 [sleep]:
time.Sleep(0x989680)
    /usr/lib/go/src/runtime/time.go:59 +0xf9
main.gen(0xc82001a0c0)
    /home/exu/src/github.com/exu/go-workshops/100-concurrency-channels/16-close-problem.go:11 +0x29
created by main.main
    /home/exu/src/github.com/exu/go-workshops/100-concurrency-channels/16-close-problem.go:34 +0x9b
exit status 2

这是合乎逻辑的 - 当第二个试图发送给它时,第一个goroutine关闭通道.在这种情况下关闭渠道的最佳方法是什么？

Answer 1

关闭频道后,您无法在其上发送更多值,否则会发生恐慌.这就是您的体验.

这是因为您启动多个使用相同通道的goroutine并在其上发送值.然后你关闭每个频道.而且由于它们不同步,一旦第一个goroutine到达它关闭它的点,其他人可能(并且他们将)继续发送它的值:恐慌!

您只能关闭一次通道(尝试关闭已关闭的通道也会发生混乱).当你在其上发送所有值的goroutine完成时,你应该这样做.为此,您需要检测何时完成所有发送方goroutine.检测这种情况的惯用方法是使用sync.WaitGroup.

对于每个已启动的发送者goroutine,我们将1添加到WaitGroup使用中WaitGroup.Add().发送值的每个goroutine都可以通过调用发出信号WaitGroup.Done().最好将此作为延迟声明,所以如果你的goroutine突然终止(例如恐慌),WaitGroup.Done()仍会被调用,并且不会让其他goroutines挂起(等待赦免 - 一个WaitGroup.Done()永远不会来的"失踪" 电话...... ).

而WaitGroup.Wait()会等到所有发件人够程完成后,只有在此之后,只有一次将它关闭通道.我们想要检测这个"全局"完成事件并关闭通道,同时处理它上面发送的值正在进行中,所以我们必须在它自己的goroutine中执行此操作.

接收器goroutine将一直运行,直到通道关闭,因为我们for ... range在通道上使用了构造.并且由于它在主goroutine中运行,因此程序将不会退出,直到从通道正确接收和处理所有值.所述for ... range构建体循环,直到接收到所有的值的信道被关闭之前已被发送.

请注意,下面的解决方案也可以使用缓冲和非缓冲通道而无需修改(尝试使用缓冲通道ch := make(chan int, 100)).

正确的解决方案(在Go Playground尝试):

func gen(ch chan int, wg *sync.WaitGroup) {
    defer wg.Done()
    var i int
    for {
        time.Sleep(time.Millisecond * 10)
        ch <- i
        i++
        // when no more data (e.g. from db, or event stream)
        if i > 100 {
            break
        }
    }
}

func receiver(ch chan int) {
    for i := range ch {
        fmt.Println("received:", i)
    }
}

func main() {
    ch := make(chan int)
    wg := &sync.WaitGroup{}

    for i := 0; i < 10; i++ {
        wg.Add(1)
        go gen(ch, wg)
    }

    go func() {
        wg.Wait()
        close(ch)
    }()

    receiver(ch)
}

注意:

请注意,receiver(ch)在主goroutine 中运行WaitGroup以及在其自己的(非主要)goroutine中等待并关闭通道的代码非常重要; 而不是相反.如果要切换这两个,可能会导致"提前退出",即并非所有值都可能从通道接收和处理.这是因为当主goroutine完成时,Go程序退出(规范:程序执行).它不会等待其他(非主要)goroutines完成.因此,如果等待和关闭通道将在主goroutine中,在关闭通道之后,程序可以随时退出,而不是等待另一个goroutine,在这种情况下将循环以从通道接收值.