sync.WaitGroup、Mutex、errgroup：Goでゴルーチンを調整する

1. 概要

Goでゴルーチンを実行するのは簡単で、go キーワードを追加するだけです。より難しいのは、ゴルーチンがいつ完了したかを知る、共有データを保護する、そしてゴルーチンが失敗した場合にエラーを処理することです。

この記事では、これら3つのタスク用の3つのツール：sync.WaitGroup、sync.Mutex、errgroup.Group を紹介します。読み終わったら、どの状況でどのツールを選ぶべきか、-race を使ってテストを実行する方法、そして避けるべき一般的なエラーが分かります。

例ではGo 1.22以上を使用しています。このバージョンから、各 for ループの反復ごとに新しい変数が作成されるため、id := id を書く必要がなくなりました。

2. サンプル問題

注文コード（Order）のリストがあるとします。各注文は遅いサービスを呼び出す必要があり、約100msかかります。10個の注文を順番に呼び出すと、約1秒かかります。並行に呼び出すと、約100msだけかかります。

これはサービス呼び出しをシミュレートする関数です：

package main

import (
	"fmt"
	"time"
)

func fetchOrder(id int) string {
	time.Sleep(100 * time.Millisecond)
	return fmt.Sprintf("order-%d", id)
}

ループ内で go fetchOrder(id) と書くだけの場合、main 関数はゴルーチンが完了する前に終了します。プログラムは終了し、結果が失われます。次のセクションでこの問題に対処します。

3. sync.WaitGroup：ゴルーチンのグループを待つ

3.1. 基本的な使い方

WaitGroup はカウンタです。Add でカウンタを増やし、Done で減らし、Wait はカウンタが0になるまでブロックします。

func main() {
	var wg sync.WaitGroup
	ids := []int{1, 2, 3}

	for _, id := range ids {
		wg.Add(1)
		go func() {
			defer wg.Done()
			fmt.Println("fetched", fetchOrder(id))
		}()
	}

	wg.Wait()
	fmt.Println("all done")
}

上記のコードは次のように動作します：

• wg.Add(1) はゴルーチンを開始する前に、親ゴルーチン内で呼ばれます。
• defer wg.Done() はゴルーチンが早期に終了したりpanicしても、カウンタが必ず減ることを保証します。
• wg.Wait() は3つすべてのゴルーチンが Done を呼ぶまで、main をブロックします。

結果は次のようになります。fetched 行の順序は実行ごとに変わります：

fetched order-2
fetched order-1
fetched order-3
all done

3.2. Go 1.25のWaitGroup.Go

Go 1.25から、WaitGroup に Go メソッドが追加されました。このメソッドは Add、go、Done を1つの呼び出しに統合します：

var wg sync.WaitGroup
for _, id := range ids {
	wg.Go(func() {
		fmt.Println("fetched", fetchOrder(id))
	})
}
wg.Wait()

こう書く方が短く、2つの一般的なエラーを避けられます：Done の呼び忘れ、または Add をゴルーチン内で呼ぶこと。

3.3. ロックなしに結果を収集する

通常、結果を保持する必要があります。画面に出力するだけではなく。安全な方法は、あらかじめ長さを持つsliceを作成し、各ゴルーチンに独自のインデックスに書き込ませることです：

results := make([]string, len(ids))
var wg sync.WaitGroup
for i, id := range ids {
	wg.Go(func() {
		results[i] = fetchOrder(id)
	})
}
wg.Wait()
fmt.Println(results)

[order-1 order-2 order-3]

ゴルーチンの中で同じ要素に書き込むことはないため、ここではデータレースがありません。wg.Wait() は happens-before の関係も作成します。したがって、Wait の後、親ゴルーチンは子ゴルーチンが書き込んだ全データを読むことができます。

注意： ゴルーチン内で wg.Add(1) を呼ばないでください。そうすると、Wait() は Add の前に実行され、作業が完了する前に返る可能性があります。

4. sync.Mutex：共有データを保護する

4.1. データレースの見え方

今、顧客（Customer）ごとの売上を累積し、複数のゴルーチンが同じmapを更新する必要があるとします：

revenue := map[string]float64{}
var wg sync.WaitGroup
for i := 0; i < 1000; i++ {
	wg.Go(func() {
		revenue["alice"] += 10
	})
}
wg.Wait()

このプログラムは次のエラーでクラッシュする可能性があります：

fatal error: concurrent map writes

単純な int 変数を使う場合はさらに危険です。プログラムはクラッシュしませんが、例えば10000の代わりに9730のように、静かに間違った結果を出力します。この種のエラーを検出するには、race detectorで実行します：

go run -race main.go

WARNING: DATA RACE
Write at 0x00c000012345 by goroutine 8:
...

4.2. Mutexで修正

