Golang 並行処理ミニブック

はじめに

このミニブックでは、Go の並行処理に関するトピックを簡潔に解説します。

• インタラクティブな例: コードを変更して「Run」ボタンをクリックすることで実験できます。
• PDF 版: こちらも提供しています。
• 対象読者: 並行処理の速習リファレンスであり、初心者向けのガイドではありません。
  • ゼロから学べる実践的な演習については、『Gist of Go: Concurrency』をご覧ください。
• AI の使用: 本書には AI は使用されていません。

トピック一覧

以下が本書で扱われる主要なトピックです：

• Goroutines
• Channels
• Select
• Pipelines
• Time
• Context
• Wait groups
• Data races
• Race conditions
• Mutexes
• Semaphores
• Signaling
• Run once
• Object pool
• Atomics
• Testing
• Scheduling
• Diagnostics
• Final thoughts

Goroutines

Go の並行処理の基礎となるのは、

go

基本

func main() {
    var wg sync.WaitGroup
    wg.Add(2)
    go func() {
        defer wg.Done()
        fmt.Println("worker 1")
    }()
    go func() {
        defer wg.Done()
        fmt.Println("worker 2")
    }()
    wg.Wait()
}

• Go ランタイムはこれらの goroutine を管理し、OS スレッドに分散させます。
• OS スレッドと比較して軽量なため、数百〜数千個作成できます。
• goroutine は完全に独立しています（

  main

  関数も goroutine です）。
• 注意点:

  main

  関数が終了すると、他の goroutine もすべてシャットダウンされます。

Wait Group（

sync.WaitGroup

• 内部的にカウンターを持ちます。

• Add(n)

  で n 増加させます。

• Done()

  で一つ減少させます。

• Wait()

  は呼び出し元がカウンターがゼロになるまでブロックします。

WaitGroup.Go

sync.WaitGroup

Go

1. カウンターを自動インクリメントする。
2. 関数を goroutine で実行する。
3. 完了時にカウンターを減らす。

func main() {
    var wg sync.WaitGroup
    wg.Go(func() {
        fmt.Println("worker 1")
    })
    wg.Go(func() {
        fmt.Println("worker 2")
    })
    wg.Wait()
}

Channels

goroutine はチャネル（channel）を通じて互いに値をやり取りできます。

基本

チャネルは、一方が送信し他方が受信できる窓のようなものです。

func main() {
    messages := make(chan string)

    go func() { messages <- "ping" }()

    msg := <-messages
    fmt.Println(msg)
}

• 値の送信（

  ch <- val

  ）は同期操作です。
• 送信側は受信側が値を読み取るまでブロックします。

Output Channel（出力チャンネル）

関数から出力チャネルを返し、内部の goroutine で埋めるのは一般的なパターンです。

func generate(start, stop int) chan int {
    out := make(chan int)
    go func() {
        for i := start; i < stop; i++ {
            out <- i
        }
    }()
    return out
}

チャンネルのクローズ

データ送信終了を知らせるため、書き込み側は

close()

func generate(start, stop int) chan int {
    out := make(chan int)
    go func() {
        defer close(out) // クロージング処理
        for i := start; i < stop; i++ {
            out <- i
        }
    }()
    return out
}

• 読取側は

  num, ok := <-in

  のように、2 つ目の値（「コンマ OK」）で状態をチェックします。
• チャネルがオープン中：次の値と

  true

  を返す。
• チャネルがクローズ：ゼロ値と

  false

  を返す。

重要:

• チャネルは一度しかクローズできません。二度目以降のパニック（panic）が発生します。
• クローズする唯一の理由は、すべてのデータを送信したことを知らせることです。
• 使わなくなったチャネルはガベージコレクタが管理するため、クローズ済みかどうかが重要ではありません。

Range によるイテレーション

range

func main() {
    nums := generate(5, 10)
    for n := range nums { // クローズ時ループ終了
        fmt.Print(n, " ")
    }
}

• スライス上の

  range

  と異なり、チャネル上の

  range

  は単一の値を返します。

