Go 中的 Fanin 模式

讓我們來看看扇入模式。當我們需要將來自多個線程的相關資料匯集在一起時，這在 Go 中非常有用。

例如，假設您對不同的服務進行了多次 API 調用，並且需要合併結果。

這是一個非常容易實現的模式，但您確實需要注意如何處理通道。容易出現死鎖情況。

// produce is used to simulate the different data sources
func produce(id int) chan int {
    ch := make(chan int)
    go func() {
        for i := 0; i < 10; i++ {
            ch <- id*10 + i
        }
        fmt.Printf("producer %d done\n", id)
        close(ch) // this is important!!!
    }()
    return ch
}

func fanin(inputs ...chan int) chan int {
    output := make(chan int)

    var wg sync.WaitGroup
    for i, input := range inputs {
        wg.Add(1)
        go func() {
            for value := range input {
                output <- value
            }
            fmt.Printf("done merging source %d\n", i)
            wg.Done()
        }()
    }
    go func() {
        wg.Wait()
        close(output) // this is important!!!
    }()

    return output
}

func main() {
    input1 := produce(0)
    input2 := produce(1)

    result := fanin(input1, input2)

    done := make(chan bool)
    go func() {
        for value := range result {
            fmt.Printf("got %d\n", value)
        }
        close(done)
    }()
    <-done
    fmt.Println("done")
}

這裡我們使用 Produce 函數來模擬不同的來源。這些來源通道被傳送到執行組合操作的 fanin 函數。

fanin 函數建立輸出通道，然後啟動一個對每個輸入進行操作的 goroutine。我們使用 WaitGroup 來指示所有輸入來源何時組合到輸出通道。

在這個簡單的範例中，主執行緒只是迭代輸出。請注意，無法保證順序，兩個輸入的值是混合的。

要提出的一個關鍵點是，當我們完成組合輸入時，我們必須關閉輸出通道。一旦通道為空，範圍運算子將無限期等待。註解掉 close(output) 行，您將會看到出現死鎖情況。

我們如何改進這一點？請在下面留言給我。

謝謝！

這篇文章以及本系列所有文章的程式碼可以在這裡找到

以上是Go 中的 Fanin 模式的詳細內容。更多資訊請關注PHP中文網其他相關文章！