Goの並行処理パターン

Q: パイプラインとワーカープールの違いは何ですか？

パイプラインとは、複数の処理段階を経るデータフローのことで、各段階では単一のステップ（例：解析 → 変換 → 保存）が実行され、データの形式が変化する場合があります。ワーカープールは、同じジョブキューからタスクを取得する固定数のワーカーで構成され、データ形式は変更されません（例：画像圧縮）。

Q: What are the core concepts behind fan-out and fan-in?

Fan-out distributes data from a single channel to multiple workers for parallel processing (to increase throughput). Fan-in merges the results from multiple channels into a single channel (for unified consumption). The two are typically used in tandem.

Q: When must you use Or-Done?

When reading from a channel whose goroutine lifecycle you cannot control (such as a channel returned by a third-party library). Or-Done ensures that the reading goroutine will not block on channel operations if the caller cancels. For simple scenarios, you can use select with a done channel.

Q: How do I choose between errgroup and sync.WaitGroup ?

Use errgroup when you need to collect errors or propagate cancellations. Use sync.WaitGroup (which is lighter-weight) when you just need to wait for a goroutine to finish. errgroup is a superset of WaitGroup in terms of functionality, but it introduces an additional goroutine to manage cancellations.

Q: パイプライン内のチャネルのバッファサイズはどのように決定すればよいですか？

基本的な原則は、プロデューサーがコンシューマーによってブロックされないようにすることです。バッファが大きすぎるとメモリを浪費し、小さすぎるとゴルーチン間のコンテキストスイッチが増加します。経験則として、バッファサイズ = 生成レート × 目標レイテンシとなります。実際のサイズは、本番環境での負荷テストを通じて決定してください。

Q: ゴルーチンリークを防ぐにはどうすればよいですか？

3つのルールがあります：(1) ゴルーチンを作成する場所には、必ず終了メカニズム（チャネルの閉じたり、 done シグナルの送信など）を設けること； (2) ライフサイクルを管理するには errgroup または WaitGroup を使用すること； (3) 制御されていないチャネルをラップするには Or-Done を使用すること。終了時期が分からない場合は、決してゴルーチンを作成してはいけません。

Q: これらのパターンは組み合わせて使用できますか？

はい、組み合わせるべきです。たとえば、パイプラインの各ステージ内で、ワーカープールとファンアウト／ファンインを組み合わせて使用することができます。また、errgroupはパイプラインのオーケストレーターとして機能し、キャンセルやエラーを一元的に管理することができます。

Goにおける並行処理は、ライブラリではなく、「ファンイン」、「ファンアウト」、「パイプライン」といったパターンに依存しています。これらのパターンを組み合わせることで、並行処理の問題の90％を解決することができます。

大量のデータを処理する必要がある場合――ジェネレータからデータを読み込み、複数のワーカーが並行して処理を行い、最終的に結果を統合する――どのようにすれば洗練されたソリューションを設計できるでしょうか？このレッスンでは、Goコミュニティで最も定番の並行処理パターンを習得します。

1. 学習内容

パイプラインパターン
ファンアウト／ファンイン
労働者プール
Or-Doneチャンネルの簡易モード
キャンセルの伝播
golang.org/x/sync/errgroup エラーグループ
実践編：リアルタイムデータ処理パイプライン

2. あるデータエンジニアの実話

(1) 課題：1,000万件のログエントリを単一のスレッドで処理するのに一晩かかってしまった。

ファティマはデータプラットフォームエンジニアです。同社では毎日1,000万件のアクセスログが生成されており、これらをリアルタイムで処理する必要があります：

「最初のバージョンでは、forループを使用していました。レコードを読み込み→解析→データベースに書き込むという流れです。本番運用開始初日、処理速度が生産量に追いつかないことが判明しました。ログキューが1時間あたり200万エントリずつ増加していたのです。上司は『明日のレポートはこのデータにかかっている』と言いました。私はサーバーを追加しましたが、コードは依然としてシングルスレッドのままで、CPUの稼働率はわずか5％にとどまっていました。」

当時の彼女のコード：

// 不良コード：シングルスレッド，全く必要ありません CPU
func processLogs(logs []string) {
    for _, log := range logs {
        parsed := parse(log)    // CPU 集約型
        enriched := enrich(parsed) // CPU 集約型
        save(enriched)           // I/O 集約型
    }
    // 1000 1万件のログ：3 時間！
}

