本章では、
並行プログラミングの基本
複数の処理を効率良く行うために、
本節では、
ゴルーチン
Goには、main()関数も、
ここではHTTPへのアクセス処理を用いて、
ゴルーチンを使わない場合
たとえば、
package main
import (
"fmt"
"log"
"net/http"
)
func main() {
urls := []string{
"http://example.com",
"http://example.net",
"http://example.org",
}
for _, url := range urls {
res, err := http.Get(url)
if err != nil {
log.Fatal(err)
}
defer res.Body.Close()
fmt.Println(url, res.Status)
}
}http.は同期処理であるため、
しかし、
ゴルーチンを使った場合
先ほどのプログラムを、
func main() {
urls := []string{
"http://example.com",
"http://example.net",
"http://example.org",
}
for _, url := range urls {
// 取得処理をゴルーチンで実行する
go func(url string) {
res, err := http.Get(url)
if err != nil {
log.Fatal(err)
}
defer res.Body.Close()
fmt.Println(url, res.Status)
}(url)
}
// main()が終わらないように待ち合わせる
time.Sleep(time.Second)
}ここではmain()が実行されたときに内部で3つのゴルーチンを起動していますが、main()は先に進んでしまうため、time.を呼んで1秒間main()を止めています
このプログラムの実行は図2のようなイメージです。各ゴルーチンが並行してリクエストを発行しているため、
sync.WaitGroup
先ほどの例ではtime.でmain()を1秒間待たせていましたが、http.を行っているすべてのゴルーチンの終了です。
起動したすべてのゴルーチンの終了を待ち合わせるにはsync.が利用できます。sync.は、Add()でカウントを増やしDone()でカウントを減らし、Wait()でカウントがゼロになるまで待ち合わせます。
func main() {
wait := new(sync.WaitGroup)
urls := []string{
"http://example.com",
"http://example.net",
"http://example.org",
}
for _, url := range urls {
// waitGroupに追加
wait.Add(1)
// 取得処理をゴルーチンで実行する
go func(url string) {
res, err := http.Get(url)
if err != nil {
log.Fatal(err)
}
defer res.Body.Close()
fmt.Println(url, res.Status)
// waitGroupから削除
wait.Done()
}(url)
}
// 待ち合わせ
wait.Wait()
}チャネル
複数のゴルーチン間でデータをやりとりしたい場合、make()関数に型を指定して生成することで、
// stringを扱うチャネルを生成
ch := make(chan string)
// チャネルにstringを書き込む
ch <- "a"
// チャネルからstringを読み出す
message := <- ch今回の場合は、main()のゴルーチンで読み出すことで、
func main() {
urls := []string{
"http://example.com",
"http://example.net",
"http://example.org",
}
statusChan := make(chan string)
for _, url := range urls {
// 取得処理をゴルーチンで実行する
go func(url string) {
res, err := http.Get(url)
if err != nil {
log.Fatal(err)
}
defer res.Body.Close()
statusChan <- res.Status
}(url)
}
for i := 0; i < len(urls); i++ {
fmt.Println(<-statusChan)
}
}ゴルーチンの中でstatusChanに値が書き込まれるまで、main()の中では値を読み出すことができません。この場合、main()内ではstatusChanの読み出しが3回完了するまで処理がブロックされるため、
これにより、
チャネルを返すパターン
先ほどはmain() 内の匿名関数でHTTPのGETを実行していましたが、getStatus()という別の関数にし、
func getStatus(urls []string) <-chan string {
// 関数でチャネルを生成
statusChan := make(chan string)
for _, url := range urls {
go func(url string) {
res, err := http.Get(url)
if err != nil {
log.Fatal(err)
}
defer res.Body.Close()
statusChan <- res.Status
}(url)
}
return statusChan // チャネルを返す
}
func main() {
urls := []string{
"http://example.com",
"http://example.net",
"http://example.org",
}
statusChan := getStatus(urls)
for i := 0; i < len(urls); i++ {
fmt.Println(<-statusChan)
}
}まず、getStatus()内で結果を渡すためのstatus Chanを生成します。次に非同期に行う処理を匿名関数にし、statusChanを返して終了し、statusChanに結果を書き込んでいきます。
main()は、main()側が非常にスッキリと記述でき、getStatus()に隠蔽
また、getStatus()はmain()がチャネルに値を書き込むことを想定していません。こうした場合は、getStatus()の戻り値を<-chan stringと読み出し専用のチャネルにすることで、main()がこのチャネルに値を書き込むことを型レベルで防ぐことができます。
このパターンはチャネルを用いる場合によく使うので、
select文を用いたイベント制御
複数のチャネルに対する読み出しや書き込みを同時に制御するためにはselect文を用います。select文はfor文と組み合わせて使う場合が多くあります。
case
複数の操作をselect文のcaseに指定しておくと、
ch1 := make(chan string)
ch2 := make(chan string)
for {
select {
case c1 := <-ch1:
// ch1からデータを読み出したときに実行される
case c2 := <-ch2:
// ch2からデータを読み出したときに実行される
case ch2 <- "c":
// ch2にデータを書き込んだときに実行される
}
}default
caseの最後にdefaultを記述すると、
ch1 := make(chan string)
ch2 := make(chan string)
for {
select {
case c1 := <-ch1:
// ch1からデータを読み出したときに実行される
case c2 := <-ch2:
// ch2からデータを読み出したときに実行される
case ch2 <- "c":
// ch2にデータを書き込んだときに実行される
default:
// caseが実行されなかった場合に実行される
}
}タイムアウト
for/
func main() {
// 1秒後に値が読み出せるチャネル
timeout := time.After(time.Second)
urls := []string{
"http://example.com",
"http://example.net",
"http://example.org",
}
statusChan := getStatus(urls)
LOOP: // ラベル名は任意
for {
select {
case status := <-statusChan:
fmt.Println(status) // 受信したデータを表示
case <-timeout:
break LOOP // このfor/selectを抜ける
}
}
}timeパケージにあるtime.関数は、
先ほどのstatusChanの読み出しを無限forループ内のselect文で受け取るようにします。ステータスを受け取った場合はそれが表示されますが、timeoutから値が読み出せると、
注意点として、
チャネルバッファ
make()でチャネルを生成するときに、
バッファなしチャネル
バッファを指定せずにmake()で生成したチャネルは、
次の場合はmain()内でチャネルの読み出す側に先に到達しますが、
func main() {
ch := make(chan string)
go func() {
time.Sleep(time.Second)
ch <- "a" // 1秒後にデータを書き込む
}()
<-ch // 1秒後にデータが書き込まれるまでブロック
}逆に次の場合は、main()内でチャネルに書き込む側に先に到達しますが、
func main() {
ch := make(chan string)
go func() {
time.Sleep(time.Second)
<-ch // 1秒後にデータを読み出す
}()
ch <- "a" // 1秒後にデータが読み出されるまでブロック
}このことを利用してバッファゼロのチャネルをゴルーチン間の同期制御に使うこともできますが、
バッファ付きチャネル
チャネルのバッファサイズはmake()の引数で指定します。たとえば次のようにバッファを3にして生成したチャネルは、
func main() {
ch := make(chan string, 3)
go func() {
time.Sleep(time.Second)
<-ch // 1秒後にデータを読み出す
}()
ch <- "a" // ブロックしない
ch <- "b" // ブロックしない
ch <- "c" // ブロックしない
ch <- "d" // 1秒後にデータが読み出されるまでブロック
}このように、
たとえば先ほどのstatus取得の例では、statusChanに値を書き込むことで終了するのですが、statusChanからデータを受け取るmain()側の処理が非常に遅かった場合、statusChanにバッファを付けることで、main()側の処理が遅くても、
func getStatus(urls []string) <-chan string {
// バッファをURLの数（3）に
statusChan := make(chan string, len(urls))
for _, url := range urls {
go func(url string) {
res, err := http.Get(url)
if err != nil {
log.Fatal(err)
}
defer res.Body.Close()
// main()の読み出しが遅くても
// 3つのゴルーチンはすぐに終わる
statusChan <- res.Status
}(url)
}
return statusChan
}ゴルーチンの同時起動数制御
getStatus()にURLが複数渡ってきた場合に、
ここではlimitというバッファ付きのチャネルを用いて、
var empty struct{} // サイズがゼロの構造体
func getStatus(urls []string) <-chan string {
statusChan := make(chan string, 3)
// バッファを5に指定して生成
limit := make(chan struct{}, 5)
go func() {
for _, url := range urls {
select {
case limit <- empty:
// limitに書き込みが可能な場合は取得処理を実施
go func(url string) {
// このゴルーチンは同時に5つしか起動しない
res, err := http.Get(url)
if err != nil {
log.Fatal(err)
}
statusChan <- res.Status
// 終わったら1つ読み出して空きを作る
<-limit
}(url)
}
}
}()
return statusChan
}特集のまとめ
本特集では、