Goの並行処理の奥義

Posted at 2025-03-06

Leapcell: The Next-Gen Serverless Platform for Golang app Hosting

Go言語におけるgoroutineの並列制御方法の分析

Go言語のプログラミングシステムにおいて、goroutineは軽量スレッドとして、低いリソース消費と低い切り替えコストという大きな利点により、並列操作の効率的な実装に強力なサポートを提供しています。しかし、これらの並列実行するgoroutineを効果的に制御する方法は、開発者が直面する重要な問題となっています。

並列制御に関して言えば、ロックメカニズムはしばしば最初に考えられる手段です。Go言語では、相互排他ロックのsync.Mutexや読み書きロックのsync.RWMutexなどの関連するロックメカニズムが提供されており、さらにアトミック操作のsync/atomicも備えています。ただし、これらの操作は主に並列時のデータセキュリティに焦点を当てており、直接goroutine自体を対象としたものではありません。

この記事では、goroutineの並列動作を制御する一般的な方法を焦点に紹介します。Go言語には、最も一般的に使用される3つの方法があり、それはsync.WaitGroup、channel、そしてContextです。

I. sync.WaitGroup

sync.WaitGroupはGo言語における非常に実用的な同期プリミティブであり、その主な機能は開発者がグループのgoroutineの実行完了を待つのを支援することです。通常、sync.WaitGroupは以下のようなシナリオで使用されます：

メイン関数でプログラムを終了する前に、グループのgoroutineがすべて実行されたことを確認する必要がある場合。
関数内で複数のgoroutineを起動し、これらのgoroutineがすべて完了するまで結果を返さないシナリオ。
関数が複数のgoroutineを起動し、それらがすべて完了した後に特定の操作を実行する必要がある場合。
関数が複数のgoroutineを起動し、それらがすべて完了した後に特定のリソースをクローズする必要がある状況。
関数が複数のgoroutineを起動し、それらがすべて完了するまでループを抜けられない場合。

sync.WaitGroupを使用する際の具体的な手順は以下の通りです：まず、sync.WaitGroupオブジェクトを作成します；次に、このオブジェクトのAddメソッドを使用して待機するgoroutineの数を指定します；その後、goキーワードを使用して複数のgoroutineを起動し、各goroutine内でsync.WaitGroupオブジェクトのDoneメソッドを呼び出して、goroutineの実行が完了したことを示します；最後に、sync.WaitGroupオブジェクトのWaitメソッドを呼び出して、すべてのgoroutineが完了するのを待ちます。

以下は簡単な例です。この例では3つのgoroutineを起動し、それぞれ0秒、1秒、2秒間スリープし、これら3つのgoroutineが終了した後にメイン関数が終了します：

package main

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

func main() {
    var wg sync.WaitGroup

    for i := 0; i < 3; i++ {
        wg.Add(1)
        go func(i int) {
            defer wg.Done()
            fmt.Printf("sub goroutine sleep: %ds\n", i)
            time.Sleep(time.Duration(i) * time.Second)
        }(i)
    }

    wg.Wait()
    fmt.Println("main func done")
}

II. channel

Go言語において、channelは開発者がgoroutineの並列性をより良く制御することができる強力なツールです。以下は、channelを使用してgoroutineの並列性を制御するいくつかの一般的な方法です：

(I) 非バッファ付きチャネルを使用した同期

非バッファ付きのchannelを使用して、プロデューサー-コンシューマーパターンを実装することができます。このパターンでは、1つのgoroutineがデータを生成する役割を担い、もう1つのgoroutineがデータを消費する役割を担います。プロデューサーgoroutineがchannelにデータを送信すると、コンシューマーgoroutineはブロック状態に入り、データの到着を待ちます。こうすることで、プロデューサーとコンシューマー間のデータ同期を保証することができます。

以下は簡単な例コードです：

package main

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

func producer(ch chan int, wg *sync.WaitGroup) {
    defer wg.Done()
    for i := 0; i < 10; i++ {
        ch <- i
        fmt.Println("produced", i)
        time.Sleep(100 * time.Millisecond)
    }
    close(ch)
}

func consumer(ch chan int, wg *sync.WaitGroup) {
    defer wg.Done()
    for i := range ch {
        fmt.Println("consumed", i)
        time.Sleep(150 * time.Millisecond)
    }
}

func main() {
    var wg sync.WaitGroup
    ch := make(chan int)

    wg.Add(2)
    go producer(ch, &wg)
    go consumer(ch, &wg)

    wg.Wait()
}

この例では、非バッファ付きのchannelを作成して、プロデューサーgoroutineとコンシューマーgoroutine間でデータを転送しています。プロデューサーgoroutineはchannelにデータを送信し、コンシューマーgoroutineはchannelからデータを受信します。プロデューサーgoroutineでは、time.Sleep関数を使用してデータを生成するのに必要な時間をシミュレートしています；コンシューマーgoroutineでも、time.Sleep関数を使用してデータを消費するのに必要な時間をシミュレートしています。最後に、sync.WaitGroupを使用してすべてのgoroutineが完了するのを待っています。

(II) バッファ付きチャネルを使用したレート制限