(2) Goにおける並行処理のアプローチ：パイプライン＋ファンアウト

// pipeline.go
パッケージ main

import (
    "fmt"
    "time"
)

type LogEntry struct {
    Raw     文字列
    Level   文字列
    Message 文字列
    Time    time.Time
}

func main() {
    logs := generateLogs(100) // データソース
    start := time.Now()

    // Pipeline: generator → parse → enrich → save
    rawCh := generator(logs)
    parsedCh := parseStage(rawCh, 4)   // 4 個 parse worker
    enrichedCh := enrichStage(parsedCh, 4) // 4 個 enrich worker
    saveStage(enrichedCh, 2) // 2 個 save worker

    fmt.Printf("処理が完了しました，所要時間: %v\n", time.Since(start))
}

// Stage 1: Generator（データソース）
func generator(logs []文字列) <-chan 文字列 {
    out := make(chan 文字列, 100)
    go func() {
        defer close(out)
        for _, log := range logs {
            out <- log
        }
    }()
    return out
}

// Stage 2: Parse（Fan-out）
func parseStage(in <-chan 文字列, workers int) <-chan LogEntry {
    out := make(chan LogEntry, 100)
    for i := 0; i < workers; i++ {
        go func() {
            for raw := range in {
                time.Sleep(1 * time.Millisecond) // シミュレーション解析
                out <- LogEntry{Raw: raw, Level: "INFO", Message: raw, Time: time.Now()}
            }
        }()
    }
    return out
}

// Stage 3: Enrich（Fan-out）
func enrichStage(in <-chan LogEntry, workers int) <-chan LogEntry {
    out := make(chan LogEntry, 100)
    for i := 0; i < workers; i++ {
        go func() {
            for entry := range in {
                time.Sleep(2 * time.Millisecond) // 豊富なデータをシミュレートする
                out <- entry
            }
        }()
    }
    return out
}

// Stage 4: Save（Fan-in to 2 workers）
func saveStage(in <-chan LogEntry, workers int) {
    var wg sync.WaitGroup
    for i := 0; i < workers; i++ {
        wg.Add(1)
        go func() {
            defer wg.Done()
            for entry := range in {
                _ = entry // データベースへの書き込みのシミュレーション
                time.Sleep(500 * time.Microsecond)
            }
        }()
    }
    wg.Wait()
}

(3) パフォーマンス：シングルスレッドとパイプラインの比較

メトリック	シングルスレッド	パイプライン (4+4+2 ワーカー)	改善率
1,000万件のレコードの処理時間	約3時間	約12分	15倍
CPU使用率	5%	85%	17倍
コードの複雑度	単純	中程度	—
スケーラビリティ	サーバーを追加	ワーカーを追加（数を変更）	—

3. パイプライン

▶ サンプル：デジタル処理パイプライン

package main

import (
    "fmt"
)

// Stage 1: Generate
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
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: Double
func double(in <-chan int) <-chan int {
    out := make(chan int)
    go func() {
        defer close(out)
        for n := range in {
            out <- n * 2
        }
    }()
    return out
}

func main() {
    // Pipeline: generate → square → double
    for result := range double(square(generate(1, 2, 3, 4, 5))) {
        fmt.Println(result)
    }
    // 出力: 2, 8, 18, 32, 50  (n²×2)
}

▶ 試してみよう

graph LR
    A[generate] -->|chan int| B[square]
    B -->|chan int| C[double]
    C -->|chan int| D[main]

💡 ヒント： パイプラインの鍵となるのは、各ステージが <-chan T（読み取り専用チャネル）を返し、<-chan T を入力として受け取るという点です。各ステージは独自の goroutine で実行され、チャネルを介して接続されています。これが Go の CSP モデルです。

4. ファンアウト／ファンイン

▶ サンプル：ファンアウト＋ファンイン

package main

import (
    "fmt"
    "sync"
)

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

// Stage 2: Worker（Fan-out target）
func worker(id int, in <-chan int) <-chan int {
    out := make(chan int)
    go func() {
        defer close(out)
        for n := range in {
            result := n * n
            fmt.Printf("Worker %d: %d^2 = %d\n", id, n, result)
            out <- result
        }
    }()
    return out
}

