goroutineによる頻出並行処理パターン2選
はじめに
goruotineはgo言語の軽量スレッドの仕組みであり、並行処理が比較的簡単に実装できます。しかしその自由度の高さから、慣れていない人にとってはどのように実装したらよいのか、という迷いが生まれてしまいます。
その中でもよく使う並行処理のパターンは決まっており、今回はよく自分が使うパターンを2つ紹介します。
前提
こういう遅くて、エラーも起こる関数をテーマにします。
func slowFunction(arg string) (string, error) {
fmt.Printf("SLOW FUNCTION START: %s\n", arg)
start := time.Now()
// なにか重い処理
time.Sleep(time.Duration(rand.Intn(3000)) * time.Millisecond)
// 2秒以上かかったらタイムアウト
if time.Since(start) > 2*time.Second {
fmt.Printf("SLOW FUNCTION TIMEOUT: %s, time=%v\n", arg, time.Since(start))
return "", fmt.Errorf("timeout")
}
// 重い処理が終わったら結果を返す
fmt.Printf("SLOW FUNCTION END: %s, time=%v\n", arg, time.Since(start))
return time.Since(start).String(), nil
}
とにかく並行で実行できていればいい
func main() {
args := []string{"1", "2", "3", "4", "5", "6", "7", "8", "9", "10"}
wg := sync.WaitGroup{}
for _, arg := range args {
wg.Add(1)
go func(arg string) {
defer wg.Done()
_, _ = slowFunction(arg)
}(arg)
}
wg.Wait()
fmt.Println("END")
}
すべての処理がとにかく終わればいいので、waitGroupを使います。for文の中でwaitGroupにキューを追加し、それがdoneされるのをfor分の外で待ちます。結果は意味を持たないので、アンダースコアで受け取っています。結果に対して軽い処理をしたい場合は値を受けとって、自由に加工するとよいでしょう。
goroutineのポイントは、ループで回している値(今だとargs)をgoroutine関数にバインドすることです。
エラーが起きたらその時点で終了する
channelを使います。channelは異なるgoroutine間での値の受け渡しを担うもので、メモリアクセスより安全に扱えます。
func main() {
args := []string{"1", "2", "3", "4", "5", "6", "7", "8", "9", "10"}
errChan := make(chan error) // エラーが発生した場合にエラーを送信するチャネル
doneChan := make(chan struct{}) // 全てのgoroutineが終了したことを通知するチャネル
defer close(errChan)
defer close(doneChan)
var wg sync.WaitGroup
var once sync.Once
for _, arg := range args {
wg.Add(1)
go func(arg string) {
defer wg.Done()
_, err := slowFunction(arg)
if err != nil {
once.Do(func() {
errChan <- err // エラーが発生した場合はエラーチャネルにエラーを送信
})
}
}(arg)
}
go func() {
wg.Wait()
if len(errChan) == 0 { // エラーチャネルにエラーがない場合
doneChan <- struct{}{} // 全てのgoroutineが終了したことを通知
}
}()
select {
case err := <-errChan: // エラーチャネルからエラーを受信
fmt.Printf("ERROR: %v\n", err)
return
case <-doneChan: // 全てのgoroutineが終了したことを受信
fmt.Println("ALL DONE")
}
fmt.Println("END")
}
errChanはエラーを通知するチャネルで、doneChanはすべてのgoroutineの終了を通知するチャネルです。必ずどちらかに値が入るまではこの関数は終わりません。エラーの詳細が知りたければ、errChanの型をより大きな構造体にし、複数の情報を詰めた構造体にすると良いでしょう。
channelは必ず閉じられる必要があるためdeferで先に宣言しておくのが好きです。syncのonceを使うと、errchanに一回以上値が送信されないことが保証されます。
おわりに
よく使うパターンを備忘録的に書いておきました。ジェネリクス的に一般化することもできそうですが、今のところそれが必要なシーンがありませんでした。goroutine, channelは怖いものではないよ! ということが伝われば十分です。ありがとうございました。
Discussion