golang-concurrency-patterns

Go の並行処理パターン（本番環境）

概要

Go の並行処理は、ゴルーチンのライフタイムを明示し、キャンセルを context.Context で伝播させ、共有状態を保護（チャネルまたはロック）することでスケールします。これらのパターンを適用して信頼性の高いサービスを構築し、ゴルーチンのリーク、デッドロック、データ競合といった一般的な障害を回避しましょう。

クイックスタート

基本となる構成要素

• キャンセルと期限の制御には context を使用します。
• ファンアウト／ファンインを早期中断可能な形で実行するには errgroup.WithContext を使用します。
• セマフォまたはワーカープールを使用して並行処理数を制限します（無制限にゴルーチンを起動しないでください）。
• 不変データを優先します。それ以外の場合は、共有状態をミューテックスで保護するか、単一のゴルーチンを所有者にします。

避けるべきこと

• リクエストハンドラー内で、起動後に結果を待たないゴルーチン。
• 高頻度で実行されるループ内での time.After。
• 受信側からチャネルを閉じること。
• 同期せずに、可変変数を複数のゴルーチン間で共有すること。

主な概念

ゴルーチンのライフサイクル

ゴルーチンは、明確な所有者と終了条件を持つリソースとして扱います。

✅ 正しい例：context でゴルーチンを停止する

ctx, cancel := context.WithCancel(context.Background())
defer cancel()

go func() {
    ticker := time.NewTicker(250 * time.Millisecond)
    defer ticker.Stop()

    for {
        select {
        case <-ctx.Done():
            return
        case <-ticker.C:
            // do work
        }
    }
}()

❌ 誤った例：停止条件のないゴルーチン

go func() {
    for {
        doWork() // leaks forever
    }
}()

チャネルとミューテックス（意図を持って選ぶ）

• 状態の所有権や直列化を表現したり、作業をパイプライン化したりするには、チャネルを使用します。
• 単純な読み書きパターンで共有メモリ内の状態を保護するには、ミューテックスを使用します。

✅ 正しい例：単一のゴルーチンがマップを所有する

type req struct {
    key   string
    reply chan<- int
}

func mapOwner(ctx context.Context, in <-chan req) {
    m := map[string]int{}
    for {
        select {
        case <-ctx.Done():
            return
        case r := <-in:
            r.reply <- m[r.key]
        }
    }
}

✅ 正しい例：ミューテックスで共有マップを保護する

type SafeMap struct {
    mu sync.RWMutex
    m  map[string]int
}

func (s *SafeMap) Get(k string) (int, bool) {
    s.mu.RLock()
    defer s.mu.RUnlock()
    v, ok := s.m[k]
    return v, ok
}

パターン

1) キャンセルを伴うファンアウト／ファンイン（errgroup）

errgroup.WithContext を使用すると、タスクを並行して実行し、エラー発生時にほかのタスクをキャンセルして、完了を待機できます。

✅ 正しい例：最初のエラーでキャンセルする

g, ctx := errgroup.WithContext(ctx)

for _, id := range ids {
    id := id // capture
    g.Go(func() error {
        return process(ctx, id)
    })
}

if err := g.Wait(); err != nil {
    return err
}

❌ 誤った例：WaitGroup は最初のエラーを失い、キャンセルも伝播しない

var wg sync.WaitGroup
for _, id := range ids {
    wg.Add(1)
    go func() {
        defer wg.Done()
        _ = process(context.Background(), id) // ignores caller ctx + captures id
    }()
}
wg.Wait()

2) 並行処理数の制限（セマフォパターン）

並列処理を制限して、CPU やメモリの枯渇、下流システムへの過負荷を防ぎます。

✅ 正しい例：並行数を制限したファンアウト

limit := make(chan struct{}, 8) // max 8 concurrent
g, ctx := errgroup.WithContext(ctx)

for _, id := range ids {
    id := id
    g.Go(func() error {
        select {
        case <-ctx.Done():
            return ctx.Err()
        case limit <- struct{}{}:
        }
        defer func() { <-limit }()

        return process(ctx, id)
    })
}

return g.Wait()

3) ワーカープール（安定したスループット）

固定数のワーカーを使用して、安定したスループットと予測可能なリソース使用量を実現します。

✅ 正しい例：context で停止するワーカープール

type Job struct{ ID string }