sync.Mutex は、各時点で1つのゴルーチンだけが Lock と Unlock の間にあるコードを実行できることを保証します。一般的な方法は、mutexを保護する同じデータと共に構造体にバンドルすることです：

package shop

import "sync"

type RevenueTracker struct {
	mu    sync.Mutex
	total map[string]float64
}

func NewRevenueTracker() *RevenueTracker {
	return &RevenueTracker{total: make(map[string]float64)}
}

func (r *RevenueTracker) Add(customer string, amount float64) {
	r.mu.Lock()
	defer r.mu.Unlock()
	r.total[customer] += amount
}

func (r *RevenueTracker) Total(customer string) float64 {
	r.mu.Lock()
	defer r.mu.Unlock()
	return r.total[customer]
}

注意すべき点がいくつかあります：

• total へのすべてのアクセス、読み込みでさえ、ロックを通す必要があります。
• defer r.mu.Unlock() は、関数が複数の return 分岐を持つ場合でも、ロックが必ず解放されることを保証します。
• メソッドはポインタレシーバを使用します（*RevenueTracker）。値レシーバを使用すると、各呼び出しでmutex全体がコピーされ、そのロックは何も保護しなくなります。

読み取りが書き込みより圧倒的に多い場合、sync.RWMutex を使用できます。読み込みメソッドは RLock/RUnlock を呼び、書き込みメソッドは Lock/Unlock を呼びます。複数のゴルーチンは同時に読み取ることができ、書き込み時は1つのゴルーチンだけロックを保持します。

注意： 使用した後、Mutex を含む構造体をコピーしないでください。go vet はこのケースに対して copylocks メッセージで警告します。また、Goには再入可能ロックがありません。ゴルーチンが同じmutexで Lock() を2回呼び出すと、デッドロックします。

4.3. Mutexが不要な場合

単純なカウンタの場合、sync/atomic の方がコンパクトで高速です：

var processed atomic.Int64
processed.Add(1)
fmt.Println(processed.Load())

データがゴルーチンからゴルーチンへストリーム形式で流れる必要がある場合、channelの方が一般的です。Mutexは、mapや構造体のような共有状態があり、複数のゴルーチンが読み書きする必要がある場合に最適です。

5. errgroup：エラー処理を含むWaitGroup

WaitGroup はエラーについて何も知りません。fetchOrder がエラーを返す可能性がある場合、mutexまたはchannelを使ってエラーを自分で集約し、残りのゴルーチンを自分でキャンセルする必要があります。errgroup パッケージはこれらを事前に処理します。

5.1. 依存関係を追加

errgroup は golang.org/x/sync モジュール内にあり、標準ライブラリの一部ではありません：

go get golang.org/x/sync

5.2. エラーを集約してcontextでキャンセル

最初のステップは、fetchOrder を書き直して関数が context.Context を受け取り、エラーを返すことができるようにすることです：

package shop

