---
id: 01933e13f963942e4c9bc8c3c89fbc2b
author: snowmerak
title: Go 並行処理スターターパック
description: Go の並行性管理のための、ゴルーチン、チャネル、ミューテックスなど、さまざまなツールと技法を紹介します。
language: ja
date: 2024-10-18T05:51:30.010680581Z
path: /blog/posts/go-concurrency-starter-pack-z221a3399
category: concurrency
---

## 概要

### 簡単な紹介

Go言語には多くの並行性管理のためのツールがあります。このアーティクルでは、その中のいくつかとテクニックを紹介いたします。

### ゴルーチン?

goroutineはGo言語でサポートされる新しい形式の並行性モデルです。一般的に、プログラムは同時に複数の作業を行うために、OSからOSスレッドを受け取り、コア数だけ並列に作業を実行します。そして、より小さな単位の並行性を実行するためには、ユーザーランドでグリーンスレッドを生成し、1つのOSスレッド内で複数のグリーンスレッドが交互に作業を実行するようにします。しかし、goroutineの場合は、このような形態のグリーンスレッドをより小さく効率的に作りました。このようなgoroutineはスレッドより少ないメモリを使用し、スレッドより速く生成および切り替えることができます。

goroutineを使用するためには、単に`go`キーワードを使用するだけで済みます。これにより、プログラムを作成する過程で直感的に同期コードを非同期コードとして実行できます。

```go
package main

import (
    "fmt"
    "time"
)

func main() {
    ch := make(chan struct{})
    go func() {
        defer close(ch)
        time.Sleep(1 * time.Second)
        fmt.Println("Hello, World!")
    }()

    fmt.Println("Waiting for goroutine...")
    for range ch {}
}
```

このコードは、簡単に1秒間待ってから`Hello, World!`を出力する同期的なコードを、非同期の流れに変更します。現在の例は単純ですが、少し複雑なコードを同期コードから非同期コードへ変更すると、コードの可読性、視認性、理解度が従来のasync awaitやpromiseのような方式よりもさらに向上します。

ただし、多くの場合、このような同期コードを単純に非同期で呼び出す流れや、`fork & join`のような流れ（分割統治に似た流れ）を理解しないまま、不適切なgoroutineコードが作られることもあります。このような場合に備えるためのいくつかの方法と技法を、このアーティクルで紹介いたします。

## 並行性管理

### context

最初の管理技法として`context`が登場するのは意外に思われるかもしれません。しかしGo言語において`context`は単純なキャンセル機能を超えて、作業ツリー全体を管理するうえで優れた役割を果たします。もしご存じない方のために、簡単にこのパッケージを説明いたします。

```go
package main

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

    go func() {
        <-ctx.Done()
        fmt.Println("Context is done!")
    }()

    time.Sleep(1 * time.Second)

    cancel()

    time.Sleep(1 * time.Second)
}
```

上記のコードは`context`を使用して、1秒後に`Context is done!`を出力するコードです。`context`は`Done()`メソッドを通じてキャンセルの有無を確認でき、`WithCancel`、`WithTimeout`、`WithDeadline`、`WithValue`などのメソッドを通じて多様なキャンセル方法を提供します。

簡単な例を作ってみましょう。もしみなさんが、何らかのデータを取得するために`aggregator`パターンを使用して、`user`、`post`、`comment`を取得するコードを書くと仮定しましょう。そして、すべてのリクエストが2秒以内に行われなければならない場合、次のように記述できます。

```go
package main

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

    ch := make(chan struct{})
    go func() {
        defer close(ch)
        user := getUser(ctx)
        post := getPost(ctx)
        comment := getComment(ctx)

        fmt.Println(user, post, comment)
    }()

    select {
    case <-ctx.Done():
        fmt.Println("Timeout!")
    case <-ch:
        fmt.Println("All data is fetched!")
    }
}
```

上記のコードは、2秒以内にすべてのデータを取得できなければ`Timeout!`を出力し、すべてのデータを取得すると`All data is fetched!`を出力します。このように`context`を使用すると、複数のgoroutineが動作するコードでもキャンセルとタイムアウトを容易に管理できます。

