Go 中的管道並發模式：綜合視覺指南

⚠️ 這個系列如何進行？

1。運行每個範例：不要只閱讀程式碼。輸入它，運行它，然後觀察其行為。
2。實驗和打破常規： 刪除睡眠並看看會發生什麼，更改通道緩衝區大小，修改 goroutine 計數。
打破東西會教你它們是如何運作的
3。關於行為的原因： 在執行修改後的程式碼之前，嘗試預測結果。當您看到意外行為時，請停下來思考原因。挑戰解釋。
4。建立心理模型：每個視覺化代表一個概念。嘗試為修改後的程式碼繪製自己的圖表。

在上一篇文章中，我們探討了生成器並發模式，它是 Go 其他並發模式的構建塊。您可以在這裡閱讀：

Go 中的生成器並發模式：視覺指南

Souvik Kar Mahapatra ・ 12 月 25 日

＃去 ＃教程 #程式設計 ＃學習

現在，讓我們看看這些原語如何組合起來形成解決現實世界問題的強大模式。

在這篇文章中，我們將介紹管道模式並嘗試將它們視覺化。因此，讓我們做好準備，因為我們將親手完成整個過程。

管道模式

管道就像工廠中的組裝線，每個階段對資料執行特定任務，並將結果傳遞到下一個階段。

我們透過將 goroutine 與通道連接來建立管道，其中 每個 goroutine 代表一個階段，用於接收資料、處理資料並將其發送到下一個階段。

讓我們實作一個簡單的管道：

1. 產生數字
2. 將它們平方
3. 列印結果

// Stage 1: Generate numbers
func generate(nums ...int) <-chan int {
    out := make(chan int)
    go func() {
        defer close(out)
        for _, n := range nums {
            out <- n
        }
    }()
    return out
}

// Stage 2: Square numbers
func square(in <-chan int) <-chan int {
    out := make(chan int)
    go func() {
        defer close(out)
        for n := range in {
            out <- n * n
        }
    }()
    return out
}

// Stage 3: Print numbers
func print(in <-chan int) {
    for n := range in {
        fmt.Printf("%d ", n)
    }
    fmt.Println()
}

func main() {
    // Connect the pipeline
    numbers := generate(2, 3, 4)    // Stage 1
    squares := square(numbers)       // Stage 2
    print(squares)                   // Stage 3
}

✏️快字節

chan int 這表示雙向通道。
chan int 類型的通道可用於傳送和接收值。

讓我們繼續想像上面的例子：

在這裡您可以看到管道的每個構建塊都是遵循生成器模式的 goroutine。這意味著只要資料在任何步驟準備好，管道中的下一步就可以開始處理它，這與順序處理不同。

管道中的錯誤處理

核心原則應該是：

1. 每個階段都清楚知道如何處理好的和壞的值
2. 錯誤不會在管道中遺失
3. 錯誤的值不會引起恐慌
4. 錯誤訊息包含有關出錯原因的上下文
5. 管道可以擴展更多階段，並且它們都會一致地處理錯誤

讓我們用一些正確的錯誤處理來更新我們的程式碼。

type Result struct {
    Value int
    Err   error
}

func generateWithError(nums ...int) <-chan Result {
    out := make(chan Result)
    go func() {
        defer close(out)
        for _, n := range nums {
            if n < 0 {
                out <- Result{Err: fmt.Errorf("negative number: %d", n)}
                return
            }
            out <- Result{Value: n}
        }
    }()
    return out
}

func squareWithError(in <-chan Result) <-chan Result {
    out := make(chan Result)
    go func() {
        defer close(out)
        for r := range in {
            if r.Err != nil {
                out <- r  // Forward the error
                continue
            }
            out <- Result{Value: r.Value * r.Value}
        }
    }()
    return out
}

func main() {
    // Using pipeline with error handling
    for result := range squareWithError(generateWithError(2, -3, 4)) {
        if result.Err != nil {
            fmt.Printf("Error: %v\n", result.Err)
            continue
        }
        fmt.Printf("Result: %d\n", result.Value)
    }
}