import (
	"context"
	"fmt"
	"time"

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

type Order struct {
	ID       int
	Customer string
}

func fetchOrder(ctx context.Context, id int) (Order, error) {
	if id < 0 {
		return Order{}, fmt.Errorf("invalid order id %d", id)
	}
	select {
	case <-time.After(100 * time.Millisecond):
		return Order{ID: id, Customer: "alice"}, nil
	case <-ctx.Done():
		return Order{}, ctx.Err()
	}
}

func LoadOrders(ctx context.Context, ids []int) ([]Order, error) {
	g, ctx := errgroup.WithContext(ctx)
	g.SetLimit(3)

	orders := make([]Order, len(ids))
	for i, id := range ids {
		g.Go(func() error {
			o, err := fetchOrder(ctx, id)
			if err != nil {
				return err
			}
			orders[i] = o
			return nil
		})
	}

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

上記のコードは次のように動作します：

• errgroup.WithContext(ctx) はグループとチャイルドcontextを作成します。最初のゴルーチンがエラーを返すと、チャイルドcontextはすぐにキャンセルされます。
• g.SetLimit(3) は最大3つのゴルーチンが同時に実行されることを許可します。次の g.Go 呼び出しはスペースが利用可能になるまでブロックします。これにより、リストに数千の注文コードがある場合、バックエンドサービスに過負荷をかけません。
• g.Go は func() error 型の関数を受け取り、Add/Done を呼ぶ必要がありません。
• g.Wait() はすべてのゴルーチンが終了するのを待ってから、最初のエラーを返します。その後に発生するエラーは無視されます。
• fetchOrder は ctx.Done() をリッスンするため、残りのゴルーチンは100ms全体を実行する代わりに早期に停止します。

無効な注文コードで関数を呼び出します：

_, err := shop.LoadOrders(context.Background(), []int{1, 2, -5, 4})
fmt.Println(err)

invalid order id -5

制限に達したときにブロックせずに実行してみたい場合は、g.TryGo(f) を使用します。この関数は待つ代わりに false を返します。

注意： contextのキャンセルは、ゴルーチン内のコードが 主動的にチェックする場合にのみ有効です。fetchOrder が time.Sleep を使用し、contextをリッスンしない場合、グループが失敗した場合でも、ゴルーチンは完全な時間を実行します。

6. -raceでテストを書く

Goには相当するJUnitやAssertJがありません。標準ライブラリの testing パッケージを使用します。最も重要なことは、常に -race フラグでテストを実行することです。多くのconcurrencyバグはrace detectorがある場合にのみ表示されるためです。

package shop

import (
	"context"
	"sync"
	"testing"
)

func TestRevenueTracker_ConcurrentAdd(t *testing.T) {
	tracker := NewRevenueTracker()
	var wg sync.WaitGroup

	for i := 0; i < 1000; i++ {
		wg.Go(func() {
			tracker.Add("alice", 10)
		})
	}
	wg.Wait()

	if got := tracker.Total("alice"); got != 10000 {
		t.Fatalf("Total(alice) = %v, want 10000", got)
	}
}

func TestLoadOrders_ReturnsAllOrders(t *testing.T) {
	orders, err := LoadOrders(context.Background(), []int{1, 2, 3})
	if err != nil {
		t.Fatalf("unexpected error: %v", err)
	}
	for i, o := range orders {
		if o.ID != i+1 {
			t.Errorf("orders[%d].ID = %d, want %d", i, o.ID, i+1)
		}
	}
}

func TestLoadOrders_StopsOnFirstError(t *testing.T) {
	orders, err := LoadOrders(context.Background(), []int{1, -1, 3})
	if err == nil {
		t.Fatal("expected error, got nil")
	}
	if orders != nil {
		t.Errorf("expected nil orders, got %v", orders)
	}
}

テストを実行します：

go test -race ./...

ok  	example.com/shop	0.412s

Add 内の Lock/Unlock の2行を削除してから再度実行してみてください。テストは WARNING: DATA RACE で失敗するか、concurrent map writes のためにクラッシュします。これは、mutexが本当に必要であることを確認する高速な方法です。

2番目のテストは、結果の順序が正しく保持されていることも確認します：orders[i] は ids[i] と一致します。ゴルーチンはどの順序でも終了する可能性があります。

7. 一般的なエラーとツール選択

最も一般的なエラーは次のとおりです：

• ゴルーチン内で Add を呼ぶ。 Wait が早期に返される可能性があります。go コマンドの前に Add を呼ぶか、wg.Go を使用します。
• WaitGroup または Mutex を値で渡す。 関数は受け取ったコピーを使用し、元のカウンタまたはロックは変わりません。ポインタ *sync.WaitGroup を渡すか、closureを使用します。
• I/O呼び出し中にロックを保持する。 例えば、Lock 内でHTTPを呼び出します。他のすべてのゴルーチンはキューに入る必要があり、プログラムは並行実行の利点を失います。ロックは最短のコード部分でのみ保持します。
• 空の errgroup.Group を使用（var g errgroup.Group）しながら、contextがキャンセルされることを期待する。 WithContext だけがキャンセル可能なcontextを作成します。
• g.Wait() の戻り値を無視する。 その場合、エラーは誰にも知られずに消えます。
• 数十万のリクエストに対して無制限のゴルーチンを起動する。 SetLimit を使用するか、ワーカープールを使用します。

ツール選択ガイド：

状況	ツール
エラーを返さない複数のタスクを待つ	sync.WaitGroup
複数のゴルーチンが共有mapまたは構造体を変更する	sync.Mutex / sync.RWMutex
シンプルなカウンタまたはフラグ	sync/atomic
失敗する可能性のあるタスク、早期キャンセル、ゴルーチン数制限	errgroup

注意： errgroup は最初のエラーだけを保持します。すべてのエラーを知る必要がある場合、例えば失敗した注文を報告するには、各ゴルーチンにエラーを errs[i] に書き込ませ、Wait 後に errors.Join(errs...) で統合します。

8. まとめ

sync.WaitGroup は「いつ完了するか」という質問に答えます。sync.Mutex は「今、誰がデータを変更できるか」という質問に答えます。errgroup は待機とエラー処理を統合し、contextでキャンセルし、同時に実行するゴルーチン数を制限します。

実際のコードでは、複数のサービスを並行に呼び出す場合、errgroup.WithContext に SetLimit を組み合わせるのが通常、合理的なデフォルト選択です。どのツールを使用する場合でも、常に go test -race を実行してください。これは、本番環境に到達する前にデータレースを検出する最もコスト効率的な方法です。