English 中文 العربية 日本語 Português

Go: GoのゴルーチンとWaitGroup：軽量な並行処理、GMPスケジューリング、およびメモリリークの防止

ゴルーチン（Goroutines）は、Go言語における並行プログラミングの基盤です。これらはスレッドでもコルーチンでもなく、Goランタイムによって管理される軽量な並行処理単位です。2 KBのスタック領域を持つため、数百万単位の並行処理を可能にします。

Goの並行処理モデルは他に類を見ないほど優れています。ゴルーチンを開始するには、goというキーワードを指定するだけで済み、Goのランタイムは裏側で何千ものゴルーチンをごく少数のOSスレッドに効率的にマッピングしています。

1. 学習内容

2. データエンジニアの実体験

(1) 課題：10万件のログエントリを単一スレッドで処理するのに5時間かかる

チャーリーは、毎日10万件のサーバーログを処理するデータエンジニアです：

「毎日、夜明けとともにログのクリーンアップ処理が実行され、10万行のデータを次々と処理していく――これには5時間以上かかる。朝のミーティングでは、PMが必ず『昨日のデータはいつ準備できるのか？』と聞いてくる。私が『午後3時です』と答えると、彼らは『なぜ午前9時までに準備できないのか？』と返すのだ。」

彼は現在のコードを開いた：

GO 
// Serial processing: 100k entries × 200ms/entry = 5.5 hours
func processLogs(logs []LogEntry) []Result {
    results := make([]Result, 0, len(logs))
    for _, log := range logs {
        result := processSingleLog(log)  // 200ms each, including network IO
        results = append(results, result)
    }
    return results
}

各ログエントリの処理には、1回の外部API呼び出し（待ち時間は約200ミリ秒）が必要ですが、実際のCPU処理にかかる時間は1ミリ秒未満であり、処理時間の99.5%はネットワークの待ち時間に費やされています。

(2) Goの解決策：Goroutineによる並行処理

GO 
// log_processor.go
package main

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

type LogEntry struct {
    ID      int
    Message string
    Level   string
}

type Result struct {
    ID     int
    Status string
}

func processSingleLog(log LogEntry) Result {
    time.Sleep(200 * time.Millisecond)
    return Result{ID: log.ID, Status: "processed"}
}

func processLogsConcurrent(logs []LogEntry, workerCount int) []Result {
    results := make([]Result, 0, len(logs))
    var mu sync.Mutex
    var wg sync.WaitGroup

    // Use channels for task distribution
    jobs := make(chan LogEntry, len(logs))
    for _, log := range logs {
        jobs <- log
    }
    close(jobs)

    // Start workerCount workers
    for w := 0; w < workerCount; w++ {
        wg.Add(1)
        go func(id int) {
            defer wg.Done()
            for log := range jobs {
                result := processSingleLog(log)
                mu.Lock()
                results = append(results, result)
                mu.Unlock()
            }
        }(w)
    }

    wg.Wait()
    return results
}

func main() {
    // Simulate 100 log entries (demo; can scale to 100k)
    logs := make([]LogEntry, 100)
    for i := 0; i < 100; i++ {
        logs[i] = LogEntry{ID: i, Message: fmt.Sprintf("log-%d", i), Level: "INFO"}
    }

    start := time.Now()
    results := processLogsConcurrent(logs, 10)
    elapsed := time.Since(start)

    fmt.Printf("Processed %d logs in %v\n", len(results), elapsed)
    fmt.Printf("Current goroutine count: %d\n", runtime.NumGoroutine())
}

出力：

TEXT 📖 参照専用 
Processed 100 logs in 2.05s
Current goroutine count: 1

(3) パフォーマンス：順次処理と並行処理の比較

処理方法 100 ログ 100,000 ログ ゴルーチン数
逐次処理 20年代 5.5時間 1
10人の同時実行ワーカー 2秒 33分 約12
100 個の同時実行ワーカー 0.2 秒 3.3 分 約 102
💡 ヒント： ゴルーチンが多いほど良いとは限りません。I/O負荷の高いタスクには worker count = GOMAXPROCS * 2–10 を、CPU負荷の高いタスクには worker count = GOMAXPROCS を推奨します。