// Fan-in: 複数を結合する channel
func fanIn(channels ...<-chan int) <-chan int {
    out := make(chan int)
    var wg sync.WaitGroup

    for _, ch := range channels {
        wg.Add(1)
        go func(c <-chan int) {
            defer wg.Done()
            for v := range c {
                out <- v
            }
        }(ch)
    }

    go func() {
        wg.Wait()
        close(out)
    }()

    return out
}

func main() {
    // Fan-out: 一つ generator 配布先： 3 個 worker
    in := generate(1, 2, 3, 4, 5, 6)
    w1 := worker(1, in)
    w2 := worker(2, in)
    w3 := worker(3, in)

    // Fan-in: 3 個 worker の結果を1つに統合する channel
    results := fanIn(w1, w2, w3)

    for result := range results {
        fmt.Printf("結果: %d\n", result)
    }
}

▶ 試してみよう

graph LR
    G[Generate] -->|fan-out| W1[Worker 1]
    G -->|fan-out| W2[Worker 2]
    G -->|fan-out| W3[Worker 3]
    W1 -->|fan-in| R[Results]
    W2 -->|fan-in| R
    W3 -->|fan-in| R

5. ワーカープール

▶ サンプル：ワーカープール

package main

import (
    "fmt"
    "sync"
    "time"
)

type Job struct {
    ID      int
    Payload string
}

type Result struct {
    JobID   int
    Output  string
    Err     error
}

func worker(id int, jobs <-chan Job, results chan<- Result, wg *sync.WaitGroup) {
    defer wg.Done()
    for job := range jobs {
        fmt.Printf("Worker %d 処理 job %d\n", id, job.ID)
        time.Sleep(100 * time.Millisecond) // 模擬業務
        results <- Result{
            JobID:  job.ID,
            Output: fmt.Sprintf("processed by worker %d", id),
        }
    }
}

func main() {
    const numJobs = 20
    const numWorkers = 5

    jobs := make(chan Job, numJobs)
    results := make(chan Result, numJobs)

    // 起動 Worker Pool
    var wg sync.WaitGroup
    for i := 0; i < numWorkers; i++ {
        wg.Add(1)
        go worker(i, jobs, results, &wg)
    }

    // 送信 job
    for i := 0; i < numJobs; i++ {
        jobs <- Job{ID: i, Payload: fmt.Sprintf("data-%d", i)}
    }
    close(jobs)

    // すべてを待つ worker 完了
    wg.Wait()
    close(results)

    // 収集結果
    for result := range results {
        fmt.Printf("Job %d → %s\n", result.JobID, result.Output)
    }
}

▶ 試してみよう

(1) パイプライン vs ワーカープール vs ファンアウト／ファンイン

パターン	基本概念	適用可能なシナリオ
パイプライン	データは複数のステージを経て流れ、各ステージでは1つの処理が行われる	処理手順が明確に定義されている（解析 → 変換 → 保存）
ワーカープール	固定数のワーカーがジョブキューからタスクを取得する	タスクのバッチ処理（画像圧縮、メール送信）
ファンアウト／ファンイン	並列処理のために複数のワーカーに分散させ、その後結果をマージする	ステートレスな並列計算（数値演算、データフィルタリング）

6. Or-Doneチャンネル

▶ サンプル：Or-Doneモード

package main

import (
    "fmt"
    "time"
)

// orDone 1つを包装する channel，…を通じて対応できるようにする done チャネルの削除
func orDone(done <-chan struct{}, c <-chan int) <-chan int {
    out := make(chan int)
    go func() {
        defer close(out)
        for {
            select {
            case <-done:
                return
            case v, ok := <-c:
                if !ok {
                    return
                }
                select {
                case out <- v:
                case <-done:
                    return
                }
            }
        }
    }()
    return out
}

func generateWithCancel(done <-chan struct{}, nums ...int) <-chan int {
    out := make(chan int)
    go func() {
        defer close(out)
        for _, n := range nums {
            select {
            case <-done:
                return
            case out <- n:
            }
        }
    }()
    return out
}

func main() {
    done := make(chan struct{})

    // 起動 generator（予定されている 3 数秒後にキャンセルされた）
    nums := generateWithCancel(done, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10)

    // を通じて orDone 安全な消費
    for n := range orDone(done, nums) {
        fmt.Printf("処理: %d\n", n)
        if n >= 5 {
            close(done) // すべての操作をキャンセルする
        }
    }

    fmt.Println("キャンセルされました")
    time.Sleep(100 * time.Millisecond) // 待つ goroutine 終了
}

