並行処理パターンと同期

.NET (C#) の経験がある開発者向けに、sync.WaitGroup、Mutex、errgroup、worker pool、race detector を解説します。C# では Task と async/await、Parallel、Channel<T> などで並行処理を組み立てますが、Go ではゴルーチン、チャネル、sync パッケージ、そして準標準ライブラリの errgroup が同じ役割を担います。戻り値を持つ Task が存在しないため、「完了をどう待つか」「結果とエラーをどう集めるか」を自分で設計する点が最大の違いです。

1. このドキュメントで扱うこと

• WaitGroup と errgroup
• Worker pool / fan-out / fan-in
• go test -race によるデータ競合検出
• 共有メモリ vs メッセージパッシング

WaitGroup と errgroup

sync.WaitGroup は「起動したゴルーチンがすべて終わるまで待つ」ための最小の道具で、Task.WhenAll の素朴な代替です。ただし戻り値もエラーも運べないため、結果は共有変数やチャネルで受け取る必要があります。

errgroup は golang.org/x/sync/errgroup パッケージ (標準ライブラリではなく go get golang.org/x/sync で追加するモジュール) で提供され、次の機能を WaitGroup に足したものです。

• g.Go(func() error) でエラーを返せるゴルーチンを起動できる。
• g.Wait() が最初に発生した非 nil のエラーを返す。
• errgroup.WithContext(ctx) を使うと、いずれかがエラーを返した時点で派生した ctx がキャンセルされ、他のゴルーチンへ中断を伝えられる。
• g.SetLimit(n) で同時に動くゴルーチン数を制限できる。上限に達している間、g.Go は呼び出し側をブロックする。

実務では「複数の処理を並行に走らせ、1 つでも失敗したら全体を失敗として扱う」場面の大半で、WaitGroup より errgroup が適しています。なお、Go 1.25 では sync.WaitGroup に Go(func()) メソッドが追加され、Add と Done の対応を手書きしなくて済むようになりました。利用する Go のバージョンを確認してください。

Worker pool / fan-out / fan-in

• Worker pool: 固定数のゴルーチンが 1 本のチャネル (ジョブキュー) から仕事を取り出して処理する構成です。C# の Parallel.ForEachAsync に MaxDegreeOfParallelism を指定した動き、あるいは Channel<T> を複数のコンシューマーで読む構成に相当します。
• Fan-out: 1 つの入力チャネルを複数のワーカーが同時に読み、仕事を分散させることです。
• Fan-in: 複数のチャネルの出力を 1 つのチャネルに集約することです。

ワーカー数を決めたいだけなら errgroup + SetLimit が最も簡潔です。段階的な処理 (パイプライン) や、結果をストリームとして処理したい場合にチャネルベースの worker pool を選びます。

go test -race によるデータ競合検出

Go には実行時にデータ競合を検出する race detector が付属しています。go test -race ./...、go run -race、go build -race で有効になります。実際に競合が起きた実行経路だけが報告される (見つからないことは安全の証明ではない) ため、並行処理を含むテストを書いたうえで CI で常に -race を付けて実行するのが定石です。実行時間とメモリ使用量は増えるので、本番バイナリに常時付けるものではありません。

C# にはこれと同等の標準ツールがないため、Go では特に重要な習慣です。

共有メモリ vs メッセージパッシング

Go の格言に「メモリを共有することで通信するな。通信することでメモリを共有せよ」があります。所有権をチャネルで受け渡す設計が推奨されますが、これは「常にチャネルを使え」という意味ではありません。

状況	向いている手段
状態 (カウンター、キャッシュ、設定) を複数のゴルーチンが読み書きする	sync.Mutex / sync.RWMutex / sync/atomic
データを段階的に流して加工する、仕事を配る	チャネル
完了を待つ、エラーを集める	WaitGroup / errgroup
キャンセルを伝える	context.Context

単純な共有状態の保護にチャネルを持ち出すと、かえって複雑になります。「所有者が 1 つのゴルーチンに決まるならチャネル、複数が触るならロック」が目安です。

2. .NET との対応表

.NET (C#)	Go
lock / Monitor	sync.Mutex / sync.RWMutex
Task.WhenAll	sync.WaitGroup / errgroup.Group
Parallel.ForEachAsync	worker pool パターン
ConcurrentDictionary	sync.Map / Mutex + map

上の表を補う対応関係です。

.NET (C#)	Go
Task.Run(() => ...)	go func() { ... }()
Task<T> (戻り値付きの非同期処理)	チャネル、または共有変数 + WaitGroup / errgroup
Parallel.ForEachAsync の MaxDegreeOfParallelism	errgroup.Group.SetLimit(n)
Channel<T> (System.Threading.Channels)	chan T (組み込み)
SemaphoreSlim	バッファ付きチャネル、golang.org/x/sync/semaphore
Interlocked.Increment	sync/atomic (atomic.Int64 など)
Lazy<T> / 一度だけ初期化	sync.Once / sync.OnceValue (Go 1.21 以降)
ReaderWriterLockSlim	sync.RWMutex
AggregateException	errors.Join (複数のエラーをまとめる)、errgroup は最初のエラーのみ返す
(標準のデータ競合検出ツールはない)	go test -race (race detector)

3. C# 開発者がつまずきやすいポイント

戻り値を持つ Task がない

C# では await Task.WhenAll(tasks) で結果の配列を受け取れます。Go のゴルーチンは戻り値を持たないので、結果はあらかじめ確保したスライスの各要素に書く、またはチャネルで送る形になります。要素ごとに書き込み先が異なるスライスへの書き込みは、ロックなしで安全です。

ゴルーチンのリーク

ゴルーチンは GC に回収されません。受信側がいなくなったチャネルへ送信し続ける、誰も閉じないチャネルを range で待ち続ける、といった状況でゴルーチンが残り続け、メモリを消費します。起動するすべてのゴルーチンについて、「いつ、何をきっかけに終わるか」を決めてから書く必要があります。context のキャンセルと select を組み合わせて終了条件を作るのが基本です。

ループ変数の捕捉 (Go 1.22 で変更)

Go 1.21 以前では、for のループ変数は全反復で 1 つの変数が共有されていたため、go func() { use(v) }() のように捕捉すると、すべてのゴルーチンが最後の値を見る問題が有名でした。Go 1.22 以降 (go.mod の go ディレクティブが 1.22 以上のモジュール) では反復ごとに新しい変数が作られるため、この問題は起きません。古い go.mod のままだと従来の挙動になるので、既存プロジェクトでは go ディレクティブの値を確認してください。

WaitGroup の Add はゴルーチンの外で呼ぶ

wg.Add(1) をゴルーチンの内側で呼ぶと、wg.Wait() が先に実行されて待たずに通過することがあります。Add は go 文の直前に呼びます。また WaitGroup や Mutex は値渡しでコピーしてはいけません。ポインターで渡すか、構造体のフィールドとして持ちます。go vet の copylocks が検出します。

チャネルを閉じるのは送信側

チャネルの close は送信側が行い、受信側は閉じません。閉じたチャネルへの送信と、二重 close は panic になります。複数の送信者がいる場合は、全送信者の完了を WaitGroup で待ってから 1 か所で閉じます。受信側は for v := range ch で、閉じられるまで読み続けられます。

map の並行アクセスは致命的エラーになる

通常の map を複数のゴルーチンから同時に書き込む (または書き込みと読み取りを同時に行う) と、ランタイムが fatal error: concurrent map writes などで即座にプロセスを終了させます。recover では捕捉できません。ConcurrentDictionary のような組み込みの並行 map はないため、sync.Mutex / sync.RWMutex + map を使うのが基本です。sync.Map は「キーが一度だけ書かれて何度も読まれる」「ゴルーチンごとに異なるキー集合を扱う」といった限られた用途向けです。型安全性も弱いため、まずは Mutex + map を検討してください。

Mutex は再入不可

C# の lock は同じスレッドからの再入を許しますが、Go の sync.Mutex は再入できません。ロックを持ったまま同じロックを取ろうとすると、デッドロックします。ロック内から呼ぶ関数が再びロックを取らない設計にしてください。ロックの解放は defer mu.Unlock() が定石です。

errgroup は最初のエラーだけを返す

g.Wait() が返すのは最初に発生した非 nil のエラーです。すべてのエラーを集めたい場合は、各ゴルーチンのエラーを自分で収集して errors.Join でまとめます。また、WithContext の ctx は Wait が返った時点でキャンセルされるため、Wait 後に流用してはいけません。

並行度は無制限にしない

for で大量の項目に対してゴルーチンを起動するのは Go では安価ですが、各ゴルーチン内で DB 接続や HTTP を呼ぶ場合は、接続数や相手先への負荷が問題になります。errgroup.SetLimit や worker pool で上限を設けます。

4. コード例

例 1: WaitGroup と Mutex による集計

C#:

var total = 0;
var gate = new object();

await Task.WhenAll(Enumerable.Range(1, 5).Select(n => Task.Run(() =>
{
    lock (gate)
    {
        total += n;
    }
})));

Console.WriteLine(total);

Go:

package main

import (
	"fmt"
	"sync"
)

func main() {
	var (
		mu    sync.Mutex
		wg    sync.WaitGroup
		total int
	)

	for n := 1; n <= 5; n++ {
		wg.Add(1) // ゴルーチンの外で Add する
		go func() {
			defer wg.Done()
			// Go 1.22 以降は n が反復ごとの変数なので、そのまま捕捉できる
			mu.Lock()
			defer mu.Unlock()
			total += n
		}()
	}

	wg.Wait()
	fmt.Println(total) // 15
}

単純なカウンターなら、sync/atomic の atomic.Int64 を使えばロックも不要です。

var total atomic.Int64
// ゴルーチン内で
total.Add(int64(n))
// 最後に
fmt.Println(total.Load())

例 2: errgroup による上限付きの並行処理

C# の Parallel.ForEachAsync に MaxDegreeOfParallelism を指定する場面です。

C#:

var urls = new[] { "a", "b", "c", "d", "e" };
var results = new string[urls.Length];

await Parallel.ForEachAsync(
    urls.Select((u, i) => (u, i)),
    new ParallelOptions { MaxDegreeOfParallelism = 2, CancellationToken = ct },
    async (item, token) =>
    {
        results[item.i] = await FetchAsync(item.u, token);
    });

Go:

package main

import (
	"context"
	"fmt"
	"time"

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

// fetch は外部呼び出しの代わりのダミー実装
func fetch(ctx context.Context, url string) (string, error) {
	select {
	case <-time.After(100 * time.Millisecond):
		if url == "d" {
			return "", fmt.Errorf("fetch %s: boom", url)
		}
		return "body of " + url, nil
	case <-ctx.Done():
		return "", ctx.Err()
	}
}

func fetchAll(ctx context.Context, urls []string) ([]string, error) {
	g, ctx := errgroup.WithContext(ctx) // 1 つ失敗すると ctx がキャンセルされる
	g.SetLimit(2)                       // 同時実行は最大 2

	results := make([]string, len(urls))
	for i, u := range urls {
		g.Go(func() error {
			body, err := fetch(ctx, u)
			if err != nil {
				return err
			}
			results[i] = body // 要素ごとに書き込み先が違うのでロック不要
			return nil
		})
	}

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

func main() {
	res, err := fetchAll(context.Background(), []string{"a", "b", "c", "d", "e"})
	fmt.Println(res, err)
}

SetLimit を指定すると、上限に達している間は g.Go が呼び出し側をブロックします。ブロックしたくない場合は、起動できたかどうかを bool で返す g.TryGo を使います。

例 3: Worker pool (fan-out / fan-in)

C# では Channel<T> を複数のコンシューマーで読む構成に近い例です。

package main

import (
	"fmt"
	"sync"
)

type result struct {
	n  int
	sq int
}

func worker(jobs <-chan int, results chan<- result, wg *sync.WaitGroup) {
	defer wg.Done()
	for n := range jobs { // jobs が閉じられるまで受信する
		results <- result{n: n, sq: n * n}
	}
}

func main() {
	jobs := make(chan int)
	results := make(chan result)

	// fan-out: 3 つのワーカーが同じ jobs チャネルを読む
	var wg sync.WaitGroup
	for range 3 { // Go 1.22 以降は整数に対する range が使える
		wg.Add(1)
		go worker(jobs, results, &wg)
	}

	// 送信側: 仕事を送って jobs を閉じる (送信側が閉じる)
	go func() {
		defer close(jobs)
		for n := 1; n <= 9; n++ {
			jobs <- n
		}
	}()

	// fan-in: 全ワーカーの終了後に results を閉じる
	go func() {
		wg.Wait()
		close(results)
	}()

	sum := 0
	for r := range results {
		sum += r.sq
	}
	fmt.Println("sum of squares:", sum) // 285
}

要点は次の 3 つです。

• jobs を閉じるのは送信側、results を閉じるのは「全ワーカーの終了を待つ」専用のゴルーチンです。
• 受信側は range で、閉じられるまで読み続けます。
• チャネルの向きを <-chan int (受信専用) / chan<- result (送信専用) と型で制限すると、誤用をコンパイル時に防げます。

例 4: データ競合と race detector

次のコードは、複数のゴルーチンが同期なしで counter を更新するため、データ競合を含みます。

package main

import (
	"fmt"
	"sync"
)

func main() {
	counter := 0
	var wg sync.WaitGroup

	for range 1000 {
		wg.Add(1)
		go func() {
			defer wg.Done()
			counter++ // 競合: 読み取り・加算・書き込みがアトミックではない
		}()
	}

	wg.Wait()
	fmt.Println(counter) // 1000 になるとは限らない
}

go run -race main.go で実行すると、WARNING: DATA RACE と、競合した読み書きのスタックトレースが出力されます。テストでは次のように実行します。

go test -race ./...

修正は sync.Mutex で保護するか、atomic.Int64 に置き換えます。

例 5: RWMutex で保護するスレッドセーフなキャッシュ

ConcurrentDictionary の代わりに、構造体で map とロックをまとめて隠蔽する書き方です。

package main

import (
	"fmt"
	"sync"
)

type Cache struct {
	mu   sync.RWMutex
	data map[string]string
}

func NewCache() *Cache {
	return &Cache{data: make(map[string]string)}
}

func (c *Cache) Get(key string) (string, bool) {
	c.mu.RLock() // 読み取りは同時に複数可
	defer c.mu.RUnlock()
	v, ok := c.data[key]
	return v, ok
}

func (c *Cache) Set(key, value string) {
	c.mu.Lock() // 書き込みは排他
	defer c.mu.Unlock()
	c.data[key] = value
}

func main() {
	c := NewCache()
	c.Set("lang", "go")
	v, ok := c.Get("lang")
	fmt.Println(v, ok)
}

Cache は Mutex を含むため、値渡しでコピーせず、ポインターで扱います。

5. まとめ

• ゴルーチンは戻り値を持たないため、完了待ちと結果の集約を WaitGroup、errgroup、チャネルで設計します。
• エラーを返す並行処理や並行度の制限には golang.org/x/sync/errgroup (WithContext と SetLimit) が最も簡潔です。
• 仕事を配る構成は worker pool で、チャネルは送信側が閉じ、結果側は全ワーカーの終了後に 1 か所で閉じます。
• 共有状態の保護には Mutex / RWMutex / atomic を使い、所有権の受け渡しや段階的な処理にはチャネルを使います。
• 通常の map は並行アクセスで致命的エラーになります。Mutex + map を基本とし、Mutex は再入できません。
• ゴルーチンは必ず終了条件を持たせ、リークを避けます。Go 1.22 以降はループ変数が反復ごとに作られます。
• go test -race を CI で常時有効にし、データ競合を早期に検出します。

参考資料

• Go Documentation
• A Tour of Go
• Effective Go
• sync package (pkg.go.dev)
• errgroup package (pkg.go.dev)
• Data Race Detector (go.dev)
• Go Concurrency Patterns: Pipelines and cancellation (go.dev/blog)
• Fixing For Loops in Go 1.22 (go.dev/blog)