3. ゴルーチン基礎

(1) go キーワード

GO 
package main

import (
    "fmt"
    "time"
)

func printNumbers() {
    for i := 1; i <= 5; i++ {
        time.Sleep(100 * time.Millisecond)
        fmt.Printf("%d ", i)
    }
}

func printLetters() {
    for _, c := range "ABCDE" {
        time.Sleep(150 * time.Millisecond)
        fmt.Printf("%c ", c)
    }
}

func main() {
    go printNumbers()  // Execute concurrently in a goroutine
    go printLetters()
    time.Sleep(1 * time.Second)
    fmt.Println()
}

出力（毎回異なる場合があります）：

TEXT 📖 参照専用 
1 A 2 B 3 C 4 D 5 E

(2) 匿名ゴルーチン

GO 
package main

import (
    "fmt"
    "time"
)

func main() {
    go func() {
        fmt.Println("Anonymous goroutine running")
    }()

    go func(msg string) {
        fmt.Println("Anonymous goroutine with parameter:", msg)
    }("hello")

    time.Sleep(100 * time.Millisecond)
    fmt.Println("main done")
}
🔥 よくある間違い： メイン関数が戻ると、すべてのゴルーチンが強制的に終了してしまいます。上記の time.Sleep は、ゴルーチンが終了するのを待つためのものです。本番環境のコードでは、sync.WaitGroup を使用する必要があります。

4. sync.WaitGroup: 同期を待つ

(1) WaitGroup の 3 つのメソッド

GO 
package main

import (
    "fmt"
    "sync"
)

func worker(id int, wg *sync.WaitGroup) {
    defer wg.Done()  // 3. Tell WaitGroup this worker is done
    fmt.Printf("Worker %d started\n", id)
    // Simulate work...
    fmt.Printf("Worker %d done\n", id)
}

func main() {
    var wg sync.WaitGroup

    for i := 1; i <= 3; i++ {
        wg.Add(1)   // 1. Increment counter
        go worker(i, &wg)
    }

    wg.Wait()  // 2. Block until all workers finish
    fmt.Println("All workers done!")
}
メソッド 関数
wg.Add(delta int) カウンタをインクリメントする（通常、ゴルーチン開始前に呼び出される）
wg.Done() カウンタを1減らす（通常、ゴルーチン内の defer 呼び出し）
wg.Wait() カウンタがゼロになるまでブロックする

▶ サンプル：WaitGroup とクロージャの落とし穴

GO 
package main

import (
    "fmt"
    "sync"
)

func main() {
    var wg sync.WaitGroup

    // ❌ Wrong: Closure captures loop variable
    for i := 1; i <= 3; i++ {
        wg.Add(1)
        go func() {
            defer wg.Done()
            fmt.Printf("Wrong i=%d\n", i)  // All print 3 or 4
        }()
    }
    wg.Wait()

    // ✅ Correct: Pass parameter (creates a copy)
    for i := 1; i <= 3; i++ {
        wg.Add(1)
        go func(id int) {
            defer wg.Done()
            fmt.Printf("Correct i=%d\n", id)
        }(i)
    }
    wg.Wait()
}
▶ 試してみよう
🔥 よくある間違い（典型的なクロージャの落とし穴）： ゴルーチン内のクロージャがループ変数 i を直接キャプチャしています。ゴルーチンが開始される時点で、ループはすでに終了しており、i の値が上書きされている可能性があります。パラメータのコピーを渡す必要があります。

5. ゴルーチンとOSスレッドの比較

(1) 主な相違点

GO 
package main

import (
    "fmt"
    "runtime"
)

func main() {
    // View current GOMAXPROCS
    fmt.Printf("GOMAXPROCS: %d\n", runtime.GOMAXPROCS(0))
    fmt.Printf("CPU cores: %d\n", runtime.NumCPU())
    fmt.Printf("Current goroutine count: %d\n", runtime.NumGoroutine())
}
ディメンション OSスレッド ゴルーチン
初期スタックサイズ 1～8 MB 2 KB
スタックの上限 1～8 MBに固定 動的に1 GBまで拡張
生成にかかるオーバーヘッド ~1 µs ~0.1 µs
コンテキストスイッチ カーネルモード（約1 µs） ユーザーモード（約0.1 µs）
最大数 約10,000 百万
スケジューラ OSカーネルのスケジューリング GoランタイムのGMP