方向付きチャンネル

誤った操作を防ぐため、チャネルの方向を設定できます。

chan T

chan<- T

<-chan T

• 送信用チャンネルからは読み取れない。
• 受信用チャンネルには書き込めない（クローズも不可）。

stream := make(chan int)

go func(in chan<- int) { // 入力は出力用（送信専用）
    in <- 42
}(stream)

func(out <-chan int) {   // 入力は入力用（受信専用）
    fmt.Println(<-out)
}(stream)

バッファ付きチャンネル

固定サイズのバッファを持つ FIFO キューのように動作します。

• バッファに空きがあれば書き込みはブロックしません。
• バッファに値があれば読み取りもブロックしません。

// サイズ指定なし（非バッファ）
stream := make(chan int)

// サイズ 3 のバッファ付きチャネル
stream := make(chan int, 3)

len()

cap()

stream := make(chan int, 1)
stream <- 11
close(stream)

val, ok := <-stream // val=11, ok=true
val, ok = <-stream  // val=0 (ゼロ値), ok=false

Nil チャンネル

• Go の型と同様にゼロ値（Nil）を持ちます。
• Nil チャネルへの書き込み・読み取り: goroutine が無限にブロックされます。
• クローズされた Nil チャネル: パニックを発生させます。

Select

select

switch

• ブロックされていないケースがある場合、それらをチェックします。
• 複数のケースが準備状態の場合、ランダムに一つを選んで実行します。
• すべてのケースがブロックされ且つ

  default

  ケースがある場合、それを実行します。
• すべてのケースがブロックされ且つ

  default

  ケースがない場合、いずれかが準備されるまで待ちます。

データのマージ

2 つの入力チャネルからの値を出力チャネルに送る関数です。

func merge(in1, in2 <-chan int) <-chan int {
    out := make(chan int)
    go func() {
        defer close(out)
        for in1 != nil || in2 != nil {
            select {
            case val1, ok := <-in1:
                if ok { out <- val1 } else { in1 = nil }
            case val2, ok := <-in2:
                if ok { out <- val2 } else { in2 = nil }
            }
        }
    }()
    return out
}

Goroutine のキャンセル

入力チャネルが枯渇するか

cancel

func process(cancel chan struct{}, in <-chan int) <-chan int {
    out := make(chan int)
    go func() {
        for val := range in {
            select {
            case out <- val*10:
            case <-cancel: // キャンセルシグナル待ち
                fmt.Println("canceled")
                return
            }
        }
    }()
    return out
}

非同期操作

チャネルが忙しければエラーを返す機能です。

func multiplier(ch chan<- int) func(n int) error {
    return func(n int) error {
        select {
        case ch <- n*10:
            return nil
        default:
            return errors.New("busy") // 非同期（ブロックしない）
        }
    }
}

Pipelines

パイプラインは、各ステップが入力データを処理し出力する一連の操作です。入力は出力であり、すべてがチャネルです。

典型的な構造

1. Reader: ファイル、データベース、ネットワークから入力データを読み取る。
2. N Processors: データを変換、フィルタ、集約、あるいは増幅する。
3. Writer: 処理されたデータを出力先へ書き出す。

コミュニケーションパターン

• Output Channel: goroutine の完了を知らせるために使用します。
• Done Channel: 結果を返さない場合、完了シグナル用に使用します。
• Cancel Channel: 呼び出し側が早期終了を要求するために使用します。

エラーハンドリングの 3 つのアプローチ

1. 最初のエラーで返却する: エラー発生時即座に返す。
2. 結果タイプを使う:

   Result

   タイプにアンサーかエラーを格納する。
3. エラーを別途収集する: 成功は出力、失敗は別のチャネルへエラーを送る。

Time

time

time.After()

指定されたタイムアウト期間後に値を受け取る初期に空のチャネルを返します。

func withTimeout(timeout time.Duration, fn func()) error {
    done := make(chan struct{})
    go func() {
        defer close(done)
        fn()
    }()

    select {
    case <-done:
        return nil
    case <-time.After(timeout):
        return errors.New("timeout")
    }
}