▶ 試してみよう

🔥 よくある間違い： チャネルからの受信を怠ると、ゴルーチンリークが発生する可能性があります。つまり、ジェネレータ・ゴルーチンが送信処理でブロックされてしまうのです。Or-Done パターンを使用すれば、キャンセル時にゴルーチンがリークされることはありません。 上流の送信側と下流の受信側の双方がキャンセルを認識する必要がある場合は、チャネルを Or-Done でラップしてください。

7. errgroup エラーの伝播

▶ サンプル：errgroup

パッケージ main

import (
    "fmt"
    "time"

    "golang.org/x/sync/errgroup"
)

func main() {
    // 作成する errgroup
    // 最初に返されるもの エラー の goroutine その他は取り消されます goroutine
    g := errgroup.Group{}

    // 起動 3 個 goroutine
    for i := 0; i < 3; i++ {
        id := i
        g.Go(func() エラー {
            return doWork(id)
        })
    }

    // すべてを待つ goroutine 完了，最初のものを取得する エラー
    if err := g.Wait(); err != nil {
        fmt.Printf("ミッション失敗: %v\n", err)
    } else {
        fmt.Println("すべて成功")
    }
}

func doWork(id int) エラー {
    fmt.Printf("Worker %d 開始\n", id)
    time.Sleep(time.Duration(id+1) * 500 * time.Millisecond)

    if id == 1 {
        return fmt.Errorf("worker %d エラーが発生しました", id)
    }

    fmt.Printf("Worker %d 完了\n", id)
    return nil
}

▶ 試してみよう

▶ サンプル：errgroup ＋コンテキストタイムアウト

package main

import (
    "context"
    "fmt"
    "time"

    "golang.org/x/sync/errgroup"
)

func main() {
    // 帯 Context の errgroup
    ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
    defer cancel()

    g, ctx := errgroup.WithContext(ctx)

    // Worker 確認 Context キャンセルするか
    for i := 0; i < 5; i++ {
        id := i
        g.Go(func() error {
            select {
            case <-time.After(time.Duration(id+1) * time.Second):
                fmt.Printf("Worker %d 完了\n", id)
                return nil
            case <-ctx.Done():
                fmt.Printf("Worker %d 取り消された: %v\n", id, ctx.Err())
                return ctx.Err()
            }
        })
    }

    if err := g.Wait(); err != nil {
        fmt.Printf("errgroup 停止: %v\n", err)
    }
}

▶ 試してみよう

(2) sync.WaitGroup と errgroup の比較

プロパティ	sync.WaitGroup	errgroup
エラーの収集	未対応	✅ 最初のエラーを返す
伝播のキャンセル	未対応	✅ WithContext による自動キャンセル
ゴルーチン管理	手動での追加・完了	✅ 自動ゴルーチン管理
タイムアウト制御	手動での実装	✅ コンテキストを使用して簡単に実装可能

8. 完全な例：リアルタイムデータ処理パイプライン

// data_pipeline.go
package main

import (
    "context"
    "fmt"
    "math/rand"
    "sync"
    "time"
)

// ---------- データ型 ----------

type DataPoint struct {
    ID        int
    Value     float64
    Timestamp time.Time
}

type ProcessedData struct {
    DataPoint
    Normalized bool
    Anomaly    bool
    Category   string
}

// ---------- Pipeline Stages ----------

// Stage 1: Source（データソース）
func source(ctx context.Context, count int) <-chan DataPoint {
    out := make(chan DataPoint, 100)
    go func() {
        defer close(out)
        for i := 0; i < count; i++ {
            select {
            case <-ctx.Done():
                return
            case out <- DataPoint{
                ID:        i,
                Value:     rand.Float64() * 100,
                Timestamp: time.Now(),
            }:
            }
        }
    }()
    return out
}

