【Go言語】Goの並行処理を理解する

はじめに

今回は、Go言語の並行処理についてまとめていこうと思います。
Go言語の勉強をするにあたって避けては通れない並行処理ですが、イメージがしにくいですよね。
本記事では、並行処理を図やコードと一緒にサクッとまとめてみました。

早速、並行処理と並列処理の違いについて解説していきます。

並行処理と並列処理の違い

まずは、「並行処理」と「並列処理」の違いを明確にしていきましょう。

• 並行処理

  • 複数のタスクを1箇所で同時に実行すること
  • 例えば、一人のシェフが複数の料理を同時に作る状況を想像してください。シェフは順番に各料理の調理を進め、全ての料理が同時に出来上がるようにします

• 並列処理

  • 複数のタスクを複数箇所で同時に実行すること
  • 複数のシェフが、それぞれ別々の料理を同時に作る状況を想像してください。各シェフは異なる料理に専念し、すべての料理が同時に出来上がります

これだけだと分かりかねる思うので、以下のイメージをご覧ください。

処理すべき複数のタスクを一箇所で処理していくのが並行処理です。
処理すべき複数のタスクを、二箇所に分けて同時に処理していくのが並列処理です。

goroutine（ゴールーチン）

Goで並行処理の核となる機能が、この「goroutine(ゴールーチン)」というものです。
Goでは複数のgoroutineを使用して並行処理を実現していきます。

goroutineの使い方

goroutineの使い方はとても簡単で、関数の前にgoと書くだけでその関数はgoroutineになります。
実際のコードを見てみましょう。

main.go

package main

import (
	"fmt"
	"time"
)

func say(s string) {
	for i := 0; i < 5; i++ {
		fmt.Println(s)
		time.Sleep(100 * time.Millisecond)
	}
}

func main() {
	go say("Hello") // goroutineとして起動
	say("Gopher")
}

The Go Playground：実行してみましょう。
"Hello"と"Gopher"の出力は並行して行われ、どちらが先に出力されるかはランダムになりますが、両方の文字列が交互に出力される様子が見られると思います。
このコードの中にあるtime.Sleep(100 * time.Millisecond)は各ループ後に100ミリ秒の遅延を発生させるものです。
なぜ、このような処理を書く必要があるのかこの箇所がある場合とない場合をそれぞれ実行して出力を比較してみましょう。

main.go

func say(s string) {
	for i := 0; i < 5; i++ {
		fmt.Println(s)
		// time.Sleep(100 * time.Millisecond)
	}
}

func main() {
	go say("Hello")
	say("Gopher")
}

The Go Playground：実行してみましょう。
time.Sleep(100 * time.Millisecond)をコメントアウトして実行してみると、"Gopher"のみが５回ループされていることが確認できたと思います。
これは、メインgoroutineの処理が完了したら他のgoroutineの処理を待たずプログラム全体が終了してしまうというGoの性質によるものです。

goroutineの処理は他の処理とは独立して実行されます。
しかしメインgoroutineが先に終了してしまうと、独立して切り離されたgoroutineは処理完了を待たずプログラムが終了してしまいます。
そのため、先ほどのtime.Sleep(100 * time.Millisecond)を書かない場合は以下のような状態になってしまいます。

そのため意図的に遅延させることで、新ゴールーチンの処理が完了するのを待つことができるようになるということです。

ただ実際のコードの中でこのような処理を書くことは現実的でない場合も多いのです。なので以下のようなアプローチを取ります。

• syncパッケージのWaitGroupの使用
• channel(チャネル)の使用
• Buffered channelの使用

以上の３点を順に解説していきます。

syncパッケージのWaitGroupの使用

実際にコードを見ていきましょう。

main.go

package main

import (
	"fmt"
	"sync"
)

func say(s string, wg *sync.WaitGroup) {
	for i := 0; i < 5; i++ {
		fmt.Println(s)
		// time.Sleep(100 * time.Millisecond)
	}
	wg.Done()
}