(2) GMPスケジューリングモデル

100% 
graph TB
    G[G: Goroutine Queue] --> P[P: Logical Processor<br/>GOMAXPROCS count]
    P --> M[M: OS Thread<br/>Scheduled by kernel]
    M --> CPU[CPU Core]

    subgraph Global Queue
        GQ[(Global Goroutine Queue)]
    end

    GQ --> P

    style P fill:#e1f5fe
    style M fill:#fff3e0

GMP： G（ゴルーチン）— P（プロセッサ、論理プロセッサ；その数は GOMAXPROCS に等しい）— M（マシン、OS スレッド）。Go ランタイムは G を P のローカルキューにスケジューリングし、P はその後に M にバインドして実行を行います。

6. GOMAXPROCS と並行処理制御

(1) GOMAXPROCS の設定

GO 
package main

import (
    "fmt"
    "runtime"
)

func main() {
    fmt.Printf("Default GOMAXPROCS: %d\n", runtime.GOMAXPROCS(0))

    // CPU-intensive: GOMAXPROCS = NumCPU
    // IO-intensive: GOMAXPROCS = NumCPU * 2~10
    runtime.GOMAXPROCS(4)  // Set to 4
    fmt.Printf("After setting GOMAXPROCS: %d\n", runtime.GOMAXPROCS(0))
}

▶ サンプル：GOMAXPROCSがパフォーマンスに与える影響

GO 
package main

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

func cpuIntensiveTask() {
    sum := 0
    for i := 0; i < 10000000; i++ {
        sum += i
    }
}

func benchmarkGOMAXPROCS(n int) time.Duration {
    runtime.GOMAXPROCS(n)
    var wg sync.WaitGroup
    start := time.Now()

    for i := 0; i < 8; i++ {
        wg.Add(1)
        go func() {
            defer wg.Done()
            cpuIntensiveTask()
        }()
    }

    wg.Wait()
    return time.Since(start)
}

func main() {
    for _, p := range []int{1, 2, 4} {
        elapsed := benchmarkGOMAXPROCS(p)
        fmt.Printf("GOMAXPROCS=%d: %v\n", p, elapsed)
    }
}
▶ 試してみよう

7. ゴルーチンリークとトラブルシューティング

(1) 漏洩のシナリオ

GO 
package main

import (
    "fmt"
    "runtime"
    "time"
)

// Leaky goroutine: reads from a channel that is never closed
func leakyGoroutine() {
    ch := make(chan int)
    go func() {
        <-ch  // Blocks forever—no one writes to ch
    }()
}

func main() {
    for i := 0; i < 10; i++ {
        leakyGoroutine()
    }

    time.Sleep(100 * time.Millisecond)
    fmt.Printf("Goroutine count after leak: %d\n", runtime.NumGoroutine())
    // Output: Goroutine count after leak: 11 (10 leaked + 1 main)
}

▶ サンプル：ゴルーチンを開始する4つの方法

GO 
package main

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

func say(msg string) {
    fmt.Println(msg)
}

func main() {
    var wg sync.WaitGroup

    // Style 1: Named function
    wg.Add(1)
    go func() {
        defer wg.Done()
        say("Style 1: Named function")
    }()

    // Style 2: Anonymous function
    wg.Add(1)
    go func() {
        defer wg.Done()
        fmt.Println("Style 2: Anonymous function")
    }()

    // Style 3: Anonymous function with parameter (recommended—avoids closure trap)
    wg.Add(1)
    go func(msg string) {
        defer wg.Done()
        fmt.Println(msg)
    }("Style 3: With parameter")

    // Style 4: Function as value
    wg.Add(1)
    fn := func() {
        defer wg.Done()
        fmt.Println("Style 4: Function variable")
    }
    go fn()

    wg.Wait()
}
▶ 試してみよう

