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

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

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

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

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

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

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

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

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

パイプラインパターン

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

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

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

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

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

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

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

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

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

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 つのステージの内部を変更しても、他のステージには影響しません
?新しいステージの追加や既存のステージの変更が簡単
?懸念事項を明確に分離

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. バースト処理を処理する必要があります

バッチ処理

このパターンは、複数の項目を 1 つのバッチにグループ化することで、I/O バウンド操作を最適化します。項目を一度に 1 つずつ処理するのではなく、項目をグループにまとめて一緒に処理します。これは次の場合に効果的です。

1. 外部システム (データベース、API) を使用している
2. ネットワークの往復は高価です
3. この操作にはリクエストごとに大幅な固定オーバーヘッドがあります
4. レイテンシよりもスループットを最適化する必要があります

これらの各パターンは、必要に応じて組み合わせることができます。たとえば、複数のワーカーがそれぞれアイテムのバッチを処理する、水平スケーリングを備えたバッチ処理を使用できます。 重要なのは、ボトルネックを理解し、それに対処するための適切なパターンを選択することです。

---

これで、ジェネレーター パターンについての詳細な説明は終わりです。次に、パイプライン同時実行パターンを検討し、ジェネレーターをチェーンして強力なデータ処理フローを構築する方法を見ていきます。

この投稿が役に立ったと思われた場合、ご質問がある場合、またはジェネレーターに関するご自身の経験を共有したい場合は、以下のコメント欄からご意見をお待ちしております。皆様の洞察や質問は、これらの説明を誰にとってもさらにわかりやすくするのに役立ちます。

Golang のゴルーチンとチャネルのビジュアルガイドを見逃した場合は、ここで確認してください:

Golang でのゴルーチンとチャネルの理解と視覚化

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

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

Go の同時実行パターンについては今後も続報をお待ちください! ?

以上がGo のパイプライン同時実行パターン: 包括的なビジュアル ガイドの詳細内容です。詳細については、PHP 中国語 Web サイトの他の関連記事を参照してください。