使用 goroutines 和 channel 的并发受限消费者

Question

我试图重现“管理资源的方法是启动固定数量的句柄 goroutine，所有这些都从请求通道读取。”从Effective Go 发现

fatal error: all goroutines are asleep - deadlock!

这个想法很简单：有 1 个队列和 1 个结果通道以及几个有限数量的“工人”。

我的代码是in Go Playground

queue := make(chan *Request)
result := make(chan int)
quit := make(chan bool)
go Serve(queue, quit)
for i := 0; i < 10; i++ {
    req := Request{i, result}
    queue <- &req
}
close(queue)
for i := 0; i < 10; i++ {
    fmt.Printf("Finished %d\n", <-result)
}
fmt.Printf("All finished\n")
quit <- true

功能服务：

func handle(queue chan *Request) {
    for r := range queue {
        //process(r)
        fmt.Printf("Processing %d\n", r.i)
        r.r <- r.i

    }
}

func Serve(clientRequests chan *Request, quit chan bool) {
    MaxOutstanding := 2
    // Start handlers
    for i := 0; i < MaxOutstanding; i++ {
        go handle(clientRequests)
    }
    <-quit // Wait to be told to exit.
}

怎么了？或者可能有更简单的解决方案来实现处理请求的有限数量的工作人员？

Answer 1

结果通道没有缓冲，所以每次发送都会阻塞 goroutine，直到有东西收到结果。因此，在处理完前两项后，这两个处理程序例程正在等待接收它们的结果。但是接收工作项的代码还没有运行，所以它们被卡住了。然后 main() 尝试发送下一个工作项，而处理程序还没有准备好接收它，所以 main() 现在也卡住了。

解决此问题的一种方法是在对工作项进行排队之前启动一个 goroutine 在后台接收结果。这是您在后台接收结果的代码版本，另外还让main() 等到所有结果都被接收和处理：http://play.golang.org/p/_CKn3CxQFc。

在您的示例中，您只需计算收到的结果即可知道何时可以安全退出。但是，如果出现需要弄清楚何时完成并且计数还不够的情况，那么 1) 每个工人都可以在完成时发出信号，2) Serve 可以在所有工人发出信号后关闭结果通道， 3) 当所有结果都处理完毕后，你的结果处理 goroutine 可以发送一个quit 到main()。有很多变化。你可以在http://play.golang.org/p/12wZbm0rxa看到代码