▶ サンプル：ゴルーチンとOSスレッドのスタックの比較

GO 
package main

import (
    "fmt"
    "runtime"
    "sync"
)

func main() {
    var memStats runtime.MemStats
    var wg sync.WaitGroup
    num := 100000  // 100k goroutines

    runtime.GC()
    runtime.ReadMemStats(&memStats)
    before := memStats.HeapAlloc

    for i := 0; i < num; i++ {
        wg.Add(1)
        go func(n int) {
            defer wg.Done()
            _ = n
        }(i)
    }
    wg.Wait()

    runtime.ReadMemStats(&memStats)
    after := memStats.HeapAlloc
    perGoroutine := float64(after-before) / float64(num)

    fmt.Printf("Started %d goroutines\n", num)
    fmt.Printf("Memory increase: %.2f MB\n", float64(after-before)/1024/1024)
    fmt.Printf("Per goroutine: ~%.2f KB\n", perGoroutine/1024)
    fmt.Printf("Total goroutine count: %d\n", runtime.NumGoroutine())
}
▶ 試してみよう

出力：

TEXT 📖 参照専用 
Started 100000 goroutines
Memory increase: 21.45 MB
Per goroutine: ~0.22 KB
Total goroutine count: 1
💡 ヒント： 各ゴルーチンには初期段階でわずか 2 KB のスタック領域しか割り当てられませんが、OS スレッドにはデフォルトで 1～8 MB が割り当てられます。100,000 個のゴルーチンで約 21 MB、100,000 個のスレッドで約 80 GB となります。

▶ サンプル：ゴルーチンライフサイクルの監視

GO 
package main

import (
    "fmt"
    "runtime"
    "time"
)

func safeWorker(done chan struct{}) {
    <-done  // Wait for exit signal
}

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

    // Start 5 workers
    for i := 0; i < 5; i++ {
        go safeWorker(done)
    }

    fmt.Printf("Goroutine count after start: %d\n", runtime.NumGoroutine())

    // Send exit signal
    close(done)
    time.Sleep(10 * time.Millisecond)

    fmt.Printf("Goroutine count after close: %d\n", runtime.NumGoroutine())
}
▶ 試してみよう

(5) 漏洩防止チェックリスト

シナリオ 予防策
チャネルから読み込んでいるが、書き込みを行っているユーザーはいない バッファ付きチャネルを使用するか、select に default オプションを指定してください
チャンネルに書き込みをしても、誰もそれを読み取っていない コンシューマーが存在することを確認するか、デフォルト値を指定して select を使用してください
ゴルーチンによる無限ループ done チャネルまたはコンテキストを用いた終了制御
time.After はループ内でメモリリークを引き起こす time.NewTimer と Stop を使用して、各反復で新しいタイマーを作成する
select {} 空 出口経路があることを確認してください

8. 完全な例：10万件のログエントリの並行処理

GO 
// log_processor.go
package main

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

type LogLevel int

const (
    Info LogLevel = iota
    Warn
    Error
    Debug
)

type LogLine struct {
    Timestamp time.Time
    Level     LogLevel
    Message   string
    Source    string
}

type ProcessedLog struct {
    Original    LogLine
    Severity    string
    Alert       bool
    ProcessedAt time.Time
}

func (l LogLine) Process() ProcessedLog {
    // Simulate processing time (IO wait)
    time.Sleep(time.Duration(50+rand.Intn(50)) * time.Millisecond)

    severity := "low"
    alert := false
    switch l.Level {
    case Error:
        severity = "critical"
        alert = true
    case Warn:
        severity = "medium"
    }

    return ProcessedLog{
        Original:    l,
        Severity:    severity,
        Alert:       alert,
        ProcessedAt: time.Now(),
    }
}

// Serial processing
func processSerial(logs []LogLine) []ProcessedLog {
    results := make([]ProcessedLog, 0, len(logs))
    for _, log := range logs {
        results = append(results, log.Process())
    }
    return results
}

