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は怖いものではないよ! ということが伝われば十分です。ありがとうございました。