Golang 中跟蹤第三個 Goroutine 中兩個 Goroutine 的完成狀態(tài)的最佳實(shí)踐是什么？
在 Golang 中，要跟蹤兩個 Goroutine 的完成狀態(tài)并在第三個 Goroutine 中處理它們的結(jié)果，最佳實(shí)踐是使用 sync 包中的 WaitGroup。WaitGroup 允許我們在主 Goroutine 中等待其他 Goroutine 的完成。首先，我們需要創(chuàng)建一個 WaitGroup 對象，并在主 Goroutine 中調(diào)用 Add 方法來設(shè)置等待的 Goroutine 數(shù)量。然后，在每個 Goroutine 的末尾調(diào)用 Done 方法來表示該 Goroutine 的完成。最后，在第三個 Goroutine 中調(diào)用 Wait 方法來等待所有 Goroutine 的完成。這樣，我們就可以安全地跟蹤并處理兩個 Goroutine 的結(jié)果了。這是 Golang 中跟蹤多個 Goroutine 完成狀態(tài)的最佳實(shí)踐。
問題內(nèi)容
我有三個并發(fā)運(yùn)行的 goroutine。其中兩個進(jìn)行一些處理并將其結(jié)果發(fā)送到結(jié)果通道。第三個 goroutine 通過讀取結(jié)果通道來“統(tǒng)計(jì)”結(jié)果。我可以使用 waitgroup 等待兩個計(jì)算 goroutine 完成，然后遍歷結(jié)果通道來統(tǒng)計(jì)結(jié)果，但這無法擴(kuò)展，并且需要我創(chuàng)建一個具有巨大緩沖區(qū)大小的緩沖結(jié)果通道，這是不可接受的在生產(chǎn)代碼中。
我想在處理發(fā)生時統(tǒng)計(jì)結(jié)果，但在所有統(tǒng)計(jì)完成之前我不想退出程序。在 Go 中實(shí)現(xiàn)這一目標(biāo)的最佳實(shí)踐是什么？
這是我目前的方法，效果很好。我想知道是否有更好的方法，因?yàn)檫@看起來有點(diǎn)笨拙？
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{}{}
}
登錄后復(fù)制
解決方法
不適合任何合理的定義很好。您在這里有一個熱門的 for 循環(huán)：
for {
select {
case transaction := <-transactions:
widgetInventory += transaction
case <-salesDone:
salesAreDone = true
case <-purchasesDone:
purchasesAreDone = true
default:
if salesAreDone && purchasesAreDone {
wg.Done()
return
}
}
}
登錄后復(fù)制
只要沒有通道可供讀取，就會執(zhí)行 default 案例。由于渠道的工作方式，這種情況經(jīng)常發(fā)生。
這個稍作調(diào)整的代碼版本說明了此循環(huán)的“熱度”。確切的結(jié)果會有所不同，可能會相當(dāng)高。
Default case ran 27305 times
登錄后復(fù)制
當(dāng) selecting 來自通道時，您不希望出現(xiàn) default 情況，除非該默認(rèn)情況也會阻止其中的某些內(nèi)容。否則你會得到這樣的熱循環(huán)。
更好的方法：使用 nilable 通道進(jìn)行選擇
通常在選擇中，您想要識別關(guān)閉的通道并將通道變量設(shè)置為 nil； select 永遠(yuǎn)不會成功地從 nil 通道讀取內(nèi)容，因此這實(shí)際上“禁用”了該選擇。
考慮代碼的此修改版本：
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)
登錄后復(fù)制
通過對消費(fèi)者的這些調(diào)整，我們不再有熱循環(huán)；我們總是阻塞直到從通道讀取數(shù)據(jù)。一旦 salesDone 和 purchasesDone 都被“發(fā)出信號”，我們 close(transactions)。一旦我們耗盡 transactions 并且它被關(guān)閉，我們將 transactions 設(shè)置為 nil。我們在 transactions 不為 nil 時循環(huán)，在這段代碼中，意味著所有通道都是 nil。
微妙但重要的一點(diǎn)：我將通道傳遞給此函數(shù)，因此它的引用不與 main 共享范圍。否則，將 transactions 設(shè)置為 nil 將寫入一個在 goroutine 之間共享的變量。然而在這種情況下，無論如何，這并不重要，因?yàn)槲覀儭爸馈蔽覀兪亲詈笠粋€從 transactions 讀取的內(nèi)容。
更簡單的選項(xiàng)：多個等待組
如果您考慮一下您在這里所做的事情，您需要等到兩個生產(chǎn)者都完成對 transactions 的生產(chǎn)。然后你想排空 transactions。一旦通道關(guān)閉并排空，main 就知道求和已完成。
您不需要 select 來執(zhí)行此操作。而 select 為每個“工人”都有一個案例，可以說是相當(dāng)不優(yōu)雅的；您必須對多個工作人員進(jìn)行硬編碼并單獨(dú)處理“完成”通道。
您需要做的是：
除了為生產(chǎn)者使用一個 var resultswgsync.WaitGroup 之外，還為消費(fèi)者添加一個。
生產(chǎn)者 defer wg.Done()
消費(fèi)者 defer resultswg.Done() 在遍歷 transactions 之前：
go func() {
defer resultswg.Done()
for transaction := range transactions {
widgetInventory += transaction
}
}()
登錄后復(fù)制
main 處理等待生產(chǎn)者、關(guān)閉事務(wù)以結(jié)束范圍，然后等待消費(fèi)者：
wg.Wait() close(transactions) resultswg.Wait()
登錄后復(fù)制
以這種方式編碼，最終會變得簡短而甜蜜：
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
}
}
登錄后復(fù)制
您可以在這里看到，在此模式中可以有任意數(shù)量的生產(chǎn)者；您只需為每個生產(chǎn)者添加 wg.Add(1) 即可。
當(dāng)我不知道每個工作人員會返回多少結(jié)果時，我一直使用這種模式來并行化工作。我發(fā)現(xiàn)它很容易理解，并且比嘗試 select 多個通道簡單得多。事實(shí)上，我什至想說，如果您發(fā)現(xiàn)自己從多個渠道進(jìn)行 selecting，您應(yīng)該退后一步，確保它對您來說確實(shí)有意義。我使用 select 的頻率遠(yuǎn)遠(yuǎn)低于使用等待組的頻率。