func normal(s string) {
	for i := 0; i < 5; i++ {
		// time.Sleep(100 * time.Microsecond)
		fmt.Println(s)
	}
}

func main() {
	var wg sync.WaitGroup
	wg.Add(1)
	//goroutineにwgのアドレスを渡す
	go say("Hello", &wg)
	normal("Gopher")
	// Doneが呼ばれるまで待つ
	wg.Wait()
}

The Go Playground：実行してみましょう。

• メインgoroutineがnormal("Gopher")を5回実行し終えるまで、go say("Hello", &wg)で始まったgoroutineは並行して実行されます

• normal関数が完了した後、say関数が実行され始めます

• wg.Wait()により、say関数内のwg.Done()が呼ばれるまでメインgoroutineは待機します

wg.Done()がないとエラーになるので気をつけましょう。

このようにsync.WaitGroupを使用することで他のゴールーチンの処理が終了するまでメインゴールーチンの完了を待つことができます。

channel(チャネル)の使用

channelはgoroutine間で値の送受信をするために使われます。
Goには「共有メモリよりも通信を通じてメモリを共有する」（Do not communicate by sharing memory; instead, share memory by communicating）という考え方があり、その通信を行うのには主にchannelが使用されます。
channelは、下図のようなイメージです。

実際にコードを見ていきましょう。

main.go

package main

import (
	"fmt"
)

func goroutine1(s []int, c chan int) {
	sum := 0
	for _, v := range s {
		sum += v
	}
	// channelに送信
	c <- sum
}

func main() {
	s := []int{1, 2, 3, 4, 5}
	// channelはmakeで作成できる
	c := make(chan int)
	//goroutineにsliceとchannelを渡す
	go goroutine1(s, c)
	// goroutineから送られてきたchannelをxに送信する
	x := <-c
	fmt.Println(x) //出力：15 
}

The Go Playground：実行してみましょう。
channelはmake関数を用いて作成することができます。
また、channel <- 送信する値と書くことでデータ送信の処理を、
<- c のように <- channelと書くことでデータ受信の処理を定義します。
channelを使用することで、goroutine間での同期やデータの共有を安全かつ効率的に行うことができます。

Buffered channelの使用

コードを見ていきましょう。

main.go

package main

import (
	"fmt"
)

func main() {
	ch := make(chan int, 2)
	ch <- 100
	fmt.Println(len(ch))
	ch <- 200
	fmt.Println(len(ch))
}

こちらは、バッファサイズ2でchannelを作成し、二つの要素(100, 200)を保持しています。では、こちらのコードに以下を追記して実行してみるとどうなるでしょうか。

    ch <- 300
    fmt.Println(len(ch))

The Go Playground：実行してみましょう。
これは、バッファサイズを2で設定しているのにも関わらず、3つ目を入れようとしているためエラーが出ましたね。

main.go

package main

import (
	"fmt"
)

func main() {
	ch := make(chan int, 2)
	ch <- 100
	fmt.Println(len(ch))
	ch <- 200
	fmt.Println(len(ch))

// channelを取り出してみる
	x := <-ch
	fmt.Println(x) // 出力：100

// 取り出した後のバッファサイズが1になっている
	fmt.Println(len(ch))

	ch <- 300
	fmt.Println(len(ch))
}

The Go Playground
：実行してみましょう。

channelから一つ値を取り出したことで300を入れることができました。
それでは、このchannelをfor文で取り出してみたいですよね。

main.go

package main

import (
	"fmt"
)