為什麼要使用管道模式？

讓我們舉個例子來更好地理解，我們有一個遵循管道模式的資料處理工作流程，如下所示。

1. 管道中的每個階段都是獨立運作的，僅透過通道進行通訊。這有幾個好處：

？每個階段都可以獨立開發、測試和修改
？一個階段內部結構的改變不會影響其他階段
？輕鬆新增階段或修改現有階段
？明確的關注分離

1. 管道模式自然地支援並行/並發處理。一旦數據可用，每個階段都可以同時處理不同的數據。

最好的部分是什麼？我們可以運行每個階段的多個實例（工作人員）以滿足更多並發要求，如下所示：

??嘿，但這不是扇入和扇出並發模式嗎？

賓果！很好的收穫就在那裡。它確實是一種扇出、扇出模式，它是管道模式的特定類型。我們將在下一篇文章中詳細介紹它，所以不用擔心;)

現實世界用例

在管道中處理影像

// Stage 1: Generate numbers
func generate(nums ...int) <-chan int {
    out := make(chan int)
    go func() {
        defer close(out)
        for _, n := range nums {
            out <- n
        }
    }()
    return out
}

// Stage 2: Square numbers
func square(in <-chan int) <-chan int {
    out := make(chan int)
    go func() {
        defer close(out)
        for n := range in {
            out <- n * n
        }
    }()
    return out
}

// Stage 3: Print numbers
func print(in <-chan int) {
    for n := range in {
        fmt.Printf("%d ", n)
    }
    fmt.Println()
}

func main() {
    // Connect the pipeline
    numbers := generate(2, 3, 4)    // Stage 1
    squares := square(numbers)       // Stage 2
    print(squares)                   // Stage 3
}

或是像日誌處理管道一樣複雜的東西

管道擴展模式

水平縮放（扇出、扇入）

此模式非常適合 CPU 密集型操作，其中工作可以獨立處理。管道將工作分配給多個工作人員，然後重新組合結果。這在以下情況特別有效：

1. 處理是CPU密集的（資料轉換、運算）
2. 可以獨立處理任務
3. 您有多個可用的 CPU 核心

緩衝管道

此模式有助於管理管道階段之間的速度不匹配。緩衝器起到減震器的作用，允許快速階段向前工作，而不會被較慢階段阻擋。這在以下情況很有用：

1. 不同階段有不同的處理速度
2. 您想要保持穩定的吞吐量
3. 緩衝的記憶體使用是可以接受的
4. 您需要處理突發處理

批次處理

此模式透過將多個項目分組為單一批次來最佳化 I/O 密集型操作。它不是一次處理一個項目，而是將它們分組並一起處理。這在以下情況下有效：

1. 您正在使用外部系統（資料庫、API）
2. 網路往返費用昂貴
3. 該操作每個請求都有大量固定開銷
4. 您需要最佳化吞吐量而不是延遲

這些模式中的每一個都可以根據需要進行組合。例如，您可以使用水平擴展的批次處理，其中多個工作人員每個處理一批專案。 關鍵是了解您的瓶頸並選擇適當的模式來解決它們。

---

我們對生成器模式的深入研究到此結束了！接下來，我們將探索管道並發模式，我們將了解如何將生成器連結在一起以建立強大的資料處理流程。

如果您發現這篇文章有幫助，有任何疑問，或者想分享您自己的生成器經驗 - 我很樂意在下面的評論中聽到您的意見。您的見解和問題有助於使這些解釋對每個人都更好。

如果您錯過了 Golang 的 goroutine 和通道的視覺指南，請在此處查看：

理解和視覺化 Golang 中的 Goroutine 和 Channel

Souvik Kar Mahapatra ・ 12 月 20 日

＃去 #程式設計 ＃學習 ＃教程

請繼續關注更多 Go 並發模式！ ？

以上是Go 中的管道並發模式：綜合視覺指南的詳細內容。更多資訊請關注PHP中文網其他相關文章！