func runPool(ctx context.Context, jobs <-chan Job, workers int) error {
    g, ctx := errgroup.WithContext(ctx)

    for i := 0; i < workers; i++ {
        g.Go(func() error {
            for {
                select {
                case <-ctx.Done():
                    return ctx.Err()
                case j, ok := <-jobs:
                    if !ok {
                        return nil
                    }
                    if err := handleJob(ctx, j); err != nil {
                        return err
                    }
                }
            }
        })
    }

    return g.Wait()
}

4) パイプラインの各段階（段階間のファンアウト）

一方向のチャネルを優先し、送信側だけがチャネルを閉じるようにします。

✅ 正しい例：送信側がチャネルを閉じる

func stageA(ctx context.Context, out chan<- int) {
    defer close(out)
    for i := 0; i < 10; i++ {
        select {
        case <-ctx.Done():
            return
        case out <- i:
        }
    }
}

❌ 誤った例：受信側がチャネルを閉じる

func stageB(in <-chan int) {
    close(in) // compile error in<-chan; also wrong ownership model
}

5) リークのない定期処理（time.Ticker と time.After）

ループには time.NewTicker を使用し、高頻度で実行される処理での time.After の割り当ては避けます。

✅ 正しい例：ticker

t := time.NewTicker(1 * time.Second)
defer t.Stop()

for {
    select {
    case <-ctx.Done():
        return
    case <-t.C:
        poll()
    }
}

❌ 誤った例：ループ内での time.After

for {
    select {
    case <-ctx.Done():
        return
    case <-time.After(1 * time.Second):
        poll()
    }
}

判断基準

チャネルとミューテックス

• 所有権／直列化（単一の書き込み元、メッセージパッシング）が必要 → チャネル + 所有者ゴルーチンを使用します
• 多数の読み取り元と単純な更新を伴う共有キャッシュ／マップが必要 → RWMutexを使用します
• 競合が少ない単純なカウンターが必要 → atomicを使用します

WaitGroup と errgroup

• エラーの伝播 + ほかのタスクのキャンセルが必要 → **errgroup.WithContext**を使用します
• 待機だけが必要で、エラーは別の場所で処理する → **sync.WaitGroup**を使用します

バッファ付きチャネルとバッファなしチャネル

• バックプレッシャーと同期的な受け渡しが必要 → バッファなしを使用します
• 既知のサイズまでのバーストを吸収する必要がある → バッファ付きを使用します（意図を持ってサイズを決めます）
• 判断がつかない → まずバッファなしで始めて計測し、既知のボトルネックを解消する場合にのみバッファを追加します

テストと検証

レースディテクターとテストの不安定さへの対処

デバッグ中は、レースディテクターを使用して対象を絞ったテストを実行し、キャッシュを無効にします。

go test -race ./...
go test -run TestName -race -count=1 ./...

テストのハングを防ぐタイムアウト

✅ 正しい例：context によるテストレベルのタイムアウト

func TestSomething(t *testing.T) {
    ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
    defer cancel()

    if err := doThing(ctx); err != nil {
        t.Fatal(err)
    }
}

トラブルシューティング

症状：デッドロック（テストがハングし、ゴルーチンがブロックされる）

対処方法：

• ブロックする処理の周囲にタイムアウト（context.WithTimeout）を追加します。
• チャネルの所有権を確認します。閉じるのは送信側だけです。受信側は ok == false で停止します。
• セマフォパターンで <-limit の解放が抜けていないか確認します。

症状：データ競合（go test -race で報告される）

対応：

• 複数のゴルーチンから変更される共有変数を特定します。
• ミューテックスを追加するか、所有権モデル（単一のゴルーチンが状態を所有する）に切り替えます。
• キャプチャされたループ変数への書き込みを避けます。

症状：ゴルーチンのリーク（メモリの増加、シャットダウンの遅延）

対応：

• すべてのゴルーチンで ctx.Done() を監視するようにします。
• time.Ticker を停止し、チャネルは送信側で閉じるようにします。
• リクエスト処理中に context.Background() を使わず、呼び出し元のコンテキストを引き継ぎます。

リソース

• Go ブログ：並行処理のパターン：https://go.dev/blog/pipelines
• errgroup：https://pkg.go.dev/golang.org/x/sync/errgroup
• Go メモリモデル：https://go.dev/ref/mem