Goことはじめ(並列処理)

http://gihyo.jp/dev/feature/01/go_4beginners/0005?page=1
より学んだことメモ。

package main

import (
	"fmt"
	"log"
	"net/http"
	"sync"
)

func main() {
	wait := new(sync.WaitGroup)

	urls := []string{
		"https://www.example.com",
		"https://www.example.co.jp",
		"https://www.example.com/",
		"https://www.example.com/jp/",
	}

	for _, url := range urls {
		// waitGroup に追加(カウントを増やす)
		wait.Add(1)

		// 取得処理をゴルーチン(軽量スレッド)で実行
		// goと書くことで、並列に処理を実行できる
		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)
	}

	// カウントが0になるまで待つ
	wait.Wait()
}

チャネルを使ってゴルーチン間でやりとりする

package main

import (
	"fmt"
	"log"
	"net/http"
)

func main() {
	urls := []string{
		"https://www.example.com",
		"https://www.example.co.jp",
		"https://www.example.com/",
		"https://www.example.com/jp/",
	}

	// 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)
	}

	for i := 0; i < len(urls); i++ {
		// チャネルからstringを読み出す
		// 書き込みがされるまで待つ
		fmt.Println(<-statusChan)
	}
}

チャネルを返す関数に切り出す版

package main

import (
	"fmt"
	"log"
	"net/http"
)

//
// 関数でチャネルを作成
//
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{
		"https://www.example.com",
		"https://www.example.co.jp",
		"https://www.example.com/",
		"https://www.example.com/jp/",
	}
	statusChan := getStatus(urls)

	for i := 0; i < len(urls); i++ {
		fmt.Println(<-statusChan)
	}
}

select文を用いたイベント制御

package main

import (
	"fmt"
	"log"
	"net/http"
	"time"
)

//
// 関数でチャネルを作成
//
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
}

//
// select文を用いたイベント制御
//
func selectFunc() {
	// 1秒後に値が読み出せるチャネル
	timeout := time.After(time.Second)
	urls := []string{
		"https://www.example.com",
		"https://www.example.co.jp",
		"https://www.example.com/",
		"https://www.example.com/jp/",
	}
	statusChan := getStatus(urls)

LOOP: // 任意のラベル名を指定
	for {
		// selectのcaseに指定すると
		// いずれかのcaseの操作が行われたときに
		// 該当する処理が実行される
		// caseが実行されるまで後の処理は待つ
		select {
		case status := <-statusChan:
			// 受信データを表示
			fmt.Println(status)
		case <- timeout:
			// このfor/selectを抜ける
			// breakだけだとselectだけ抜けるためラベルを指定
			break LOOP
		}
	}
}


func main() {
	selectFunc()
}

同時期同数制御

チャネルバッファを使って、ゴルーチンの同時期同数を制御する。

package main

import (
	"fmt"
	"log"
	"net/http"
)

// サイズ0の構造体
// 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
}

func main() {
	urls := []string{
		"https://example.com",
		"https://example.com",
		"https://example.com",
		"https://example.com",
		"https://example.com",
		"https://example.com",
	}
	statusChan := getStatus(urls)

	for i := 0; i < len(urls); i++ {
		fmt.Println(<-statusChan)
	}
}