Timer

time.Timer

• Stop()

  で停止すると、期限切れになっていない場合は

  true

  を返します。

• time.AfterFunc()

  ワラッパーを使うと便利です（指定時間後に関数を実行・キャンセル可能）。

Ticker

タイマーのように機能しますが、停止するまで継続的に発火します。

ticker := time.NewTicker(duration)
defer ticker.Stop() // リソース解放必須
for tick := range ticker.C {
    // 処理
}

• 読取側が追いつけない場合、チック（ティック）はスキップされます。

Context

コンテキストの主な目的は、手動またはタイムアウト/デッドラインによって操作をキャンセルすることです。関数はコンテキストを受け取り、その

Done()

キャンセルの種類

• context.Canceled

  :

  context.WithCancel

  で作成したコンテキストによる手動キャンセル。

• context.DeadlineExceeded

  :

  context.WithTimeout

  や

  WithDeadline

  による時間超過。

特徴と制限

• 層化（親子関係）: 新しいプロパティは古い（親）コンテキストを基に作成します。短いタイムアウトが優先されます。子は親より短くできても延長できません。
• 不変性: コンテキストオブジェクト自体は不変です。
• 複数回のキャンセル: 安全です。最初のキャンセルのみが機能し、残りは無視されます。
• カスタム原因:

  WithCancelCause

  などを使用できます（

  context.Cause

  でアクセス可能）。
• クリーンアップ:

  context.AfterFunc

  を使用してキャンセル時に実行する関数を登録します。
• 値の保持:

  context.WithValue

  で値を持つことは可能ですが、一般に推奨されません（明示的なパラメータを使う方がよい）。

Wait Groups

sync.WaitGroup

• Add(n)

  : カウンターを増加させる。

• Done()

  : カウンターを減少させる（デフォルトは 1）。

• Wait()

  : カウンターがゼロになるまで呼び出し元をブロックする。

sync.WaitGroup.Go

Add

Done

安全な使いどころ

• すべてのメソッドは複数の goroutine からも安全に使用できます。
• 通常、すべての

  Add

  が

  Wait

  の前に行われますが、別の goroutine から後に追加しても問題ありません。

• Wait

  は複数の goroutine から呼び出すこともでき、すべてはグループのカウンターがゼロになるまでブロックされます。

Data Races (データ競合)

データ競合とは、複数の goroutine が共有データにアクセスし、そのうち少なくとも一方が書き込みを行う場合です。

影響と対策

• データ競合は常にランタイムパニックを引き起こすわけではありません。
• Go は特殊なツールであるレース検出器 (race detector) を提供しています。

  • -race

    フラグをオンにして

    go test

    ,

    go run

    ,

    go build

    コマンドで有効にできます。

防止方法

1. 並行データ変更の回避: 通常はチャネルを使用する。
2. Mutex で同期化: アクセスを保護する。
3. 原子操作のみを使用: 原子型の変数を利用する。

Race Conditions (競合条件)

競合条件とは、複数の goroutine の不可視な操作順序によって不正確なシステム状態が生じる場合です。

• 個々の操作が並行安全なら、レース検出器は問題を発見しません。
• 完全に排除することはできませんが、Mutex で複合操作を保護することで防ぐことができます。

Compare-and-Swap (CAS)

Mutex を使用せずに競合条件を防ぐための原子比較・セット操作です。

• CompareAndSet(old, new)

  : 現在の値が

  old

  と一致する場合に

  new

  に変更する。（成功か否かを返す）

• CompareAndSwap(old, new)

  : 同様に動作し、古い値を返す。

• CompareAndDelete(old)

  : 現在の値が

  old

  と一致する場合に削除する。

仕組み: 仮定した（古い）状態が現実と一致するかチェックします。一致すれば変更し、不一致なら何もしません。

Mutexes (排他ロック)

sync.Mutex

使用場面