func main() {
	ch := make(chan int, 2)
	ch <- 100
	fmt.Println(len(ch))
	ch <- 200
	fmt.Println(len(ch))

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

The Go Playground
：実行してみましょう。
エラーが出ましたね。
これは、1つ目(100)を取り出し、2つ目(200)を取り出した後存在しない3つ目の値を取り出そうとしたためエラーが発生しました。
こういったchannelの取り出しをするときは、以下の通りにコードを修正しましょう。

package main

import (
	"fmt"
)

func main() {
	ch := make(chan int, 2)
	ch <- 100
	fmt.Println(len(ch))
	ch <- 200
	fmt.Println(len(ch))
 // closeしてあげる
	close(ch)

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

The Go Playground：実行してみましょう。
rangeで取り出すときは closeで channelの終了を教えてあげないといけません。

並行処理の注意点

コードの実行順が予測できない

Goにおける並行処理では、goroutineの実行順序がランタイムのスケジューリングやgoroutine間の相互作用、外部リソースの影響などにより実行するたびに変化するため実行順を予測することができません。

The Go Playgroundこちらで同じ処理を行っても、実行結果の順序がバラバラになってしまうことを確認できるかと思います。

この性質によりデータが予期せぬ方法で上書きされたりプログラムの挙動が実行するたびに異なり、デバッグが困難になる可能性があります。そのためGoで並行処理を行う際にはプログラムがgoroutineの実行順序に依存しないように設計する必要があります。

競合状態を避ける必要がある

複数のgoroutineが同じデータやリソースに同時にアクセスして読み書きする場合、競合状態が発生する可能性があります。これを防ぐために、以下の対策が必要となってきます。

• 排他制御の実施
• ゴルーチンより広いスコープを持つ変数への参照を避ける設計
• チャネルの適切な利用
  競合状態が発生しないように工夫していきましょう。

実行時間が速くなるとは限らない

• 並行処理では、goroutineの作成やスケジューリング、同期メカニズムの使用など追加的な処理が発生します。これらの追加的な処理の影響が大きい場合、並行処理による性能向上が相殺される可能性があります
• 先ほど並行処理の注意点に書いたように、並行処理は適切に使用しないと共有リソースへの同時アクセスやスレッドセーフでない操作により、競合状態が発生する可能性があります。これを防ぐための処理がボトルネックになり、ここでも並行処理の性能が相殺されてしまう可能性が出てきます
• 並行処理が有効なのは、「タスクが独立しており、分割可能である場合」です。しかし、タスクの特性によっては、並行処理に適さない場合もあります。例えば、タスクAの後にタスクB、その後にタスクCのように順を追って処理を行う必要がある場合、並行処理による性能向上は限られる可能性があります

goroutine leak（ゴールーチンリーク）

Goにおいて、goroutineが不要になったにもかかわらず終了しないで実行し続ける状態をgoroutine leakと言います。
goroutine leakが発生するとその処理に利用しているメモリスタック領域がガベージコレクトされないままになりパフォーマンスに悪影響を及ぼすことになります。
以下のコードで意図的にgoroutine leakを引き起こしてみましょう。

main.go

package main

import (
    "fmt"
    "time"
)

func main() {
    c := make(chan int)

    for i := 0; i < 10; i++ {
        go func(n int) {
            fmt.Println("Goroutine", n, "started")
            select {
            case <-c:
                // ここでブロックされる
            }
            fmt.Println("Goroutine", n, "finished")
        }(i)
    }
    // 少し待ってからプログラムを終了
    time.Sleep(2 * time.Second)
    fmt.Println("Main function finished")
}

The Go Playground：実行してみましょう。
こちらのコードでは、goroutine が channel c からのメッセージを待機する際に永遠にブロックされます。このため、goroutineは終了せずにメモリを消費し続けることになります。
goroutine leakはパフォーマンス低下に繋がるだけでなく、アプリケーションのクラッシュを引き起こす可能性もあるので適切に対処する必要があります。
goroutine leakが発生していないか検知してくれる外部パッケージもあるので、このようなツールを活用して対策をとるのも一つの手です。

さいごに

今回はGoの並行処理についてまとめてみました！
並行処理はGoを扱う上で非常に強力なツールですが、その挙動と特性を理解し、適切に使用しなければあまり意味がなく、むしろ悪影響を及ぼす可能性もあるので丁寧に理解した上で使っていきましょう。