1. 同期メカニズムの役割
並行プログラミングでは、複数のゴールーチンがリソースを共有する場合、競合状態を防ぐためにリソースが一度に1つのゴールーチンだけによってアクセスされることを保証する必要があります。これには同期メカニズムの使用が必要です。同期メカニズムは、異なるゴールーチンのアクセス順序を調整し、並行環境でのデータの整合性と状態の同期を確保します。
Go言語は、以下の同期メカニズムを含む豊富なセットを提供していますが、これに限定されません。
- Mutex（sync.Mutex）および読み取り-書き込みMutex（sync.RWMutex）
- チャネル（Channels）
- WaitGroup
- アトミック関数（atomicパッケージ）
- 条件変数（sync.Cond）
2. 同期プリミティブ
2.1 ミューテックス（sync.Mutex）
2.1.1 ミューテックスの概念と役割
ミューテックスは、共有リソースへのアクセスを1度に1つのゴールーチンだけが許可することで、共有リソースの安全な操作を保証する同期メカニズムです。ミューテックスは、LockおよびUnlockメソッドを使用して同期を実現します。Lockメソッドを呼び出すと、ロックが解放されるまでブロックされ、その間、他のゴールーチンがロックを取得しようとすると待機します。Unlockを呼び出すと、ロックが解放され、待機中の他のゴールーチンがそれを取得できるようになります。
var mu sync.Mutex
func criticalSection() {
// ロックを取得してリソースに排他的にアクセスする
mu.Lock()
// ここで共有リソースにアクセス
// ...
// 他のゴールーチンがロックを取得できるようにロックを解除する
mu.Unlock()
}
2.1.2 ミューテックスの実際の使用例
グローバルなカウンターを維持する必要があり、複数のゴールーチンがその値を増分させる場合、ミューテックスを使用することでカウンターの正確性を確保できます。
var (
mu sync.Mutex
counter int
)
func increment() {
mu.Lock() // カウンターを変更する前にロックする
counter++ // カウンターを安全に増分させる
mu.Unlock() // 操作後にロックを解除し、他のゴールーチンがカウンターにアクセスできるようにする
}
func main() {
for i := 0; i < 10; i++ {
go increment() // 複数のゴールーチンを開始してカウンターの値を増分させる
}
// しばらく待機する（実際にはWaitGroupなどを使用して全てのゴールーチンの完了を待機すべきです）
time.Sleep(1 * time.Second)
fmt.Println(counter) // カウンターの値を出力する
}
2.2 読み取り-書き込みミューテックス（sync.RWMutex）
2.2.1 読み取り-書き込みミューテックスの概念
RWMutexは、複数のゴールーチンが同時に共有リソースを読むことを可能にし、書き込み操作が排他的である特別なタイプのロックです。ミューテックスに比べて、読み取り-書き込みロックは複数の読み手のシナリオで性能を向上させることができます。これには、読み取り操作をロックするRLock、ロックを解除するRUnlock、書き込み操作をロックするLock、およびロックを解除するUnlockの4つのメソッドがあります。
2.2.2 読み取り-書き込みミューテックスの実際の使用例
データベースアプリケーションでは、読み取り操作の頻度が書き込み操作よりも遥かに高いことがあります。読み取り-書き込みロックを使用することで、複数のゴールーチンが同時に読むことができるため、システムのパフォーマンスが向上します。
var (
rwMu sync.RWMutex
data int
)
func readData() int {
rwMu.RLock() // 読み取りロックを取得し、他の読み取り操作が同時に実行できるようにする
defer rwMu.RUnlock() // deferを使用してロックが解除されるようにする
return data // データを安全に読む
}
func writeData(newValue int) {
rwMu.Lock() // 書き込みロックを取得し、この時点で他の読み書き操作をブロックする
data = newValue // 新しい値を安全に書き込む
rwMu.Unlock() // 書き込みが完了したらロックを解除する
}
func main() {
go writeData(42) // 書き込み操作を実行するゴールーチンを開始
fmt.Println(readData()) // メインのゴールーチンが読み取り操作を実行
// 全てのゴールーチンの完了を確認するためにWaitGroupなどの同期方法を使用する
}
上記の例では、複数の読み手はreadData関数を同時に実行できますが、writeDataを実行する書き手は新しい読み手や他の書き手をブロックします。このメカニズムは、書き込みよりも読み取りが多いシナリオでパフォーマンス上の利点を提供します。
2.3 条件変数（sync.Cond）
2.3.1 条件変数の概念
Go言語の同期メカニズムでは、条件変数は同期の原始的手段として、ある条件の変化を待機したり通知したりするために使用されます。条件変数は常に条件そのものの整合性を保護するために使用されるミューテックス（sync.Mutex）と一緒に使用されます。
条件変数の概念はオペレーティングシステムのドメインから来ており、ゴールーチンのグループが特定の条件が満たされるのを待つことができます。より具体的には、ゴールーチンは条件が満たされるのを待ちながら実行を一時停止し、他のゴールーチンが条件変数を使用して条件を変更し、他のゴールーチンに実行を再開させることができます。
Go標準ライブラリでは、条件変数はsync.Cond型を通じて提供され、その主なメソッドは以下の通りです：
-
Wait: このメソッドを呼び出すと、保持されていたロックが解放され、同じ条件変数でSignalまたはBroadcastを呼び出して目覚めさせるまでブロックされ、その後再びロックを取得しようとします。 -
Signal: この条件変数を待っているゴールーチンのうち1つを目覚めさせます。待機しているゴールーチンがいない場合、このメソッドを呼び出しても効果はありません。 -
Broadcast: この条件変数を待っているすべてのゴールーチンを目覚めさせます。
条件変数はコピーされるべきではないため、一般的には特定の構造体のポインタフィールドとして使用されます。
2.3.2 条件変数の実用例
以下は条件変数を使用した簡単なプロデューサー・コンシューマーモデルを示す例です：
package main
import (
"fmt"
"sync"
"time"
)
// SafeQueueはミューテックスによって保護された安全なキューです
type SafeQueue struct {
mu sync.Mutex
cond *sync.Cond
queue []interface{}
}
// Enqueueはキューの末尾に要素を追加し、待機しているゴールーチンに通知します
func (sq *SafeQueue) Enqueue(item interface{}) {
sq.mu.Lock()
defer sq.mu.Unlock()
sq.queue = append(sq.queue, item)
sq.cond.Signal() // キューが空でないことを待っているゴールーチンに通知
}
// Dequeueはキューの先頭から要素を取り出し、キューが空の場合は待機します
func (sq *SafeQueue) Dequeue() interface{} {
sq.mu.Lock()
defer sq.mu.Unlock()
// キューが空の場合は待機
for len(sq.queue) == 0 {
sq.cond.Wait() // 条件の変化を待ちます
}
item := sq.queue[0]
sq.queue = sq.queue[1:]
return item
}
func main() {
queue := make([]interface{}, 0)
sq := SafeQueue{
mu: sync.Mutex{},
cond: sync.NewCond(&sync.Mutex{}),
queue: queue,
}
// プロデューサーゴールーチン
go func() {
for i := 0; i < 5; i++ {
time.Sleep(1 * time.Second) // 製造時間をシミュレート
sq.Enqueue(fmt.Sprintf("item%d", i)) // 要素を生産
fmt.Println("Produce:", i)
}
}()
// コンシューマーゴールーチン
go func() {
for i := 0; i < 5; i++ {
item := sq.Dequeue() // 要素を消費し、キューが空の場合は待機
fmt.Printf("Consume: %v\n", item)
}
}()
// すべての生産と消費が完了することを保証するために十分な時間を待ちます
time.Sleep(10 * time.Second)
}
この例では、内部キューと条件変数を持つSafeQueue構造体を定義しています。コンシューマーがDequeueメソッドを呼び出し、キューが空の場合はWaitメソッドを使用して待機します。また、プロデューサーがEnqueueメソッドを呼び出して新しい要素をキューに追加する際には、待機しているコンシューマーを起こすためにSignalメソッドを使用します。
2.4 WaitGroup
2.4.1 WaitGroupの概念と使用法
sync.WaitGroupは、複数のゴールーチンの完了を待つための同期メカニズムです。ゴールーチンを開始すると、Addメソッドを呼
2.4.2 WaitGroupの実用例
以下はWaitGroupの使用例です：
package main
import (
"fmt"
"sync"
"time"
)
func worker(id int, wg *sync.WaitGroup) {
defer wg.Done() // 完了時にWaitGroupに通知
fmt.Printf("Worker %d 開始\n", id)
time.Sleep(time.Second) // 実行に時間がかかる操作をシミュレート
fmt.Printf("Worker %d 完了\n", id)
}
func main() {
var wg sync.WaitGroup
for i := 1; i <= 5; i++ {
wg.Add(1) // ゴルーチンを開始する前にカウンタを増やす
go worker(i, &wg)
}
wg.Wait() // すべてのワーカーゴルーチンが完了するのを待つ
fmt.Println("すべてのワーカーが完了")
}
この例では、worker関数がタスクの実行をシミュレートしています。main関数では、5つのworkerゴルーチンを開始します。各ゴルーチンを開始する前に、新しいタスクが実行されることをWaitGroupに通知するためにwg.Add(1)を呼び出します。各worker関数が完了したら、defer wg.Done()を呼び出してWaitGroupにタスクが完了したことを通知します。すべてのゴルーチンを開始した後、main関数はwg.Wait()でブロックし、すべてのワーカーが完了するまで待ちます。
2.5 アトミック操作 (sync/atomic)
2.5.1 アトミック操作の概念
アトミック操作は、並行プログラミングにおいて、他の操作によって実行中に割り込まれることなく、一連の操作を行うことを指します。複数のゴルーチンに対して、アトミック操作を使用することで、ロックを使用せずにデータの整合性と状態の同期を確保することができます。アトミック操作自体が実行のアトミシティを保証するため、その使用は重要です。
Go言語では、sync/atomicパッケージが低レベルのアトミックなメモリ操作を提供しています。int32、int64、uint32、uint64、uintptr、およびpointerなどの基本的なデータ型に対して、sync/atomicパッケージのメソッドを使用して安全な並行操作を行うことができます。アトミック操作の重要性は、他の並行プリミティブ（ロックや条件変数など）を構築する基盤となり、ロックメカニズムよりも効率的であることが一般的であることにあります。
2.5.2 アトミック操作の実用例
ウェブサイトへの同時訪問者数の追跡が必要なシナリオを考えてみましょう。通常、単純なカウンタ変数を使用すると、訪問者が到着したときにカウンタを増やし、訪問者が去ったときに減らすというアプローチが考えられますが、並行環境ではデータ競合が発生する可能性があります。そのため、sync/atomicパッケージを使用してカウンタを安全に操作することができます。
package main
import (
"fmt"
"sync"
"sync/atomic"
"time"
)
var visitorCount int32
func incrementVisitorCount() {
atomic.AddInt32(&visitorCount, 1)
}
func decrementVisitorCount() {
atomic.AddInt32(&visitorCount, -1)
}
func main() {
var wg sync.WaitGroup
for i := 0; i < 100; i++ {
wg.Add(1)
go func() {
incrementVisitorCount()
time.Sleep(time.Second) // 訪問者の滞在時間
decrementVisitorCount()
wg.Done()
}()
}
wg.Wait()
fmt.Printf("現在の訪問者数: %d\n", visitorCount)
}
この例では、100のゴルーチンを作成して訪問者の到着と退出をシミュレートしています。atomic.AddInt32()関数を使用することで、高度に並行した状況でもカウンタの増加と減少がアトミックに行われるため、visitorCountの正確性が保証されます。
2.6 チャネル同期メカニズム
2.6.1 チャネルの同期特性
チャネルは、Go言語においてゴルーチン間で通信する手段を提供します。チャネルはデータの送受信を可能にします。ゴルーチンがチャネルからデータを読み取ろうとするときにデータがない場合、データが利用可能になるまでブロックされます。同様に、チャネルが一杯になっている場合（非バッファリングチャネルの場合、すでにデータがある場合）、データを送信しようとするゴルーチンもブロックされます。この特性により、チャネルはゴルーチン間の同期に非常に有用です。
2.6.2 チャネルを使用した同期のユースケース
複数のゴルーチンによって完了する必要があるタスクがあり、それぞれがサブタスクを処理し、その結果を集約する必要があるとします。全てのゴルーチンの処理完了を待つためにチャネルを使用することができます。
package main
import (
"fmt"
"sync"
)
func worker(id int, wg *sync.WaitGroup, resultChan chan<- int) {
defer wg.Done()
// 何らかの操作を行う...
fmt.Printf("Worker %d 開始\n", id)
// サブタスクの結果をワーカーのIDと仮定する
resultChan <- id
fmt.Printf("Worker %d 完了\n", id)
}
func main() {
var wg sync.WaitGroup
numWorkers := 5
resultChan := make(chan int, numWorkers)
for i := 0; i < numWorkers; i++ {
wg.Add(1)
go worker(i, &wg, resultChan)
}
go func() {
wg.Wait()
close(resultChan)
}()
// すべての結果を収集
for result := range resultChan {
fmt.Printf("結果を受信：%d\n", result)
}
}
この例では、5つのゴルーチンを開始してタスクを実行し、その結果を resultChan チャネルを介して収集します。メインゴルーチンは全ての作業が別のゴルーチンで完了するのを待ち、その後結果チャネルを閉じます。その後、メインゴルーチンは resultChan チャネルをトラバースし、すべてのゴルーチンの結果を収集し、出力します。
2.7 一度だけ実行 (sync.Once)
sync.Once は、プログラムの実行中に特定の操作が一度だけ実行されることを保証する同期プリミティブです。sync.Once の典型的な使用例は、シングルトンオブジェクトの初期化や遅延初期化が必要なシナリオです。この操作を呼び出すゴルーチンの数に関係なく、その操作は一度しか実行されず、そのため Do 関数という名前が付けられています。
sync.Once は、並行性の問題と実行効率を完璧にバランスし、繰り返し初期化によるパフォーマンス問題についての懸念を取り除きます。
sync.Once の使用方法を示す簡単な例として次のようになります：
package main
import (
"fmt"
"sync"
)
var once sync.Once
var instance *Singleton
type Singleton struct{}
func Instance() *Singleton {
once.Do(func() {
fmt.Println("単一のインスタンスを作成中です。")
instance = &Singleton{}
})
return instance
}
func main() {
for i := 0; i < 10; i++ {
go Instance()
}
fmt.Scanln() // 出力を見るために待機
}
この例では、Instance 関数が同時に複数回呼び出されても、Singleton インスタンスの作成は一度だけ行われます。その後の呼び出しは最初に作成されたシングルトンインスタンスを直接返し、インスタンスの一意性を保証します。
2.8 ErrGroup
ErrGroup は、複数のゴルーチンを同期し、それらのエラーを収集するために使用されるGo言語のライブラリです。これは "golang.org/x/sync/errgroup" パッケージの一部であり、並行操作におけるエラーシナリオを扱う簡潔な方法を提供します。
2.8.1 ErrGroupの概念
ErrGroup の核となるアイデアは、関連するタスクのグループ（通常は並行して実行される）をバインドし、そのうちの1つのタスクが失敗した場合、グループ全体の実行をキャンセルすることです。同時に、これらの並行操作のいずれかがエラーを返した場合、ErrGroup はこのエラーをキャプチャして返します。
ErrGroup を使用するためには、まずパッケージをインポートします：
import "golang.org/x/sync/errgroup"
その後、ErrGroup のインスタンスを作成します：
var g errgroup.Group
その後、クロージャーの形式でタスクを ErrGroup に渡し、Go メソッドを呼び出して新しいゴルーチンを開始します：
g.Go(func() error {
// 特定のタスクを実行
// 全てが正常に進行した場合
return nil
// エラーが発生した場合
// return fmt.Errorf("エラーが発生しました")
})
最後に、Wait メソッドを呼び出します。これは全てのタスクが完了するのを待ち、その中でどれか1つのタスクがエラーを返した場合、Wait はそのエラーを返します：
if err := g.Wait(); err != nil {
// エラーを処理
log.Fatalf("タスクの実行中にエラーが発生しました：%v", err)
}
2.8.2 ErrGroupの実用例
3つの異なるデータソースからデータを並行して取得する必要があり、データソースのいずれかが失敗した場合は他のデータ取得操作を即座にキャンセルしたいというシナリオを考えます。このタスクは、ErrGroupを使用して簡単に実現できます。
package main
import (
"fmt"
"golang.org/x/sync/errgroup"
)
func fetchDataFromSource1() error {
// ソース1からデータを取得するシミュレーション
return nil // またはエラーを返して失敗をシミュレート
}
func fetchDataFromSource2() error {
// ソース2からデータを取得するシミュレーション
return nil // またはエラーを返して失敗をシミュレート
}
func fetchDataFromSource3() error {
// ソース3からデータを取得するシミュレーション
return nil // またはエラーを返して失敗をシミュレート
}
func main() {
var g errgroup.Group
g.Go(fetchDataFromSource1)
g.Go(fetchDataFromSource2)
g.Go(fetchDataFromSource3)
// すべてのゴルーチンが完了するのを待ち、エラーを収集する
if err := g.Wait(); err != nil {
fmt.Printf("データの取得中にエラーが発生しました: %v\n", err)
return
}
fmt.Println("すべてのデータを正常に取得しました！")
}
この例では、fetchDataFromSource1、fetchDataFromSource2、fetchDataFromSource3関数は異なるデータソースからデータを取得するシミュレーションをしています。これらの関数はg.Goメソッドに渡され、別々のゴルーチンで実行されます。関数のいずれかがエラーを返すと、g.Waitはそのエラーを即座に返し、エラーが発生した場合に適切なエラーハンドリングが可能になります。すべての関数が正常に実行された場合、g.Waitはすべてのタスクが正常に完了したことを示すnilを返します。
ErrGroupのもう一つの重要な機能は、ゴルーチンのいずれかがパニックを起こした場合、そのパニックを回復しエラーとして返そうと試みます。これにより、他の並行して実行されているゴルーチンが正常にシャットダウンできなくなるのを防ぎます。もちろん、タスクが外部のキャンセルシグナルに応答するようにしたい場合は、errgroupのWithContext関数とcontextパッケージを組み合わせてキャンセル可能なコンテキストを提供することができます。
このようにして、ErrGroupはGoの並行プログラミングにおいて非常に実用的な同期とエラーハンドリングのメカニズムとなります。