バッファ付きのchannelを使用して、並列実行するgoroutineの数を制限することができます。具体的なアプローチは、channelの容量を望ましい最大並列数に設定することです。各goroutineを起動する前に、channelに値を送信します；goroutineの実行が完了したときに、channelから値を受信します。こうすることで、同時に実行されるgoroutineの数が指定された最大並列数を超えないことを保証することができます。

以下は簡単な例コードです：

package main

import (
    "fmt"
    "sync"
)

func main() {
    var wg sync.WaitGroup
    maxConcurrency := 3
    semaphore := make(chan struct{}, maxConcurrency)

    for i := 0; i < 10; i++ {
        wg.Add(1)
        go func() {
            defer wg.Done()
            semaphore <- struct{}{}
            fmt.Println("goroutine", i, "started")
            // do some work
            fmt.Println("goroutine", i, "finished")
            <-semaphore
        }()
    }

    wg.Wait()
}

この例では、バッファサイズが3のバッファ付きのchannelを作成しています。そして10個のgoroutineを起動しています。各goroutineでは、空の構造体をchannelに送信してgoroutineの実行が開始されたことを示します；goroutineが完了したときに、channelから空の構造体を受信してgoroutineの実行が完了したことを示します。こうすることで、同時に実行されるgoroutineの数が3を超えないことを保証することができます。

III. Context

Go言語において、Contextはgoroutineの並列性を制御する重要な手段です。以下は、Contextを使用してgoroutineの並列性を制御するいくつかの一般的な方法です：

(I) タイムアウト制御

ある場合、プログラムが長期間ブロックされたりデッドロックになったりする問題を避けるために、goroutineの実行時間を制限する必要があります。Contextは開発者がgoroutineの実行時間をより良く制御することを助けることができます。具体的な操作は、タイムアウト期間を持つContextを作成し、それをgoroutineに渡すことです。goroutineがタイムアウト期間内に実行を完了できない場合、ContextのDoneメソッドを使用してgoroutineの実行をキャンセルすることができます。

以下は簡単な例コードです：

package main

import (
    "context"
    "fmt"
    "time"
)

func main() {
    ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
    defer cancel()

    go func() {
        for {
            select {
            case <-ctx.Done():
                fmt.Println("goroutine finished")
                return
            default:
                fmt.Println("goroutine running")
                time.Sleep(500 * time.Millisecond)
            }
        }
    }()

    time.Sleep(3 * time.Second)
}

この例では、タイムアウト期間を持つContextを作成し、それをgoroutineに渡しています。goroutine内では、select文を使用してContextのDoneメソッドを監視しています。Contextがタイムアウトすると、goroutineの実行がキャンセルされます。

(II) キャンセル操作

プログラムの実行中には、時には特定のgoroutineの実行をキャンセルする必要があります。Contextは開発者がgoroutineのキャンセル操作をより良く制御することを助けることができます。具体的なアプローチは、キャンセル関数を持つContextを作成し、それをgoroutineに渡すことです。goroutineの実行をキャンセルする必要がある場合、ContextのCancelメソッドを呼び出すことができます。

以下は簡単な例コードです：

package main

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

func main() {
    ctx, cancel := context.WithCancel(context.Background())

    var wg sync.WaitGroup
    wg.Add(1)
    go func() {
        defer wg.Done()
        for {
            select {
            case <-ctx.Done():
                fmt.Println("goroutine finished")
                return
            default:
                fmt.Println("goroutine running")
                time.Sleep(500 * time.Millisecond)
            }
        }
    }()

    time.Sleep(2 * time.Second)
    cancel()
    wg.Wait()
}

この例では、キャンセル関数を持つContextを作成し、それをgoroutineに渡しています。goroutine内では、select文を使用してContextのDoneメソッドを監視しています。Contextがキャンセルされると、goroutineの実行がキャンセルされます。メイン関数では、time.Sleep関数を使用してプログラムの実行中のある時点でgoroutineの実行をキャンセルする必要がある状況をシミュレートし、その後ContextのCancelメソッドを呼び出しています。

(III) リソース管理

あるシナリオでは、goroutineが使用するリソースを管理する必要があり、リソースリークや競合状態などの問題を防ぐ必要があります。Contextは開発者がgoroutineが使用するリソースをより良く管理することを助けることができます。具体的な操作は、リソースをContextに関連付け、Contextをgoroutineに渡すことです。goroutineの実行が完了したときに、Contextを使用してリソースを解放したり、他のリソース管理操作を実行したりすることができます。

以下は簡単な例コードです：

package main

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

func worker(ctx context.Context, wg *sync.WaitGroup) {
    defer wg.Done()
    for {
        select {
        case <-ctx.Done():
            fmt.Println("goroutine finished")
            return
        default:
            fmt.Println("goroutine running")
            time.Sleep(500 * time.Millisecond)
        }
    }
}

func main() {
    ctx, cancel := context.WithCancel(context.Background())

    var wg sync.WaitGroup
    wg.Add(1)
    go worker(ctx, &wg)

    time.Sleep(2 * time.Second)
    cancel()
    wg.Wait()
}