// Stage 2: Normalize（正規化，Fan-out）
func normalize(ctx context.Context, in <-chan DataPoint, workers int) <-chan ProcessedData {
    out := make(chan ProcessedData, 100)
    var wg sync.WaitGroup

    for i := 0; i < workers; i++ {
        wg.Add(1)
        go func() {
            defer wg.Done()
            for dp := range orDone(ctx.Done(), in) {
                select {
                case <-ctx.Done():
                    return
                case out <- ProcessedData{
                    DataPoint:  dp,
                    Normalized: true,
                    Category:   categorize(dp.Value),
                }:
                }
            }
        }()
    }

    go func() {
        wg.Wait()
        close(out)
    }()

    return out
}

// Stage 3: Detect Anomaly（異常検知，Fan-out）
func detectAnomaly(ctx context.Context, in <-chan ProcessedData, workers int) <-chan ProcessedData {
    out := make(chan ProcessedData, 100)
    var wg sync.WaitGroup

    for i := 0; i < workers; i++ {
        wg.Add(1)
        go func() {
            defer wg.Done()
            for pd := range orDone(ctx.Done(), in) {
                pd.Anomaly = pd.Value > 90 || pd.Value < 10
                select {
                case <-ctx.Done():
                    return
                case out <- pd:
                }
            }
        }()
    }

    go func() {
        wg.Wait()
        close(out)
    }()

    return out
}

// Stage 4: Sink（出力，Fan-in）
func sink(ctx context.Context, in <-chan ProcessedData) {
    var normalCount, anomalyCount int
    for pd := range orDone(ctx.Done(), in) {
        if pd.Anomaly {
            anomalyCount++
            fmt.Printf("[異常] ID=%d, Value=%.2f, Category=%s\n",
                pd.ID, pd.Value, pd.Category)
        } else {
            normalCount++
        }
    }
    fmt.Printf("\n統計: 正常=%d, 異常=%d\n", normalCount, anomalyCount)
}

// 補助関数
func orDone(done <-chan struct{}, in <-chan DataPoint) <-chan DataPoint {
    out := make(chan DataPoint)
    go func() {
        defer close(out)
        for {
            select {
            case <-done:
                return
            case v, ok := <-in:
                if !ok {
                    return
                }
                select {
                case out <- v:
                case <-done:
                    return
                }
            }
        }
    }()
    return out
}

func categorize(value float64) string {
    switch {
    case value < 30:
        return "low"
    case value < 70:
        return "medium"
    default:
        return "high"
    }
}

func main() {
    rand.Seed(time.Now().UnixNano())
    ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
    defer cancel()

    fmt.Println("リアルタイムデータ処理パイプラインの起動...")

    // 構築 Pipeline
    data := source(ctx, 1000)
    normalized := normalize(ctx, data, 5)    // 5 個の正規化 worker
    detected := detectAnomaly(ctx, normalized, 3) // 3 異常検出 worker
    sink(ctx, detected)

    fmt.Println("配管の処理が完了しました")
}

💡 ヒント： 本番環境では、パイプラインの各ステージごとに、独自のバッファサイズ、再試行ロジック、および監視メトリクスを設定する必要があります。expvar または prometheus を使用すると、各ステージの処理速度、キューの長さ、エラー率を確認できます。

❓ よくある質問

Q パイプラインとワーカープールの違いは何ですか？

A パイプラインとは、複数の処理段階を経るデータフローのことで、各段階では単一のステップ（例：解析 → 変換 → 保存）が実行され、データの形式が変化する場合があります。ワーカープールは、同じジョブキューからタスクを取得する固定数のワーカーで構成され、データ形式は変更されません（例：画像圧縮）。

Q What are the core concepts behind fan-out and fan-in?

A Fan-out distributes data from a single channel to multiple workers for parallel processing (to increase throughput). Fan-in merges the results from multiple channels into a single channel (for unified consumption). The two are typically used in tandem.

Q When must you use Or-Done?

A When reading from a channel whose goroutine lifecycle you cannot control (such as a channel returned by a third-party library). Or-Done ensures that the reading goroutine will not block on channel operations if the caller cancels. For simple scenarios, you can use select with a done channel.

Q How do I choose between errgroup and sync.WaitGroup?

A Use errgroup when you need to collect errors or propagate cancellations. Use sync.WaitGroup (which is lighter-weight) when you just need to wait for a goroutine to finish. errgroup is a superset of WaitGroup in terms of functionality, but it introduces an additional goroutine to manage cancellations.

Q パイプライン内のチャネルのバッファサイズはどのように決定すればよいですか？