context関連のさまざまな関数とメソッドは[godoc context](https://pkg.go.dev/context)で確認できます。簡単なものは学習して、気軽に利用できるようになっていただければと思います。

### channel

#### unbuffered channel

`channel`はゴルーチン間の通信のためのツールです。`channel`は`make(chan T)`で生成できます。この時、`T`は該当`channel`が伝達するデータの型です。`channel`は`<-`でデータを送受信でき、`close`で`channel`を閉じることができます。

```go
package main

func main() {
    ch := make(chan int)
    go func() {
        ch <- 1
        ch <- 2
        close(ch)
    }()

    for i := range ch {
        fmt.Println(i)
    }
}
```

上記のコードは`channel`を使用して1と2を出力するコードです。このコードでは、単純に`channel`に値を送って受け取ることのみを示しています。しかし`channel`はこれよりも多くの機能を提供します。まず`buffered channel`と`unbuffered channel`について見てみましょう。始める前に、上に書かれた例は`unbuffered channel`であり、チャネルにデータを送る動作とデータを受け取る動作が同時に行われなければなりません。もしこれらの動作が同時に行われなければ、デッドロックが発生する可能性があります。

#### buffered channel

もし上のコードが単純な出力ではなく、重い作業を行う2つのプロセスだとしたらどうでしょうか？2番目のプロセスが読み取って処理を行う途中で長時間かかると、1番目のプロセスもその間停止してしまうでしょう。私たちはこのような状況を防ぐために`buffered channel`を使用できます。

```go
package main

func main() {
    ch := make(chan int, 2)
    go func() {
        ch <- 1
        ch <- 2
        close(ch)
    }()

    for i := range ch {
        fmt.Println(i)
    }
}
```

上記のコードは`buffered channel`を使用して1と2を出力するコードです。このコードでは`buffered channel`を使用して、チャネルにデータを送る動作とデータを受け取る動作が同時に行われなくてもよいようにしました。このようにチャネルにバッファを設けると、その長さだけ余裕が生まれ、後順位の作業の影響によって生じる作業の遅延を防ぐことができます。

#### select

複数のチャネルを扱う際、`select`構文を使用すると`fan-in`構造を簡単に実装できます。

```go
package main

import (
    "fmt"
    "time"
)

func main() {
    ch1 := make(chan int, 10)
    ch2 := make(chan int, 10)
    ch3 := make(chan int, 10)

    go func() {
        for {
            ch1 <- 1
            time.Sleep(1 * time.Second)
        }
    }()
    go func() {
        for {
            ch2 <- 2
            time.Sleep(2 * time.Second)
        }
    }()
    go func() {
        for {
            ch3 <- 3
            time.Sleep(3 * time.Second)
        }
    }()

    for i := 0; i < 3; i++ {
        select {
        case v := <-ch1:
            fmt.Println(v)
        case v := <-ch2:
            fmt.Println(v)
        case v := <-ch3:
            fmt.Println(v)
        }
    }
}
```

上記のコードは、周期的に1、2、3を伝達する3つのチャネルを作り、`select`を使用してチャネルから値を受け取って出力するコードです。このように`select`を使用すると、複数のチャネルから同時にデータを受け取りながら、チャネルから値を受け取った順に処理できます。

#### for range

`channel`は`for range`を使用して簡単にデータを受け取ることができます。`for range`をチャネルに使用すると、該当チャネルにデータが追加されるたびに動作し、チャネルが閉じられるとループを終了します。

```go
package main

func main() {
    ch := make(chan int)
    go func() {
        ch <- 1
        ch <- 2
        close(ch)
    }()

    for i := range ch {
        fmt.Println(i)
    }
}
```

上記のコードは`channel`を使用して1と2を出力するコードです。このコードでは`for range`を使用して、チャネルにデータが追加されるたびにデータを受け取って出力します。そして、チャネルが閉じられるとループを終了します。

上で何度か書いたとおり、この構文は単純な同期手段としても使用できます。

```go
package main

func main() {
    ch := make(chan struct{})
    go func() {
        defer close(ch)
        time.Sleep(1 * time.Second)
        fmt.Println("Hello, World!")
    }()

    fmt.Println("Waiting for goroutine...")
    for range ch {}
}
```

上記のコードは、1秒間待ってから`Hello, World!`を出力するコードです。このコードでは`channel`を使用して同期的なコードを非同期的なコードに変更しました。このように`channel`を使用すると、同期的なコードを非同期的なコードへ容易に変更でき、`join`地点を設定することができます。

#### etc

1. nil channelにデータを送信または受信すると、無限ループに陥り、デッドロックが発生する可能性があります。
2. チャネルを閉じた後にデータを送信すると、panicが発生します。
3. チャネルをあえて閉じなくても、GCが回収する際にチャネルを閉じます。

### mutex

#### spinlock

`spinlock`は反復文を回りながら継続してロックを試みる同期方法です。Go言語ではポインタを使用して簡単にスピンロックを実装してみることができます。

```go
package spinlock

import (
    "runtime"
    "sync/atomic"
)

type SpinLock struct {
    lock uintptr
}

func (s *SpinLock) Lock() {
    for !atomic.CompareAndSwapUintptr(&s.lock, 0, 1) {
        runtime.Gosched()
    }
}

func (s *SpinLock) Unlock() {
    atomic.StoreUintptr(&s.lock, 0)
}

func NewSpinLock() *SpinLock {
    return &SpinLock{}
}
```

上記のコードは`spinlock`パッケージを実装したコードです。このコードでは`sync/atomic`パッケージを使用して`SpinLock`を実装しました。`Lock`メソッドでは`atomic.CompareAndSwapUintptr`を使用してロックを試み、`Unlock`メソッドでは`atomic.StoreUintptr`を使用してロックを解除します。この方式は休まずにロックを試みるため、ロックを取得するまで継続してCPUを使用することになり、無限ループに陥る可能性があります。したがって、`spinlock`は単純な同期に使用するか、短い時間の間だけ使用するのが良いでしょう。

#### sync.Mutex

`mutex`はゴルーチン間の同期のためのツールです。`sync`パッケージが提供する`mutex`は`Lock`、`Unlock`、`RLock`、`RUnlock`などのメソッドを提供します。`mutex`は`sync.Mutex`で生成でき、`sync.RWMutex`を用いて読み取り/書き込みロックを使用することもできます。

```go
package main

import (
    "sync"
)

func main() {
    var mu sync.Mutex
    var count int

    go func() {
        mu.Lock()
        count++
        mu.Unlock()
    }()

    mu.Lock()
    count++
    mu.Unlock()

    println(count)
}
```

上記のコードでは、ほぼ同時に2つのゴルーチンが同じ`count`変数にアクセスします。この時、`mutex`を使用して`count`変数にアクセスするコードをクリティカルセクションにすると、`count`変数への同時アクセスを防ぐことができます。そうすると、このコードは何度実行しても同じく`2`を出力します。

#### sync.RWMutex

`sync.RWMutex`は読み取りロックと書き込みロックを区別して使用できる`mutex`です。`RLock`、`RUnlock`メソッドを使用して読み取りロックを取得および解放できます。

```go
package cmap

import (
    "sync"
)

type ConcurrentMap[K comparable, V any] struct {
    sync.RWMutex
    data map[K]V
}

func (m *ConcurrentMap[K, V]) Get(key K) (V, bool) {
    m.RLock()
    defer m.RUnlock()

    value, ok := m.data[key]
    return value, ok
}

func (m *ConcurrentMap[K, V]) Set(key K, value V) {
    m.Lock()
    defer m.Unlock()

    m.data[key] = value
}
```

上記のコードは`sync.RWMutex`を使用して`ConcurrentMap`を実装したコードです。このコードでは、`Get`メソッドで読み取りロックを取得し、`Set`メソッドで書き込みロックを取得して、`data`マップに安全にアクセスして修正できます。読み取りロックが必要な理由は、単純な読み取り作業が多い場合に、書き込みロックをかけず読み取りロックのみをかけて、複数のゴルーチンが同時に読み取り作業を行えるようにするためです。これにより、状態の変更がないためにわざわざ書き込みロックをかける必要がない場合には、読み取りロックのみをかけて性能を向上させることができます。

#### fakelock

`fakelock`は`sync.Locker`を実装する簡単なトリックです。この構造体は`sync.Mutex`と同じメソッドを提供しますが、実際の動作はしません。

```go
package fakelock

type FakeLock struct{}

func (f *FakeLock) Lock() {}

func (f *FakeLock) Unlock() {}
```

上記のコードは`fakelock`パッケージを実装したコードです。このパッケージは`sync.Locker`を実装して`Lock`、`Unlock`メソッドを提供しますが、実際には何の動作もしません。なぜこのようなコードが必要なのかは、機会があれば説明いたします。

### waitgroup

#### sync.WaitGroup

`sync.WaitGroup`はゴルーチンの作業がすべて終わるまで待つツールです。`Add`、`Done`、`Wait`メソッドを提供し、`Add`メソッドでゴルーチンの数を追加し、`Done`メソッドでゴルーチンの作業が終わったことを知らせます。そして`Wait`メソッドで、すべてのゴルーチンの作業が終わるまで待機します。

```go
package main

import (
    "sync"
    "sync/atomic"
)

func main() {
    wg := sync.WaitGroup{}
    c := atomic.Int64{}

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

    wg.Wait()
    println(c.Load())
}
```

上記のコードは`sync.WaitGroup`を使用して、100個のゴルーチンが同時に`c`変数に値を加算するコードです。このコードでは`sync.WaitGroup`を使用して、すべてのゴルーチンが終わるまで待った後、`c`変数に加算した値を出力します。単純にいくつかの作業を`fork & join`する場合はチャネルのみを利用しても十分ですが、多量の作業を`fork & join`する場合は`sync.WaitGroup`を使用することも良い選択肢です。

#### with slice

スライスと一緒に使われる場合、`waitgroup`はロックなしで優れた並行実行作業を管理するツールになることができます。

```go
package main

import (
	"fmt"
	"sync"
    "rand"
)

func main() {
	var wg sync.WaitGroup
	arr := [10]int{}

	for i := 0; i < 10; i++ {
		wg.Add(1)
		go func(id int) {
			defer wg.Done()

			arr[id] = rand.Intn(100)
		}(i)
	}

	wg.Wait()
	fmt.Println("Done")

    for i, v := range arr {
        fmt.Printf("arr[%d] = %d\n", i, v)
    }
}
```

上記のコードは`waitgroup`のみを使用して、各ゴルーチンが同時に10個のランダムな整数を生成し、割り当てられたインデックスに保存するコードです。このコードでは`waitgroup`を使用して、すべてのゴルーチンが終わるまで待った後、`Done`を出力します。このように`waitgroup`を使用すると、複数のゴルーチンが同時に作業を行い、すべてのゴルーチンが終わるまでロックなしでデータを保存し、作業終了後に一括して後処理を行うことができます。

#### golang.org/x/sync/errgroup.ErrGroup

`errgroup`は`sync.WaitGroup`を拡張したパッケージです。`errgroup`は`sync.WaitGroup`と異なり、ゴルーチンの作業のうち一つでもエラーが発生すると、すべてのゴルーチンをキャンセルしてエラーを返します。

```go
package main

import (
    "context"
    "fmt"
    "golang.org/x/sync/errgroup"
)

func main() {
    g, ctx := errgroup.WithContext(context.Background())
    _ = ctx

    for i := 0; i < 10; i++ {
        i := i
        g.Go(func() error {
            if i == 5 {
                return fmt.Errorf("error")
            }
            return nil
        })
    }

    if err := g.Wait(); err != nil {
        fmt.Println(err)
    }
}
```

上記のコードは`errgroup`を使用して10個のゴルーチンを生成し、5番目のゴルーチンでエラーを発生させるコードです。意図的に5番目のゴルーチンでエラーを発生させ、エラーが発生する場合を示しました。ただし実際に使用する際には、`errgroup`を使用してゴルーチンを生成し、各ゴルーチンでエラーが発生した場合にさまざまな後処理を行う方式で使用すれば良いです。

### once

一度だけ実行されるべきコードを実行するツールです。以下のコンストラクタを通じて関連コードを実行できます。

```go
func OnceFunc(f func()) func()
func OnceValue[T any](f func() T) func() T
func OnceValues[T1, T2 any](f func() (T1, T2)) func() (T1, T2)
```

#### OnceFunc

`OnceFunc`は単純に該当関数が全体にわたってただ一度だけ実行されるようにします。

```go
package main

import "sync"

func main() {
    once := sync.OnceFunc(func() {
        println("Hello, World!")
    })

    once()
    once()
    once()
    once()
    once()
}
```

上記のコードは`sync.OnceFunc`を使用して`Hello, World!`を出力するコードです。このコードでは`sync.OnceFunc`を使用して`once`関数を生成し、`once`関数を何度呼び出しても`Hello, World!`は一度だけ出力されます。

#### OnceValue

`OnceValue`は単純に該当関数が全体にわたってただ一度だけ実行されるのではなく、該当関数の戻り値を保存して、再び呼び出される際に保存された値を返します。

```go
package main

import "sync"

func main() {
    c := 0
    once := sync.OnceValue(func() int {
        c += 1
        return c
    })

    println(once())
    println(once())
    println(once())
    println(once())
    println(once())
}
```

上記のコードは`sync.OnceValue`を使用して`c`変数を1ずつ増加させるコードです。このコードでは`sync.OnceValue`を使用して`once`関数を生成し、`once`関数を何度呼び出しても`c`変数は一度だけ増加した1を返します。

#### OnceValues

`OnceValues`は`OnceValue`と同様に動作しますが、複数の値を返すことができます。

```go
package main

import "sync"

func main() {
    c := 0
    once := sync.OnceValues(func() (int, int) {
        c += 1
        return c, c
    })

    a, b := once()
    println(a, b)
    a, b = once()
    println(a, b)
    a, b = once()
    println(a, b)
    a, b = once()
    println(a, b)
    a, b = once()
    println(a, b)
}
```

上記のコードは`sync.OnceValues`を使用して`c`変数を1ずつ増加させるコードです。このコードでは`sync.OnceValues`を使用して`once`関数を生成し、`once`関数を何度呼び出しても`c`変数は一度だけ増加した1を返します。

### atomic

`atomic`パッケージは原子的演算を提供するパッケージです。`atomic`パッケージは`Add`、`CompareAndSwap`、`Load`、`Store`、`Swap`などのメソッドを提供しますが、最近では`Int64`、`Uint64`、`Pointer`などの型の使用が推奨されています。

```go
package main

import (
    "sync"
    "sync/atomic"
)

func main() {
    wg := sync.WaitGroup{}
    c := atomic.Int64{}

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

    wg.Wait()
    println(c.Load())
}
```

先ほど使用した例です。`atomic.Int64`型を使用して`c`変数を原子的に増加させるコードです。`Add`メソッドと`Load`メソッドで変数を原子的に増加させ、変数を読み取ることができます。また、`Store`メソッドで値を保存し、`Swap`メソッドで値を交換し、`CompareAndSwap`メソッドで値を比較して適合すれば交換することができます。

### cond

#### sync.Cond

`cond`パッケージは条件変数を提供するパッケージです。`cond`パッケージは`sync.Cond`で生成でき、`Wait`、`Signal`、`Broadcast`メソッドを提供します。

```go
package main

import (
    "sync"
)

func main() {
    c := sync.NewCond(&sync.Mutex{})
    ready := false

    go func() {
        c.L.Lock()
        ready = true
        c.Signal()
        c.L.Unlock()
    }()

    c.L.Lock()
    for !ready {
        c.Wait()
    }
    c.L.Unlock()

    println("Ready!")
}
```

上記のコードは`sync.Cond`を使用して、`ready`変数が`true`になるまで待つコードです。このコードでは`sync.Cond`を使用して`ready`変数が`true`になるまで待った後、`Ready!`を出力します。このように`sync.Cond`を使用すると、複数のゴルーチンが同時に特定の条件を満たすまで待機させることができます。

これを活用して、簡単な`queue`を実装できます。

```go
package queue

import (
    "sync"
    "sync/atomic"
)

type Node[T any] struct {
    Value T
    Next  *Node[T]
}

type Queue[T any] struct {
    sync.Mutex
    Cond *sync.Cond
    Head *Node[T]
    Tail *Node[T]
    Len  int
}

func New[T any]() *Queue[T] {
    q := &Queue[T]{}
    q.Cond = sync.NewCond(&q.Mutex)
    return q
}

func (q *Queue[T]) Push(value T) {
    q.Lock()
    defer q.Unlock()

    node := &Node[T]{Value: value}
    if q.Len == 0 {
        q.Head = node
        q.Tail = node
    } else {
        q.Tail.Next = node
        q.Tail = node
    }
    q.Len++
    q.Cond.Signal()
}

func (q *Queue[T]) Pop() T {
    q.Lock()
    defer q.Unlock()

    for q.Len == 0 {
        q.Cond.Wait()
    }

    node := q.Head
    q.Head = q.Head.Next
    q.Len--
    return node.Value
}
```

このように`sync.Cond`を活用すると、`spin-lock`で多くのCPU使用量を消費する代わりに、効率的に待機し、条件が満たされると再び動作できます。

### semaphore

#### golang.org/x/sync/semaphore.Semaphore

`semaphore`パッケージはセマフォを提供するパッケージです。`semaphore`パッケージは`golang.org/x/sync/semaphore.Semaphore`で生成でき、`Acquire`、`Release`、`TryAcquire`メソッドを提供します。

```go
package main

import (
    "context"
    "fmt"
    "golang.org/x/sync/semaphore"
)

func main() {
    s := semaphore.NewWeighted(1)

    if s.TryAcquire(1) {
        fmt.Println("Acquired!")
    } else {
        fmt.Println("Not Acquired!")
    }

    s.Release(1)
}
```

上記のコードは`semaphore`を使用してセマフォを生成し、`Acquire`メソッドでセマフォを取得して、`Release`メソッドでセマフォを解放するコードです。このコードでは`semaphore`を使用してセマフォを取得して解放する方法を示しました。

## おわりに

基本的な内容はここまであれば十分だと思われます。このアーティクルの内容をもとに、みなさんがゴルーチンを使用して並行性を管理する方法を理解し、実際に使用できるようになっていただければ幸いです。このアーティクルがみなさんのお役に立てば嬉しいです。ありがとうございました。