並行・非同期処理¶
はじめに¶
本サイトにつきまして、【IT技術の知見】はじめに - 俺の技術ノート をご認識のほど宜しくお願いいたします。
01. 処理の種類¶
単一処理¶
▼ 単一処理とは¶
特定の処理のみを実行する。
package main
import "fmt"
func main() {
fmt.Println("1")
fmt.Println("2")
fmt.Println("3")
}
// 1
// 2
// 3
並行処理¶
▼ Goroutine による並行処理¶
指定した処理をバラバラに開始し、またそれぞれの処理がバラバラに終了する。
関数で Goroutine (go func()) を宣言すると、その関数の完了を待たずに後続の処理を並行的に実行できる。
結果、完了する順番は順不同になる。
プログラムを実行する親プロセス
├─ スレッド
│ └─ プロセッサ
│ ├─ goroutine #1（HTTPリクエスト処理）
│ └─ goroutine #3（トランザクション処理）
│
├─ スレッド
│ └─ プロセッサ
│ ├─ goroutine #1（HTTPリクエスト処理）
│ └─ goroutine #2（ファイルI/O処理）
│
└─ スレッド
└─ プロセッサ
├─ goroutine #1（HTTPリクエスト処理）
└─ goroutine #2（計算処理）
＊実行例＊
package main
import (
"time"
"fmt"
)
func main() {
fmt.Println("main")
go func() {
fmt.Println("hoge")
}()
time.Sleep(time.Second)
}
▼ 時間のかかる処理と相性がいい¶
時間のかかる処理（例：for 文、HTTP リクエスト、ループ、ポーリング、DB 操作、ファイル操作、計算、暗号化、圧縮解凍など）と相性がいい。
package main
import (
)
func main() {
go func() {
// for文、HTTPリクエスト、ポーリング、DB操作、ファイル操作、計算、暗号化、圧縮解凍など
// 実装方法を調べて書く
}()
// 並行して実行したい処理
}
▼ 返却処理はエラーになる¶
返却処理 (return) は、処理の完了を待つことになるため、Goroutine とは矛盾する。
そのため、Goroutine で実行する関数に返却処理があると、エラーになる。
代わりに、Goroutine 間で channel を介して処理結果を送受信する。
package main
import (
"fmt"
)
func main() {
// "unexpected go" というエラーになる
result := go foo()
fmt.Println(result)
}
func foo() string {
return "foo"
}
▼ Goroutine を入れ子にできる¶
入れ子で実行した場合でも、全く独立して Goroutine を実行する。
Goroutine の親子間に依存関係はない。
package main
import (
"fmt"
)
func main() {
go foo()
// Goroutineを宣言したfoo関数の完了を待たない
fmt.Println("baz")
}
func foo() string {
go func() {
fmt.Println("bar")
}
// Goroutineを宣言した無名関数の完了を待たない
fmt.Println("foo")
}
02. Goroutine と合わせて使用する処理¶
channel (チャネル)¶
▼ channel とは¶
異なる Goroutine 間で値を送受信するキューとして動作する。
キューに値を送信し、加えてキューから値を受信できる。
package main
import "fmt"
func main() {
// チャネルを作成
channel := make(chan string)
// Goroutineを宣言して並行化
go func() {
// 時間のかかる処理
// チャネルに値を送信する。
channel <- "ping"
}()
// チャネルから値を受信する。
value := <-channel
fmt.Println(value)
}
▼ Goroutine 中断方法１ (done channel、close)¶
Goroutine を中断するための done チャネルを作成し、これを close() 関数で中断する。
done チャネルは使い捨てのため、サイズがゼロの空構造体として定義するとよい。
package main
import (
"fmt"
"log"
)
func main() {
// チャネルを作成
channel := make(chan string)
// Goroutineを中断するためのdoneチャネルを作成
// 空構造体はサイズがゼロなため、使い捨てのチャネルと相性がいい
done := make(chan struct{})
// Goroutineを宣言して並行化
go func() {
// 時間のかかる処理
// チャネルに値を送信する。
channel <- "ping"
// Goroutineを中断する
close(done)
}()
for {
select {
// 先に終了したcaseに条件分岐する
// チャネルから値を受信した場合
case value := <-channel:
fmt.Println(value)
// cancel関数が実行された場合
case <-done:
log.Printf("Goroutineが完了しました")
return
}
}
}
▼ Goroutine 中断方法２ (context.cancel、context.Done)¶
cancel() 関数の Goroutine の中断を検知する。
sync.WaitGroup.Done() 関数とは区別する。
package main
import (
"time"
"context"
"fmt"
"log"
)
func main() {
// タイムアウト時間を設定し、コンテキストを作成する
ctx, cancel := context.WithTimeout(
context.Background(),
5 * time.Second,
)
// チャネルを作成
channel := make(chan string)
// Goroutineを宣言して並行化
go func() {
// 時間のかかる処理
// チャネルに値を送信する。
channel <- "ping"
// Goroutineを中断する
cancel()
}()
for {
select {
// 先に終了したcaseに条件分岐する
// チャネルから値を受信した場合
case value := <-channel:
fmt.Println(value)
// cancel関数が実行された場合
case <-ctx.Done():
log.Printf("Goroutineが完了しました")
return
}
}
}
▼ select¶
チャネルに対する格納を非同期で待機する。
package main
import (
"fmt"
"time"
)
func main() {
// チャネルを作成する。
channel1 := make(chan string)
channel2 := make(chan string)
// Goroutineを宣言して並行化
go func() {
// 時間のかかる処理
// 完了までに2秒かかるとする。
time.Sleep(2 * time.Second)
// 値を送信する。
channel1 <- "one"
}()
// Goroutineを宣言して並行化
go func() {
// 時間のかかる処理
// 完了までに1秒かかるとする。
time.Sleep(1 * time.Second)
// 値を送信する。
channel2 <- "two"
}()
for i := 0; i < 2; i++ {
select {
// Goroutineの処理の完了タイミングがバラバラになる
// channel1とchannel2の受信を非同期で待機し、受信した順番で処理する。
// channel1 <- "one" を受信したタイミングで出力する
case msg1 := <-channel1:
fmt.Println("received", msg1)
// channel2 <- "two" を受信したタイミングで出力する
case msg2 := <-channel2:
fmt.Println("received", msg2)
// 受信が成功しなければ、defaultで処理する。
default:
fmt.Println("default")
}
}
}
▼ WaitGroup との使い分け¶
channel を使用すると、異なる Goroutine 間で値を送受信できる。
そのため、Goroutine の処理結果を使用したい場合は、WaitGroup ではなく channel を使用するほうがよい。
一方で、channel では select() 関数が必要になって、処理が複雑になる点は注意する。
WaitGroup¶
▼ WaitGroup とは¶
Goroutine を宣言した関数が終了するまで、後続の処理の実行開始を待機する。
sync.WaitGroup.Add() 関数、sync.WaitGroup.Done() 関数、sync.WaitGroup.Wait() 関数、で制御する。
Goroutine の関数の反復処理や異なる Goroutine の関数の並行実行を待機し、そのうえで後続の処理を実行するような場合に、WaitGroup は役立つ。
単一の Goroutine を待機するのは順次実行と変わらないため、WaitGroup は使用しない。
実行例
実行完了に 1 秒かかる関数があると仮定する。
反復処理でこの関数をコールする場合、毎回の走査に 1 秒かかるため、反復の回数だけ秒数が増える。
しかし、各関数を Goroutine として実行すると、各反復処理を並行実行できる。
そのため、反復回数が何回であっても、およそ 1 秒で処理が終了する。
package main
import (
"sync"
"log"
"fmt"
"time"
)
func main() {
slice := []string{"a", "b", "c"}
wg := &sync.WaitGroup{}
// 処理の開始時刻を取得
start := time.Now()
// Goroutineの関数を反復処理する
// a、b、c、の分だけ反復する
for key, value := range slice {
// インクリメントする
// 完了の待機カウンターを設定する
wg.Add(1)
// Goroutineを宣言して並行化
go func(key int, value string) {
print(key, value)
// デクリメントする
// Wait関数のカウンターを減らす
wg.Done()
}(key, value)
}
// カウンターがゼロになるまで待機する
// 全てのGorourineでDone関数を実行するのを待機する
wg.Wait()
// 開始時刻から経過した秒数を取得する
log.Printf("経過秒数: %s\n", time.Since(start))
}
func print(key int, value string) {
fmt.Println(key, value)
}
// 2 c
// 0 a
// 1 b
// 経過秒数: 1s
▼ channel との使い分け¶
WaitGroup を使用すると、Add() 関数、Done() 関数、Wait() 関数を使用して、Goroutine を簡単に制御できる。
そのため、Goroutine の処理結果を使用したい場合は、WaitGroup ではなく channel を使用するほうがよい。
一方で、WaitGroup では Goroutine 間で値の送受信はできないことに注意する。
errgroup¶
エラー処理を含む関数で Goroutine を宣言したいときに使用する。