Go Academy · レッスン

パイプラインとステージのパターン

Q: 「パイプラインとステージのパターン」レッスンは無料ですか？

はい。「パイプラインとステージのパターン」の完全なテキストはこのウェブで無料で読めます。インタラクティブに演習し（組み込みコードエディタと24時間対応のAIチューター）、Go Academyコースの残りをアンロックするには、CoddyKit PROにアップグレードしてください。 Go Academyコースには全4レッスンが含まれています。

Q: 「パイプラインとステージのパターン」で何を学びますか？

複数ステージの並行パイプラインを組み合わせます。 ブラウザで直接実行するハンズオンコードでGo Academyを演習し、24時間対応のAIチューターがレッスンを進める中での質問に答えます。

Q: Go Academyを始めるのに経験は必要ですか？

事前経験は必要ありません。CoddyKitのGo Academyは初級者から上級者向けに構成されているため、ここから始めるか最初から始めて、自分のペースで進むことができます。 これはレッスン1/4です。

Q: 「パイプラインとステージのパターン」レッスンにはどのくらい時間がかかりますか？

ほとんどのCoddyKitレッスンは約5～10分かかります。各レッスンはコンパクトでインタラクティブなので、着実に進歩し、ウェブとアプリ全体で正確に前回の場所から再開できます。

Q: このGo Academyレッスンでコードを書いて実行できますか？

はい。すべてのGo Academyレッスンに組み込みコードエディタが含まれているため、ブラウザでリアルコードを書いて実行し、即座のAIフィードバックを取得できます。ローカル設定は不要です。

複数ステージの並行パイプラインを組み合わせます。

「パイプラインとステージのパターン」はCoddyKit上の無料Go Academyレッスンです。これはレッスン1/4です。下記で完全なレッスンを無料で読むことができます。その後、ブラウザ内の組み込みコードエディタと24時間対応のAIチューターでハンズオン演習できます。これはGo Academy学習パスの一部であり、ウェブとCoddyKitアプリ全体で進捗が同期されます。 Go Academyコースには全4レッスンが含まれています。

パイプラインとは

並行パイプラインは、チャネルで接続された一連のステージです。各ステージは上流から値を受け取り、変換して下流へ結果を送信するため、ステージを並行して実行できます。

3段階の単純なパイプライン

生成 → 変換 → 消費：

func generate(nums ...int) <-chan int {
    out := make(chan int)
    go func() { defer close(out); for _, n := range nums { out <- n } }()
    return out
}
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
}

for v := range square(generate(2, 3, 4)) {
    fmt.Println(v) // 4, 9, 16
}

ステージパターン

各ステージは入力チャネルを受け取り、出力チャネルを返す関数です。ステージは容易に組み合わせることができ、それぞれを独立して並列化できます。

パイプライン内のファンアウト

ステージをN個のgoroutineに分岐して、CPU負荷の高い処理を並列化します。

func parallelSquare(in <-chan int, n int) <-chan int {
    outs := make([]<-chan int, n)
    for i := 0; i < n; i++ {
        outs[i] = square(in) // all read from same in
    }
    return merge(outs...)
}

contextによるキャンセル

各ステージにcontextを渡します。キャンセルされると、ステージは読み取りを停止して出力チャネルを閉じ、パイプライン全体を最後まで処理します。

func stage(ctx context.Context, in <-chan int) <-chan int {
    out := make(chan int)
    go func() {
        defer close(out)
        for {
            select {
            case v, ok := <-in:
                if !ok { return }
                select {
                case out <- process(v):
                case <-ctx.Done(): return
                }
            case <-ctx.Done(): return
            }
        }
    }()
    return out
}

バックプレッシャー

ステージ間にバッファ付きチャネルを使用して、プロデューサーとコンシューマーの速度差を吸収します。バッファリングがないと、遅いステージによってパイプライン全体が停止します。

out := make(chan Result, 50) // buffer 50 results

エラーの伝播

値を結果構造体でラップして、パニックを発生させずにパイプライン内へエラーを伝播させます。

type item struct { val int; err error }

doneチャネルパターン

単純なパイプラインでは、contextの代わりにdoneチャネルを使用できます。doneを閉じると、すべてのステージに停止を通知できます。

done := make(chan struct{})
defer close(done)

順序の保証

各ステージが逐次的に実行される場合、単純なパイプラインでは順序が保持されます。並列化したステージでは順序が保持されないため、必要に応じてインデックスフィールドを使い、下流で結果を並べ替えてください。

WaitGroupを使用したパイプライン

最後のステージでチャネルを閉じる必要はありません。チャネルをrangeで走査するだけで十分です。ファンインでは、すべてのプロデューサーgoroutineが終了した後に結合チャネルを閉じるため、WaitGroupを使用します。

実際の用途

パイプラインの一般的な用途には、ETLデータ処理、画像のリサイズ、ログ解析、ビルドシステム、ストリーミングリクエストハンドラーなどがあります。

クイックチェック

複数ステージのパイプライン全体にキャンセルを伝播させるには、どうすればよいですか。

まとめ：パイプラインパターン

重要なポイント：

ステージ：func(in <-chan T) <-chan U。組み合わせてパイプラインを構築します
キャンセルにはcontext、バックプレッシャーにはバッファ付きチャネルを使用します
ステージ内のファンアウトで並列性を高めます
エラーの伝播には値を結果構造体でラップします

よくある質問