A 基本的な原則は、プロデューサーがコンシューマーによってブロックされないようにすることです。バッファが大きすぎるとメモリを浪費し、小さすぎるとゴルーチン間のコンテキストスイッチが増加します。経験則として、バッファサイズ = 生成レート × 目標レイテンシとなります。実際のサイズは、本番環境での負荷テストを通じて決定してください。

Q ゴルーチンリークを防ぐにはどうすればよいですか？

A 3つのルールがあります：(1) ゴルーチンを作成する場所には、必ず終了メカニズム（チャネルの閉じたり、doneシグナルの送信など）を設けること； (2) ライフサイクルを管理するには errgroup または WaitGroup を使用すること； (3) 制御されていないチャネルをラップするには Or-Done を使用すること。終了時期が分からない場合は、決してゴルーチンを作成してはいけません。

Q これらのパターンは組み合わせて使用できますか？

A はい、組み合わせるべきです。たとえば、パイプラインの各ステージ内で、ワーカープールとファンアウト／ファンインを組み合わせて使用することができます。また、errgroupはパイプラインのオーケストレーターとして機能し、キャンセルやエラーを一元的に管理することができます。

📖 まとめ

パイプライン：データは複数の処理段階を経て流れ、各段階ごとに1つのgoroutineが割り当てられる
ファンアウト：1つのチャネルを複数のワーカーに分散させる（並列処理のため）
ファンイン：複数のチャンネルを1つのチャンネルに統合すること（一元的な視聴のため）
ワーカープール：一定数のワーカーがジョブキューからタスクを取得する
Or-Done：キャンセル機能を備えた、安全で制御不能な消費用チャネル
errgroup: ゴルーチン・グループを管理し、エラーの収集とキャンセル伝播をサポートする
組み合わせモード：Pipelineは内部でワーカープールとファンアウト／ファンインを併用しています

📝 練習問題

基本問題（難易度 ⭐）：3段階の数値処理パイプラインを実装してください：生成（1～100）→ フィルタリング（偶数のみを残す）→ 合計（累積）。各段階は個別のgoroutineで実行され、チャネルを介して接続されています。
上級問題（難易度 ⭐⭐）：並行ログ処理システムを実装してください。構成は、10個のパースワーカー（ファンアウト）＋3個の書き込みワーカー（ファンイン）＋errgroupによるエラー処理です。1,000件のログエントリをシミュレートし、統計情報（総経過時間、処理されたエントリ数、エラー数）を出力してください。
課題（難易度：⭐⭐⭐）：負荷分散と動的スケーリングを備えたワーカープールを設計・実装してください。要件：(1) 最初はワーカーを5つで開始すること；(2) ジョブキューのバックログが閾値を超えた場合、ワーカーの数を自動的に増やすこと（最大20まで）； (3) キューが空になった際に、ワーカーの数を（最大2まで）減らすこと；(4) Context を使用してワーカーの終了を制御すること；(5) -race を使用して、レースコンディションが発生していないことを検証すること。

Web-Tutorial 技術チーム

複数の開発者によって共同維持されているプログラミングチュートリアルプラットフォーム。各チュートリアルは専門分野の開発者が執筆・レビューしています。正確で信頼性の高いコンテンツを目指しています — 問題を見つけた場合はお知らせください。

私たちについて

Goの並行処理パターン

1. 学習内容

2. あるデータエンジニアの実話

(1) 課題：1,000万件のログエントリを単一のスレッドで処理するのに一晩かかってしまった。

(2) Goにおける並行処理のアプローチ：パイプライン＋ファンアウト

(3) パフォーマンス：シングルスレッドとパイプラインの比較

3. パイプライン

▶ サンプル：デジタル処理パイプライン

4. ファンアウト／ファンイン

▶ サンプル：ファンアウト＋ファンイン

5. ワーカープール

▶ サンプル：ワーカープール

(1) パイプライン vs ワーカープール vs ファンアウト／ファンイン

6. Or-Doneチャンネル

▶ サンプル：Or-Doneモード

7. errgroup エラーの伝播

▶ サンプル：errgroup

▶ サンプル：errgroup ＋ コンテキストタイムアウト

(2) sync.WaitGroup と errgroup の比較

8. 完全な例：リアルタイムデータ処理パイプライン

❓ よくある質問

📖 まとめ

📝 練習問題

▶ サンプル：errgroup ＋コンテキストタイムアウト