// Concurrent processing: worker pool pattern
func processConcurrent(logs []LogLine, workers int) []ProcessedLog {
    jobs := make(chan LogLine, len(logs))
    results := make(chan ProcessedLog, len(logs))

    // Fill jobs
    for _, log := range logs {
        jobs <- log
    }
    close(jobs)

    var wg sync.WaitGroup

    for w := 0; w < workers; w++ {
        wg.Add(1)
        go func(id int) {
            defer wg.Done()
            for log := range jobs {
                results <- log.Process()
            }
        }(w)
    }

    // Wait for all workers to finish, then close results
    go func() {
        wg.Wait()
        close(results)
    }()

    // Collect results
    processed := make([]ProcessedLog, 0, len(logs))
    for r := range results {
        processed = append(processed, r)
    }
    return processed
}

func main() {
    const logCount = 10000  // 10k for demo; can scale to 100k

    levels := []LogLevel{Info, Warn, Error, Debug}
    sources := []string{"api-gateway", "user-service", "payment", "database"}

    logs := make([]LogLine, logCount)
    for i := 0; i < logCount; i++ {
        logs[i] = LogLine{
            Timestamp: time.Now(),
            Level:     levels[rand.Intn(len(levels))],
            Message:   fmt.Sprintf("log-entry-%d", i),
            Source:    sources[rand.Intn(len(sources))],
        }
    }

    fmt.Printf("Log count: %d\n", logCount)
    fmt.Printf("CPU cores: %d, GOMAXPROCS: %d\n", runtime.NumCPU(), runtime.GOMAXPROCS(0))

    // Serial processing
    start := time.Now()
    serialResults := processSerial(logs)
    serialTime := time.Since(start)
    fmt.Printf("\nSerial: %v (%d entries/sec)\n", serialTime,
        int(float64(logCount)/serialTime.Seconds()))

    // Concurrent processing (10 workers)
    runtime.GC()
    start = time.Now()
    concurrentResults := processConcurrent(logs, 10)
    concurrentTime := time.Since(start)
    fmt.Printf("Concurrent(10): %v (%d entries/sec)\n", concurrentTime,
        int(float64(logCount)/concurrentTime.Seconds()))

    // Concurrent processing (50 workers)
    runtime.GC()
    start = time.Now()
    concurrentResults = processConcurrent(logs, 50)
    concurrentTime = time.Since(start)
    fmt.Printf("Concurrent(50): %v (%d entries/sec)\n", concurrentTime,
        int(float64(logCount)/concurrentTime.Seconds()))

    fmt.Printf("\nResult validation: serial=%d, concurrent=%d\n",
        len(serialResults), len(concurrentResults))
    fmt.Printf("Current goroutine count: %d\n", runtime.NumGoroutine())

    // Count alerts
    alertCount := 0
    for _, r := range concurrentResults {
        if r.Alert {
            alertCount++
        }
    }
    fmt.Printf("Alert count: %d\n", alertCount)
}

期待される出力：

TEXT 📖 参照専用 
Log count: 10000
CPU cores: 8, GOMAXPROCS: 8

Serial: 5.2s (1923 entries/sec)
Concurrent(10): 520ms (19230 entries/sec)
Concurrent(50): 110ms (90909 entries/sec)

Result validation: serial=10000, concurrent=10000
Current goroutine count: 1
Alert count: 2453
🔥 よくある間違い： ゴルーチン内でパニックが発生すると、プログラム全体がクラッシュし、外側の recover ではこれを捕捉できません。各ゴルーチンエントリポイントは defer recover() で保護する必要があります。チャネルに関する第14課では、このパターンについてさらに詳しく解説します。

❓ よくある質問

