goroutines の基本的内容に関して、調べた点をメモしておきます。
環境
- go 1.19.1
goroutines
goroutines とは
goroutines とは、Go が提供するマルチスレッドにより並行処理を実現する仕組みです。しかしながら、Go は OS の Kernel Thread を直接操作するのではなく、 Kernel Thread のリソースを効率的に管理する仕組みを持っており、その仕組みを介して並行処理を実現しています。
They're called goroutines because the existing terms—threads, coroutines, processes, and so on—convey inaccurate connotations. A goroutine has a simple model: it is a function executing concurrently with other goroutines in the same address space. It is lightweight, costing little more than the allocation of stack space. And the stacks start small, so they are cheap, and grow by allocating (and freeing) heap storage as required.
Goroutines are multiplexed onto multiple OS threads so if one should block, such as while waiting for I/O, others continue to run. Their design hides many of the complexities of thread creation and management.
goroutines の仕組みについては、ネット上で解説されている記事が多数あり幾つか読んでみましたが、なるほど、難しいですね。。。
とりあえず goroutines の内部動作については「へー、すごいなー」くらいの認識にしておき、goroutines を利用する具体的コードを書いてみます。
goroutines を利用するコード
goroutines としたい関数の前に go キーワードをつけて実行すると、その関数が goroutines として動作します。以下は、実行する関数を無名関数として記述している例です。
package main import ( "fmt" "time" ) func main() { fmt.Println("start") // "goroutine A" と出力する関数 go func() { for i := 0; i < 3; i++ { time.Sleep(time.Millisecond) fmt.Printf("goroutine A: %d\n", i) } }() // "goroutine B" と出力する関数 go func() { for i := 0; i < 3; i++ { time.Sleep(time.Millisecond) fmt.Printf("goroutine B: %d\n", i) } }() time.Sleep(time.Second) // 1秒待たないと goroutine の処理が終了する前に main 関数が終了してしまうため待っているだけ fmt.Println("end") } // outputs // start // goroutine B: 0 // goroutine A: 0 // goroutine A: 1 // goroutine B: 1 // goroutine A: 2 // goroutine B: 2 // end
channels
channels とは、goroutines 間で値をやり取りするための仕組みです。
A channel provides a mechanism for concurrently executing functions to communicate by sending and receiving values of a specified element type.
※
a mechanism for concurrently executing functionsとは goroutines の意。
Channels Type - The Go Programming Language Specification
goroutines を利用時に値をやり取りする場合、共有メモリ空間を利用するのではなく、この channels を利用するのが、Go のお作法らしいです。
基本的な channel の使い方
make関数にて、新しい channel を初期化します<-により、channel に対する値の送受信を指定します
package main import ( "fmt" "time" ) // channel へ値を送信する関数 func snd(ch chan<- string) { ch <- "Hello." } // channel から値を受信する関数 func rev(ch <-chan string) { msg := <-ch fmt.Println(msg) } func main() { ch := make(chan string) // string 型の channel を作成 go snd(ch) go rev(ch) time.Sleep(time.Second) } // outputs // Hello.
channel はバッファを持てる
make による channel の初期化時、第２引数に数値を入力すると、その数だけのバッファを持った channel が作成されます。何も数値を指定しない場合、バッファ数が1の channel が作成されます。
package main import ( "fmt" "time" ) func snd(ch chan<- string) { ch <- "1st msg." ch <- "2nd msg." } func rev(ch <-chan string) { msg1 := <-ch fmt.Println(msg1) msg2 := <-ch fmt.Println(msg2) } func main() { ch := make(chan string, 2) // バッファ数が2の channel を作成 go snd(ch) go rev(ch) time.Sleep(time.Second) } // outputs // 1st msg. // 2nd msg.
channel のクローズ
channel をクローズすることで、その channel に新しい送信がないことを、受信側へ明示的に伝えることができます。
package main import ( "fmt" "time" ) func snd(ch chan<- string) { defer close(ch) // 関数終了時に channel をクローズする ch <- "1st msg." ch <- "2nd msg." } func rev(ch <-chan string) { // `for .. range` で channel をイテレーションできる for msg := range ch { fmt.Println(msg) } // 上記は以下と同じ意 // channel から値を受信時、channel のクローズ状態を取得することができ、非クローズ時は True が返る // for { // msg, ok := <-ch // if ok { // fmt.Println(msg) // } // } } func main() { ch := make(chan string, 2) go snd(ch) go rev(ch) time.Sleep(time.Second) } // outputs // 1st msg. // 2nd msg.
以下とのこと。
注意: 送り手のチャネルだけをcloseしてください。受け手はcloseしてはいけません。 もしcloseしたチャネルへ送信すると、パニック( panic )します。
もう一つ注意: チャネルは、ファイルとは異なり、通常は、closeする必要はありません。 closeするのは、これ以上値が来ないことを受け手が知る必要があるときにだけです。 例えば、 range ループを終了するという場合です。
Range and Close - A Tour of Go
select を利用した channel（メッセージの受信）
channel からメッセージを受信時、select を利用することでメッセージの有無により処理を分岐できます。
package main import ( "fmt" "time" ) func snd1(ch1 chan<- string) { ch1 <- "message ch1." } func snd2(ch2 chan<- string) { ch2 <- "message ch2." } func rev(ch1, ch2 <-chan string) { time.Sleep(time.Second) // channel 受信時の select 文 // 該当する channel 内のメッセージの有無により case 文が実行される for { select { // ch1 にメッセージがある場合 case msg := <-ch1: fmt.Printf("Received 1: %s\n", msg) // ch2 にメッセージがある場合 case msg := <-ch2: fmt.Printf("Received 2: %s\n", msg) // どの case 文にも該当しない場合 default: fmt.Println("end") return } } } func main() { ch1 := make(chan string) ch2 := make(chan string) go snd1(ch1) go snd2(ch2) go rev(ch1, ch2) time.Sleep(time.Second) } // outputs // Received 1: message ch1. // Received 2: message ch2. // end
select を利用した channel（メッセージの送信）
channel へのメッセージを送信時にも select を利用できます。こちらは送信先 channel の利用可否により、処理を分岐できます。
package main import ( "fmt" "time" ) func snd1(ch1 chan<- string) { ch1 <- "message ch1." } func snd2(ch2 chan<- string) { // channel 送信時の select 文 // 該当する channel の送信準備ができているか否かにより case 文が実行される select { case ch2 <- "message ch2.": fmt.Println("Send message: ch2.") default: fmt.Println("No send message: ch2.") } } func rev(ch1, ch2 <-chan string) { msg1 := <-ch1 fmt.Printf("Received 1: %s\n", msg1) // channel ch2 の受信を準備しない // msg2 := <-ch2 // fmt.Printf("Received 2: %s\n", msg2) } func main() { ch1 := make(chan string) ch2 := make(chan string) go snd1(ch1) go snd2(ch2) go rev(ch1, ch2) time.Sleep(time.Second) } // outputs // Received 1: message ch1. // No send message: ch2.
共有メモリ部分の排他制御
goroutines 間で channel を用いず、共有メモリ上の変数で値のやり取りをする場合、標準パッケージ内にある sync.Mutex を利用してメモリの排他制御ができるようです。
package main import ( "fmt" "sync" "time" ) func main() { var mu sync.Mutex x, y := 0, 0 // 排他制御しないで、goroutines を実行する for i := 0; i < 1000000; i++ { go func() { x++ }() } // 排他制御して、goroutines を実行する for i := 0; i < 1000000; i++ { go func() { mu.Lock() // ロック取得 defer mu.Unlock() // ロック開放 y++ }() } time.Sleep(time.Second) fmt.Println("x:", x) fmt.Println("y:", y) } // outputs // x: 972513 ※排他制御していないため、インクリメント中に数値がズレてしまう // y: 1000000