• 複数の goroutine が同じデータを修正する場合。
• 1 つの goroutine がデータを修正し、他のものが読み取っている場合。
• 例外: すべての goroutine が読み取りのみを行う場合は Mutex の必要はありません。

TryLock

TryLock

false

RWMutex

sync.RWMutex

• Lock / Unlock

  : 読み取りと書き込みの両方に使用。

• RLock / RUnlock

  : 読み取り専用。

動作ルール:

1. Lock()

   でロックされた場合、他はブロックされる。

2. RLock()

   でロックされた場合、他の goroutine も

   RLock()

   でロックできる（ブロックされない）。
3. 少なくとも 1 つの goroutine が

   RLock()

   中なら、

   Lock()

   はブロックされる。

これにより、**「単一書き込み・複数読み取り」**セッティングが実現できます。

Locker Interface

両方のタイプは

sync.Locker

チャンネルを Mutex として使用

共有データを保護するためにチャネル（バッファサイズ 1）を使用することもできます。

Semaphores (シグナル)

セマフォは N つの可用スロットを持つコンテナのようなもので、

acquire

release

• acquire

  : スロットがない場合は呼び出し側 goroutine をブロックする。

• release

  : ブロック中の goroutine がいたら、解放されたスロットを取得して解除される。

実装方法

• バッファ付きチャネル（サイズ N）で簡単に実装できます。（送信=acquire、受信=release）
• より複雑な状況には

  golang.org/x/sync/semaphore

  パッケージを使用します。

Rendezvous (待ち合わせ)

2 つの goroutine が互いに待ち合わせる機能です。

• G1 が準備をシグナルしても G2 でなければブロックする。
• 両方がシグナルしたら、双方が解除され継続する。

Wait Group を使用して簡単に実装できます。

Barrier (バリア)

Rendezvous の一般化です。N つの goroutine が互いに待つことができます。

• カウンター（初期値 0）があり、閾値 N があります。
• goroutine がバリアに到達するとカウンターを +1 し、バリアはそれまでの goroutine をブロックします。
• カウンターが N に達すると待機中のすべての goroutine が解除されます。

Signaling (シグナリング)

sync.Cond

• Wait

  : Mutex をロック解除し、シグナルを受けるまで停止する。

• Signal

  : 待機中の goroutine の一つを覚醒させる（Mutex は再ロックされる）。

• Broadcast

  : 全ての待機中 goroutine を覚醒させます。

チャネルとの違い:

• 条件変数でのブロードキャストはデータを送信せず一度きりです。
• チャネルでは Pub/Sub システムを構築できます。

Run Once (単発実行)

sync.Once

• 複数の goroutine が同時に

  Once.Do

  を呼び出した場合、1 つだけが関数を走り、残りは完了を待ちます。
• 用途: 初期化やクリーンアップに最適です。

その他の便利な once 関数

• Do(f func())

  : f を一度だけ呼び出す。

• OnceFunc(f func()) func()

  : f を一度だけ呼び出す関数を返す。

• OnceValue[T](f func() T) func() T

  : 最初の呼び出しの値を返す。

• OnceValues[T1, T2](f func() (T1, T2)) func() (T1, T2)

  : 最初の呼び出しのペアを返す。

Object Pool (オブジェクトプール)

sync.Pool

• Get()

  : プールから項目を取得。なければ

  New

  で作成する（自分で定義が必要）。

• Put(item)

  : 項目をプールに戻す。

注意点

1. New

   はポインタを返すべきです: メモリコピーや追加割り当てを防ぐため。
2. サイズ制限なし: 同時 1000 goroutine が

   Get

   を呼べば、1000 のバッファが作成されます。
3. 再利用性:

   Put

   された項目は直ちに再利用される可能性があるため、再度使用してはいけません（リサイクル可能状態）。

Atomics (原子操作)

同期なしの操作は単一のプロセッサ命令に翻訳されないと真に原子になりません。これらはロックを必要とせず、並行呼び出しでも問題ありません。

sync/atomic パッケージのタイプ

• Int32

  ,

  Int64

  ,

  Uint32

  ,

  Uint64

  ,

  Pointer

  ,

  Bool

  など。

