Go のパイプライン同時実行パターン: 包括的なビジュアルガイド-Golang-php.cn

⚠️ このシリーズをどのように進めていきますか?

1.すべての例を実行: コードを読むだけではありません。入力して実行し、動作を観察してください。
2.実験と破壊: スリープを削除して何が起こるか確認し、チャネルバッファーサイズを変更し、ゴルーチン数を変更します。
物を壊すことで、その仕組みがわかる
3.動作に関する理由: 変更されたコードを実行する前に、結果を予測してみてください。予期せぬ動作が見られた場合は、立ち止まってその理由を考えてください。解説に挑戦してください。
4.メンタルモデルの構築: 各ビジュアライゼーションはコンセプトを表します。変更されたコード用に独自の図を描いてみてください。

前回の投稿では、Go の他の同時実行パターンの構成要素であるジェネレーターの同時実行パターンについて説明しました。ここで読んでみてください:

Pipeline Concurrency Pattern in Go: A Comprehensive Visual Guide

Go のジェネレーター同時実行パターン: ビジュアルガイド

スーヴィク・カル・マハパトラ・12月25日

＃行く＃チュートリアル＃プログラミング＃学ぶ

ここで、これらのプリミティブがどのように組み合わされて、現実世界の問題を解決する強力なパターンを形成するかを見てみましょう。

この投稿では、パイプラインパターン について説明し、視覚化してみます。プロセス全体を通じて実際に作業を進めていきますので、準備を整えましょう。

Pipeline Concurrency Pattern in Go: A Comprehensive Visual Guide

パイプラインパターン

パイプラインは工場の組立ラインに似ており、各ステージがデータに対して特定のタスクを実行し、結果を次のステージに渡します。

ゴルーチンをチャネルに接続することでパイプラインを構築します。各ゴルーチンは、データを受信し、処理し、次のステージに送信するステージを表します。

Pipeline Concurrency Pattern in Go: A Comprehensive Visual Guide

次のような単純なパイプラインを実装してみましょう。

数値を生成します
二乗します
結果を印刷します

// 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 型のチャネルは、値を送信するのではなく、値を受信するためにのみ使用できます。これは、より厳格な通信パターンを適用し、受信者によるチャネルへの誤った書き込みを防ぐのに役立ちます。

chan int これは双方向チャネルを示します。
chan int 型のチャネルは、値の送信と受信の両方に使用できます。

上記の例を視覚化してみましょう:

Pipeline Concurrency Pattern in Go: A Comprehensive Visual Guide

ここでは、パイプラインの各構成要素がジェネレーターパターンに従ったゴルーチンであることがわかります。順次処理とは異なり、いずれかのステップでデータの準備が整うとすぐに、パイプラインの次のステップでデータの処理を開始できることを意味します。

パイプラインでのエラー処理

核となる原則は次のとおりです:

各ステージは、良い値と悪い値の両方をどうすべきかを正確に知っています
エラーがパイプラインで失われることはありません
間違った値はパニックを引き起こしません
エラーメッセージには、問題の内容が記載されています
パイプラインはステージを増やして拡張でき、すべてのステージで一貫してエラーが処理されます

適切なエラー処理を追加してコードを更新しましょう。

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)
    }
}

ログイン後にコピー

パイプラインパターンを使用する理由

理解を深めるために例を挙げてみましょう。以下に示すようなパイプラインパターンに従うデータ処理ワークフローがあります。

Pipeline Concurrency Pattern in Go: A Comprehensive Visual Guide

パイプラインの各ステージは独立して動作し、チャネルを通じてのみ通信します。これにより、次のようないくつかの利点が得られます。

?各段階は個別に開発、テスト、変更できます
? 1 つのステージの内部を変更しても、他のステージには影響しません
?新しいステージの追加や既存のステージの変更が簡単
?懸念事項を明確に分離

Pipeline Concurrency Pattern in Go: A Comprehensive Visual Guide

パイプラインパターンにより、並列/同時処理が自然に可能になります。データが利用可能になるとすぐに、各ステージで異なるデータを同時に処理できます。

Pipeline Concurrency Pattern in Go: A Comprehensive Visual Guide

そして一番いいところは?次のように、より多くの同時要件に対して各ステージ (ワーカー) の複数のインスタンスを実行できます。

Pipeline Concurrency Pattern in Go: A Comprehensive Visual Guide

??あれ、でもそれは ファンインおよびファンアウトの同時実行パターン ではないですか?

ビンゴ！そこに良いキャッチがあります。これは確かにファンアウト、ファンインパターンであり、特定のタイプのパイプラインパターンです。次の投稿で詳しく説明するので、心配しないでください ;)

現実世界の使用例

パイプラインで画像を処理しています

// 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
}

ログイン後にコピー

またはログ処理パイプラインのような複雑なもの

Pipeline Concurrency Pattern in Go: A Comprehensive Visual Guide