Go言語、ゴルーチン(goroutine)で並列処理を

1 はじめに
2 ゴルーチン
3 チャネル
4 コンテキスト
- 4.1 基本的な使い方
- 4.2 タイムアウトのサンプル
5 さいごに
6 おすすめ書籍

はじめに

こんにちは、nukkyです。Go連載７回目を担当させていただきます。

プログラムを書く上ではなかなか避けては通れない非同期処理について今回は記事にしたいと思います。

ゴルーチン

Goにはスレッドより小さい単位「ゴルーチン」が並行して動作する様に設計されています。

go文

ゴルーチンを実行するにはgo文で関数を呼び出します。次のコードをみてください。

func sub() {
    for {
        fmt.Println("sub loop")
    }
}

func main() {
    go sub() // ゴルーチン開始
    for {
        fmt.Println("main loop")
    }
}

func sub() {

for {

fmt.Println("sub loop")

}

func main() {

go sub() // ゴルーチン開始

for {

fmt.Println("main loop")

}

goの後に関数subを呼び出すことでゴルーチンを開始しています。

ゴルーチンの終了条件

ゴルーチンは、下記の条件で終了されます。

関数が終わる
returnで抜ける
runtime.Goexit()を実行する

たとえば、Go文のサンプルを以下の様に書き換えた場合、

func sub() {
    for {
        fmt.Println("sub loop")
    }
}

func main() {
    go sub() // ゴルーチン開始
}

func sub() {

for {

fmt.Println("sub loop")

}

func main() {

go sub() // ゴルーチン開始

}

この場合mainが終了した際にmainで実行したゴルーチンは終了されてしまいます。なぜかというとmain関数も1つのゴルーチンの中で実行されており、mainの中で宣言したゴルーチンが実行されるにはmain関数も実行中(終了されていない)でないといけないからです。

WaitGroup

ゴルーチンですが宣言から実行されるまでの間隔は保証されておらずmainでスリープなどしても何秒待てばよいかはわかりません。

そこでsyncパッケージのWaitGroupを使いゴルーチンの終了を待ってみます。

import (
    "fmt"
    "sync"
)

func sub() {
    fmt.Println("sub")
    defer wg.Done()   // 関数終了時にデクリメント
}

func main() {
    var wg sync.WaitGroup
    wg.Add(1) // ゴルーチン起動のたびにインクリメント
    go sub()  // ゴルーチン開始
    wg.Wait() // 処理をブロックし、カウンタが0になったら次に進む
    fmt.Println("Done.")
}

import (

"fmt"

"sync"

)

func sub() {

fmt.Println("sub")

defer wg.Done() // 関数終了時にデクリメント

}

func main() {

var wg sync.WaitGroup

wg.Add(1) // ゴルーチン起動のたびにインクリメント

go sub() // ゴルーチン開始

wg.Wait() // 処理をブロックし、カウンタが0になったら次に進む

fmt.Println("Done.")

}

WaitGroupは厳密にどのゴルーチンが終了したかは管理せず、WaitGroup内のカウンタで実行中か終了かを判断しています。

チャネル

チャネルとは、ゴルーチンとゴルーチンの間でデータの受け渡しを行うためにデザインされたGo特有のデータ構造です。

チャネルの型

チャネルの型名は「chan [データ型]」のように記述します。

// int型のチャネル
var ch chan int

1 2	// int型のチャネル var ch chan int

チャネルには他のデータ型にはない特殊なサブタイプを指定できます。「<-chan」を使用すると受信専用チャネル、「chan<-」を使用すると送信専用チャネルになります。サブタイプを指定しない「chan」は送受信可能な双方向チャネルとして機能します。

チャネルの生成

チャネルはmakeを使って生成します。makeへの２番目の引数を指定することでバッファサイズを指定できます。

// バッファサイズ0のチャネル
ch := make(chan int)
// バッファサイズ8のチャネル
ch := make(chan int, 8)

// バッファサイズ0のチャネル

ch := make(chan int)

// バッファサイズ8のチャネル

ch := make(chan int, 8)

チャネルはキューの性質をもつデータ構造になっており、チャネルのバッファとはキューを格納する領域であるためバッファサイズとはキューのサイズとみなすことができます。キューには「FIFO(先入れ先出し)」の性質があり、先にキューに挿入されたデータを先に取得できるという特性があり、これはデータを取り出す順序が保証されるという性質があります。

チャネルの送受信

チャネルが保持するデータにアクセスするには送信か受信の２パターンのみです、送受信ともに演算子「<-」を使用します。

ch := make(chan int, 10)

// チャネルに整数を送信
ch <- 5
// チャネルから整数を受信
i := <-ch

ch := make(chan int, 10)

// チャネルに整数を送信

ch <- 5

// チャネルから整数を受信

i := <-ch

チャネルとゴルーチン

チャネルを使用することでゴルーチン間での値のやりとりが可能になります。次のコードを見てください。

package main

import (
    "fmt"
    "math/rand"
    "time"
)

func init() {
    rand.Seed(time.Now().UnixNano())
}

func from(ch chan<- int) {
    ch <- rand.Intn(100)
}

func to(ch <-chan int) {
    fmt.Printf("Received %d\n", <-ch) // chに値が入ってくるまで待機
}

func main() {
    ch := make(chan int)
    go from(ch)
    go to(ch)
    time.Sleep(100 * time.Millisecond)
}

package main

import (

"fmt"

"math/rand"

"time"

)

func init() {

rand.Seed(time.Now().UnixNano())

}

func from(ch chan<- int) {

ch <- rand.Intn(100)

}

func to(ch <-chan int) {

fmt.Printf("Received %d\n", <-ch) // chに値が入ってくるまで待機

}

func main() {

ch := make(chan int)

go from(ch)

go to(ch)

time.Sleep(100 * time.Millisecond)

}

この様に呼び出し元でチャネルを宣言しゴルーチンに引き渡すことで、チャネルを使用した値のやりとりが可能になります。

チャネルのクローズ

使用しなくなったチャネルはクローズ(close)する。こうすることでクローズ済みのチャネルに値を送信することはできなくなります。

close(ch)

close(ch)

チャネルをクローズしてもバッファには値が残っており、クローズ済みのチャネルから値を受信しようとすると、チャネルでバッファされている値がなくなるまで受信でき、バッファ内の値を全て受信し尽くした後は、そのチャネルの要素型のゼロ値が返る様になります。

select

次の様な単純なプログラムの場合、変数ch1とch2はチャネル型、変数e1とe2はチャネルが内包するデータ型とした場合、

e1 := <-ch1
e2 := <-ch2

1 2	e1 := <-ch1 e2 := <-ch2

変数ch1の指すチャネルからデータが受信できない場合、この処理を実行しているゴルーチンは停止してしまいます。

複数のチャネルを扱いながら処理を停止させずに行いたい場合はselectを使います。

select {
case e1 := <- ch1:
    // ch1から受信が成功した場合の処理
case e2 := <- ch2:
    // ch2から受信が成功した場合の処理
default:
    fmt.Println("no value")
}

select {

case e1 := <- ch1:

// ch1から受信が成功した場合の処理

case e2 := <- ch2:

// ch2から受信が成功した場合の処理

default:

fmt.Println("no value")

}

複数のチャネルへの送信、受信処理をゴルーチンを停止させることなくコントロールすることが可能になります。

コンテキスト

Goの典型的な利用例であるWebアプリケーションを考えた場合、Goのサーバにおいてリクエストはそれぞれ個別のゴルーチンで処理されます。リクエストハンドラは新たなゴルーチンを生成しバックエンドのDBや別のサーバにリクエストを投げ結果を得てユーザに対してレスポンスを返すといった流れがあった際に注意すべきなのは、その処理に適切なTimeoutやDeadlineを設定して処理が停滞するのを防ぐこと、例えば別のサーバにリクエストを投げる場合にネットワークの問題でリクエストに時間がかかってしまうことが考えられます。この場合リクエストにTimeoutを設定して早めにレスポンスを返しリトライを促すべきだと思います。

さらに生成したゴルーチンを適切にキャンセルしリソースを解放することも求められます。例えば別のサーバにリクエストを投げる場合に適切なキャンセル処理を行わないとTimeout後もネットワークリソースが使われ続けることになるかもしれないからです。さらにネットワークアクセスに使用したゴルーチンは別のゴルーチンを呼び出しそれがまた別のと呼び出しの連鎖が考えられます。その場合も親のTimeoutに合わせてその子は全て適切にキャンセルされリソースは解放されることが望ましいです。

contextパッケージはキャンセルのためのシグナルをAPIの境界を超えて受け渡すための仕組みです。ある関数から別の関数、親から子へとキャンセルを伝搬させることができます。

基本的な使い方

ゴルーチンを呼び出す元の方でオブジェクトを生成します。「context.Background()」は空のコンテキストを生成します(nilではない)。

ctx := context.Background()

1	ctx := context.Background()

これをゴルーチンに引き渡し伝搬させていきます。

go hoge(ctx)

1	go hoge(ctx)

タイムアウトでキャンセルする場合はWithTimeoutをつかいます。

ctx, cancel := context.WithTimeout(ctx, time.Second) // 1秒後にキャンセル
defer cancel()

1 2	ctx, cancel := context.WithTimeout(ctx, time.Second) // 1秒後にキャンセル defer cancel()

手動でキャンセルする場合はWithCancelを使います。

ctx, cancel := context.WithCancel(ctx)
// 処理中略
cancel() // 即キャンセル

ctx, cancel := context.WithCancel(ctx)

// 処理中略

cancel() // 即キャンセル

キャンセルされたかの判断はselectでコンテキストのDoneを待ちます。

select {
case <-ctx.Done():
fmt.Println("done:", ctx.Err()) // done: context canceled
}

select {

case <-ctx.Done():

fmt.Println("done:", ctx.Err()) // done: context canceled

}

タイムアウトのサンプル

コンテキストのタイムアウトを使用したキャンセルのサンプルになります。

package main

import (
    "context"
    "fmt"
    "time"
)

func infiniteLoop(ctx context.Context) {
    innerCtx, cancel := context.WithCancel(ctx)
    defer cancel()
    for {
        fmt.Println("Waiting for time out")

        select {
        case <-innerCtx.Done():
            fmt.Println("Exit now!")
            return
        default:
        }
    }
}

func main() {
    // コンテキストの生成
    ctx := context.Background()
    // 5秒後にキャンセル
    ctx, cancel := context.WithTimeout(ctx, 5*time.Second)
    defer cancel()

    go infiniteLoop(ctx)

    select {
    case <-ctx.Done():
        fmt.Println(ctx.Err()) // context deadline exceeded
    }
}

package main

import (

"context"

"fmt"

"time"

)

func infiniteLoop(ctx context.Context) {

innerCtx, cancel := context.WithCancel(ctx)

defer cancel()

for {

fmt.Println("Waiting for time out")

select {

case <-innerCtx.Done():

fmt.Println("Exit now!")

return

default:

}

func main() {

// コンテキストの生成

ctx := context.Background()

// 5秒後にキャンセル

ctx, cancel := context.WithTimeout(ctx, 5*time.Second)

defer cancel()

go infiniteLoop(ctx)

select {

case <-ctx.Done():

fmt.Println(ctx.Err()) // context deadline exceeded

}

実行結果

Waiting for time out
Waiting for time out
Waiting for time out
.
.
.
Waiting for time out
Exit now!
context deadline exceeded

Waiting for time out

Exit now!

context deadline exceeded

さいごに

Goの非同期処理について調べましたが、Goは非同期について他の言語よりも取り扱いがしやすい印象を受けました、この記事がGo初心者の方の一助になれれば嬉しいです。

Go記事の連載などは、こちらをご覧ください。

日	月	火	水	木	金	土
	1	2	3	4	5	6
7	8	9	10	11	12	13
14	15	16	17	18	19	20
21	22	23	24	25	26	27
28	29	30	31

Go言語、ゴルーチン(goroutine)で並列処理を

はじめに

ゴルーチン

go文

ゴルーチンの終了条件

WaitGroup

チャネル

チャネルの型

チャネルの生成

チャネルの送受信

チャネルとゴルーチン

チャネルのクローズ

select

コンテキスト

基本的な使い方

タイムアウトのサンプル

さいごに

おすすめ書籍

フォロー

アプリ情報