goroutineを使った並行処理の基礎

Goを使う利点の一つとして、goroutineを使った並行処理の書きやすさがあります。
具体的にどんな感じで書くのかサンプルを見ながら理解を深めます。
なんとなくGoだと並行処理書きやすそうだなと思ってもらえると良いかなと思います。

goroutineとは

Goで並行処理を実現するための仕組みで、Goランタイムによって管理される軽量スレッドのことです。
並行処理として実行したい処理の前にgoキーワードをつけると並行処理として実行できます。
ちなみにgoroutineで実現できるのは並行処理(concurrent)であり、並列処理(parallel)ではありません。
（このコーヒーマシンの例が個人的にはわかりやすい気がします）

例えば以下のように使用します。

main.go

package main

import (
    "fmt"
    "time"
)

func main() {
    // 呼び出す関数の前にgoキーワードを付けて呼び出す
    go hello()
    time.Sleep(time.Second * 2)
}

func hello() {
    fmt.Println("Hello")
}

実行結果

Hello

簡単ですね。
ちなみにtime.Sleepなしだと、goroutineの処理が終わる前にmainの実行が終了してしまい、何も出力されません。

WaitGroupを使った制御

time.Sleepでの制御では処理時間等の不確定な要素に対応しきれないことが多いため、WaitGroupを利用した制御をすることが多いです。
先程のサンプルをWaitGroupを使って書き換えてみます。

main.go

package main

import (
    "fmt"
    "sync"
)

func main() {
    // WaitGroupを宣言する
    wg := new(sync.WaitGroup)
    // 終了待ちするgoroutineの数を設定する
    wg.Add(1)
    // goroutineとして呼び出す
    go hello(wg)
    // WaitGroupに設定された数だけDoneが実行されるまで待機
    wg.Wait()
}

// 引数にWaitGroupを受け取るように修正している
func hello(wg *sync.WaitGroup) {
    fmt.Println("Hello")
    // 処理終了時にDoneを実行する
    wg.Done()
}

WaitGroupを使用することで、すべてのgoroutineが終了した時点で
次の処理へ移ることができるようになりました。

もう少し並行処理っぽく、例えばfor文でループする場合は以下のような形になります

main.go

package main

import (
    "fmt"
    "sync"
)

func main(){
    wg := new(sync.WaitGroup)
    for i := 0; i < 5; i++ {
        // 終了待ちするgoroutineの数を設定する（今回の場合は最終的に5回分加算される）
        wg.Add(1)
        go hello(wg)
    }
    wg.Wait()
}

func hello(wg *sync.WaitGroup) {
    fmt.Println("Hello")
    wg.Done()
}

実行結果

Hello
Hello
Hello
Hello
Hello

また、wg.Done()を実行するときはFile IOのときと同じようにdeferをつけてあげるとGoっぽくなります。

main.go

// deferを使うように修正
func hello(wg *sync.WaitGroup) {
    defer wg.Done()
    fmt.Println("Hello")
}

channelを使って値の送受信を行う

これまでの例ではgoroutineで処理を実行しましたが、実行結果は受け取っていませんでした。
channelという仕組みを使うことでgoroutineで値のやり取りを行うことができます。

今回の例では受け取った文字列に「World」という文字列を付与する処理をgoroutineとして実行し、結果を受け取ってみます。

main.go

package main

import (
    "fmt"
)

func main() {
    // goroutineに渡す文字列
    hello := "Hello"
    // channelを用意する
    c := make(chan string)
    
    // goroutineとして実行
    go world(hello, c)
    
    // channelに入った値を受け取る
    x := <- c
    fmt.Println(x)
}

func world(str string, c chan string) {
    // channelにデータを送信する
    c <- str + "World"
}

実行結果

HelloWorld

channelを使うときはchanというキーワードを使用し、make(chan 型)と書くことでchannelを生成できます。
channelはキューのようなデータ構造になっており、上記のようにgoroutineに引数として渡すことで各goroutineから参照することができます。
channelにデータを渡すときにはc <- データ、データを取り出すときは<- cのように書きます。
右から左にデータが向かっていくようなイメージを持つと覚えやすいです。

また、channelからデータを取り出す際は、データが入ってくるまで待ち受けるため、
WaitGroupのような制御をしなくてよかったりします。

ちなみに以下のような例で、データを取り出す際に無限に待ちが発生するような場合だとdeadlockとエラーが発生します。

main.go

package main

import (
    "fmt"
)

func main() {
    hello := "Hello"
    c := make(chan string)

    // データが5回入る
    for i := 0; i < 5; i++ {
        go world(hello, c)
    }

    // 取り出しは6回なので無限に待ちが発生する
    for i := 0; i < 6; i++ {
        x := <-c
        fmt.Println(x)
    }
}

func world(str string, c chan string) {
	c <- str + "World"
}

実行結果

HelloWorld
HelloWorld
HelloWorld
HelloWorld
HelloWorld
fatal error: all goroutines are asleep - deadlock!

goroutine 1 [chan receive]:
main.main()
        /workspaces/goroutine-sample/main.go:16 +0xba
exit status 2

rangeを使ったループ

先程channelはキューのようなデータ構造という話をしましたが、len()やrangeといった、配列やスライスで使うような構文がいくつか使えます。
例えばrangeを使ったループは以下のように記述します。

main.go

package main

import (
    "fmt"
)

func main() {
    hello := "Hello"
    c := make(chan string)

    go world(hello, c)

    // channelに入ったデータを逐次受け取る
    for result := range c {
        fmt.Println(result)
    }
}

func world(str string, c chan string) {
    // 5回データをchannelに送信する
    for i := 0; i < 5; i++ {
        c <- str + "World"
    }
    // 【重要】データを送信し終わったらcloseする
    close(c)
}

channelの場合は受信データを待ち受けてしまったり、中身が可変になる可能性があるため、
データが確定したあとにclose()してあげないとrangeを使ってもエラーが発生します。

実行結果（closeなしの場合）

HelloWorld
HelloWorld
HelloWorld
HelloWorld
HelloWorld
fatal error: all goroutines are asleep - deadlock!

goroutine 1 [chan receive]:
main.main()
        /workspaces/goroutine-sample/main.go:13 +0x10e
exit status 2

Buffered channelを使う

先程channelはキューのようなデータ構造という話をしましたが、キューに格納するデータ数を制限するためにBeffered channelという仕組みがあります。

main.go

package main

import "fmt"

func main() {
    // channel生成時にバッファの数を指定する
    c := make(chan string, 1)

    c <- "Hello"
    fmt.Println(len(c)) // ここまでは実行される
    c <- "World" // ここでエラーが発生する
}

実行結果

1
fatal error: all goroutines are asleep - deadlock!

goroutine 1 [chan send]:
main.main()
        /workspaces/goroutine-sample/main.go:11 +0xe3
exit status 2

実行数が限られているような場合はこの仕組みを使った方が予想外の事故を防げて良いかもしれません。

最後に

今回紹介したのは基本的な部分になります。
実際に使っていこうと思うと排他制御など考えないといけないことがたくさんありそうです。
Goだと比較的簡単に書けるけど、なんだかんだで並行処理ってやっぱ複雑なんだなと個人的には思いました。

参考

https://go-tour-jp.appspot.com/concurrency/1
https://qiita.com/gold-kou/items/8e5342d8a30ae8f34dff