Q ゴルーチンとOSスレッドの違いは何ですか？
A ゴルーチンとは、Goランタイムによって管理されるユーザーモードの「コルーチン」です。そのスタックは初期状態でわずか2 KB（スレッドの場合は1～8 MB）であり、作成速度も切り替え速度も10倍速いです。Goでは数百万のゴルーチンが存在するのが一般的ですが、数万のスレッドを持つシステムでは、その負荷を処理するのに苦労することになります。
Q すべてのゴルーチンが終了するのを待つにはどうすればよいですか？
A sync.WaitGroup を使用します。—wg.Add(n) はカウントをインクリメントし、wg.Done() はデクリメントし、wg.Wait() はカウントがゼロになるまでブロックします。なお、Addは外側のスコープ内、またはゴルーチン起動前に呼び出す必要があり、Doneは確実に実行されるようにdeferで囲む必要があります。
Q ゴルーチンリークのトラブルシューティングはどのように行えばよいですか？
A runtime.NumGoroutine() を使用してカウントが継続的に増加しているかどうかを確認し、net/http/pprof を使用してゴルーチンスタックトレースを調査してください。リークの一般的な原因としては、チャネルの読み取りや書き込みがブロックされている場合、select句にdefault句が含まれていない場合、およびゴルーチンが終了するのを制御せずにゴルーチン起動を行うforループなどが挙げられます。
Q runtime.Gosched の目的は何ですか？
A P を積極的に譲り渡し、他のゴルーチンが実行される機会を与えるためです。手動で呼び出す必要はほとんどありません。Go は、I/O 待機、チャネル操作、および time.Sleep などの状況において、自動的にこれをスケジューリングします。
Q ゴルーチンの数の上限はどれくらいですか？
A 理論的にはメモリによって制限されます。各ゴルーチンのスタックは 2 KB から始まるため、4 GB のメモリで約 200 万個のゴルーチンが実行可能です。実用的な推奨値：CPU負荷の高いアプリケーションでは GOMAXPROCS 以下、I/O負荷の高いアプリケーションでは 1,000～10,000 以下。
Q GOMAXPROCS の値はどのように設定すべきですか？
A Go 1.5 以降では、デフォルトは NumCPU（CPU コア数）です。一般的に、変更の必要はありません。CPU負荷の高いワークロードではデフォルトのままにしておき、I/O負荷の高いワークロードでは適切に（2～10倍程度）増やすことができますが、根本的なボトルネックは並列プロセスの数ではなく、I/O速度にあります。
Q ゴルーチンでのパニックは他のゴルーチンに影響しますか？
A はい！どのゴルーチンであれ、回復されないパニックが発生すると、プロセス全体がクラッシュします。したがって、すべてのゴルーチンのエントリポイントには defer func() { if r := recover(); r != nil { ... } }() を記述する必要があります。
Q go キーワードで呼び出される関数は、引数を持たない必要がありますか？
A いいえ、引数を受け取ることができます：go myFunc(arg1, arg2)。クロージャ内で変数をキャプチャする際は、循環変数の罠に注意してください。直接キャプチャではなく、パラメータ渡し（コピーが生成される）を使用してください。

📖 まとめ

📝 練習問題

  1. 基本問題（難易度 ⭐）：5つのgoroutineを起動するプログラムを作成してください。各goroutineは、自身の番号と「Hello」を出力します。sync.WaitGroup を使用して、すべてが終了するのを待ち、goroutineの実行順序がランダムであることを確認してください。

  2. 上級問題（難易度 ⭐⭐）：並行処理による素数フィルタを実装してください。整数 n が与えられたとき、4 つの goroutine を使用して 1 から n までの範囲のどの数が素数であるかを並行してチェックし、素数のリストを返してください。同期には WaitGroup を使用し、並行処理と順次処理のアプローチのパフォーマンスの違いを比較してください。

  3. 課題（難易度 ⭐⭐⭐）：並行タスクスケジューラを構築する。タスクのバッチ（[]func() Result）が与えられた場合、ワーカープールモデルを用いてそれらを並行して実行する。このスケジューラは、以下の機能をサポートする必要があります：(1) ワーカー数の設定が可能であること；(2) フォールトトレランス（単一のgoroutineでパニックが発生しても、他のgoroutineに影響を与えないこと）；(3) 進行状況のコールバック（タスクの10%が完了するたびにメッセージを出力すること）；(4) runtime.NumGoroutineのリーク検出。

Web-Tutorial.com

Web-Tutorial 技術チーム

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

100%