我有三个并发运行的 goroutine。其中两个进行一些处理并将其结果发送到结果通道。第三个 goroutine 通过读取结果通道来“统计”结果。我可以使用 waitgroup 等待两个计算 goroutine 完成，然后遍历结果通道来统计结果，但这无法扩展，并且需要我创建一个具有巨大缓冲区大小的缓冲结果通道，这是不可接受的在生产代码中。
我想在处理发生时统计结果，但在所有统计完成之前我不想退出程序。在 Go 中实现这一目标的最佳实践是什么？
这是我目前的方法，效果很好。我想知道是否有更好的方法，因为这看起来有点笨拙？
package main
import (
"fmt"
"sync"
)
type T struct{}
func main() {
var widgetInventory int = 1000
transactions := make(chan int, 100)
salesDone := make(chan T)
purchasesDone := make(chan T)
var wg sync.WaitGroup
fmt.Println("Starting inventory count = ", widgetInventory)
go makeSales(transactions, salesDone)
go newPurchases(transactions, purchasesDone)
wg.Add(1)
go func() {
salesAreDone := false
purchasesAreDone := false
for {
select {
case transaction := <-transactions:
widgetInventory += transaction
case <-salesDone:
salesAreDone = true
case <-purchasesDone:
purchasesAreDone = true
default:
if salesAreDone && purchasesAreDone {
wg.Done()
return
}
}
}
}()
wg.Wait()
fmt.Println("Ending inventory count = ", widgetInventory)
}
func makeSales(transactions chan int, salesDone chan T) {
for i := 0; i < 3000; i++ {
transactions <- -100
}
salesDone <- struct{}{}
}
func newPurchases(transactions chan int, purchasesDone chan T) {
for i := 0; i < 3000; i++ {
transactions <- 100
}
purchasesDone <- struct{}{}
}
不适合任何合理的定义很好。您在这里有一个热门的 for 循环：
for {
select {
case transaction := <-transactions:
widgetInventory += transaction
case <-salesDone:
salesAreDone = true
case <-purchasesDone:
purchasesAreDone = true
default:
if salesAreDone && purchasesAreDone {
wg.Done()
return
}
}
}
只要没有通道可供读取，就会执行 default 案例。由于渠道的工作方式，这种情况经常发生。
这个稍作调整的代码版本说明了此循环的“热度”。确切的结果会有所不同，可能会相当高。
Default case ran 27305 times
当 selecting 来自通道时，您不希望出现 default 情况，除非该默认情况也会阻止其中的某些内容。否则你会得到这样的热循环。
nilable 通道进行选择通常在选择中，您想要识别关闭的通道并将通道变量设置为 nil； select 永远不会成功地从 nil 通道读取内容，因此这实际上“禁用”了该选择。
考虑代码的此修改版本：
go func(transactions chan int, salesDone <-chan T, purchasesDone <-chan T) {
defer wg.Done()
for transactions != nil {
select {
case transaction, ok := <-transactions:
if ok {
widgetInventory += transaction
} else {
transactions = nil
}
case <-salesDone:
salesDone = nil
if purchasesDone == nil {
close(transactions)
}
case <-purchasesDone:
purchasesDone = nil
if salesDone == nil {
close(transactions)
}
}
}
}(transactions, salesDone, purchasesDone)
通过对消费者的这些调整，我们不再有热循环；我们总是阻塞直到从通道读取数据。一旦 salesDone 和 purchasesDone 都被“发出信号”，我们 close(transactions)。一旦我们耗尽 transactions 并且它被关闭，我们将 transactions 设置为 nil。我们在 transactions 不为 nil 时循环，在这段代码中，意味着所有通道都是 nil。
微妙但重要的一点：我将通道传递给此函数，因此它的引用不与 main 共享范围。否则，将 transactions 设置为 nil 将写入一个在 goroutine 之间共享的变量。然而在这种情况下，无论如何，这并不重要，因为我们“知道”我们是最后一个从 transactions 读取的内容。
如果您考虑一下您在这里所做的事情，您需要等到两个生产者都完成对 transactions 的生产。然后你想排空 transactions。一旦通道关闭并排空，main 就知道求和已完成。
您不需要 select 来执行此操作。而 select 为每个“工人”都有一个案例，可以说是相当不优雅的；您必须对多个工作人员进行硬编码并单独处理“完成”通道。
您需要做的是：
var resultswgsync.WaitGroup 之外，还为消费者添加一个。defer wg.Done()defer resultswg.Done() 在遍历 transactions 之前：
go func() {
defer resultswg.Done()
for transaction := range transactions {
widgetInventory += transaction
}
}()
wg.Wait() close(transactions) resultswg.Wait()
package main
import (
"fmt"
"sync"
)
func main() {
var widgetInventory int = 1000
transactions := make(chan int, 100)
var wg, resultswg sync.WaitGroup
fmt.Println("Starting inventory count = ", widgetInventory)
wg.Add(2)
go makeSales(transactions, &wg)
go newPurchases(transactions, &wg)
resultswg.Add(1)
go func() {
defer resultswg.Done()
for transaction := range transactions {
widgetInventory += transaction
}
}()
wg.Wait()
close(transactions)
resultswg.Wait()
fmt.Println("Ending inventory count = ", widgetInventory)
}
func makeSales(transactions chan int, wg *sync.WaitGroup) {
defer wg.Done()
for i := 0; i < 3000; i++ {
transactions <- -100
}
}
func newPurchases(transactions chan int, wg *sync.WaitGroup) {
defer wg.Done()
for i := 0; i < 3000; i++ {
transactions <- 100
}
}
您可以在这里看到，在此模式中可以有任意数量的生产者；您只需为每个生产者添加 wg.Add(1) 即可。
当我不知道每个工作人员会返回多少结果时，我一直使用这种模式来并行化工作。我发现它很容易理解，并且比尝试 select 多个通道简单得多。事实上，我什至想说，如果您发现自己从多个渠道进行 selecting，您应该退后一步，确保它对您来说确实有意义。我使用 select 的频率远远低于使用等待组的频率。