提供されるメソッド:

• Load()

  : 値を読み取る。

• Store(value)

  : 新しい値を設定する。

• Swap(new)

  : 新しい値を設定し、古い値を返す。

• CompareAndSwap(old, new)

  : 現在の値が

  old

  と一致する場合に

  new

  に設定する。

追加機能: 数値タイプには

Add(delta)

重要：合成操作は非原子

// 危険な例（レース条件）
counter.Add(1)
time.Sleep(...) // スリープして間を開ける
delta := counter.Load() 
counter.Add(delta) 

• delta.Add(1)

  後にスリープして再度加算するのは危険です。
• これを防ぐには Mutex を使用します。

Testing (テスト)

並行プログラムがチャネルや同期メソッド (

Wait

コードに適した同期ハンドラがない場合、

synctest

• synctest.Test

  : 隔離されたバブルでテストを走行させる（フィケイクロックを使用）。

• synctest.Wait

  : バブル内のすべての goroutine（呼び出し元を除く）が完了または永続的にブロックするまで待機する。

永続的ブロックとみなされるもの:

• チャネルでの blocking send/receive
• すべてのケースがチャネルの blocking select

• Cond.Wait

• Wait Group の Add がバブル内で行われた場合の

  Wait

• time.Sleep

処理できないブロック（永続的ではない）:

• Mutex、I/O、システムコールでのブロック。

Scheduling (スケジューリング)

• ハードウェア: CPU コアが並行タスクを実行。
• OS レベル: スレッドが基本単位で、スケジューラが決定する。
• Go ランタイムレベル: goroutine が基本単位で、ランタイムスケジューラーが OS スレッド（通常 CPU コア数分）上で多くの goroutine を実行・待機させる。

スケジューリングアルゴリズム

1. 空きスレッドにキューから goroutine を割り当てる。
2. 実行中の goroutine がブロックしたら、キューに戻し別の goroutine を割り当てる。
3. システムコールで止まったら新しいスレッドを起動して他の goroutine を走らせる。
4. 10ms ごとにチェック: 長時間実行されている goroutine をプリエンプトしてキューに戻す（餓死防止）。

• Goroutine スケジューラは M goroutine を N OS スレッドに実行します（M >> N が可能）。
• ゴルーティンのスタックサイズは通常2KBで必要に応じて拡張できます。
• 軽量のため、小さなマシンでも数万〜数十万個を走らせることができます。
• スレッド数は

  GOMAXPROCS

  環境変数または

  runtime.GOMAXPROCS

  で制御されます。

Diagnostics (診断)

生産環境での並行プログラムトラブルシューニングにはメトリクス、プロファイリング、トレースを使用します。

• Metrices: メモリ使用量、GC パウゼーション時間など。
  • Prometheus や OpenTelemetry 経由で自動エクスポートされます。
• Profiling:
  • CPU プロファイル（関数ごとの処理時間）、ヒーププロファイル（メモリ使用）など。
  • Goroutine/block/mutex プロファイル（並行問題特定）。

  • /debug/pprof/{name}

    エンドポイントで収集し、

    go tool pprof

    で確認。
• Tracing:
  • 並行・メモリ関連イベントの記録。

  • /debug/pprof/trace

    で収集し、

    go tool trace

    で確認。
  • フライトレコーディング（スライディングウィンドウ）で自動的に最近のトレースを維持できます。

Final Thoughts (結び)

本書では、並行プログラムを書くための Go ツールを概観しました：

• Goroutines: 並行タスク実行
• Channels & Select: 柔軟な通信ツール
• Timers & Tickers: タイム処理
• Context: 操作キャンセル
• Wait Groups: goroutine 同期
• Mutexes: データ競合防止
• Condition Variables: イベントシグナリング
• Once: 安全な一回実行初期化
• Pools: GC 負荷削減
• Atomic Operations: ロックレス加算・比較

本書が気に入ったらご友人や同僚に推奨ください。興味があれば他の書籍もご覧ください。ありがとうございました。