はじめに

引き続き「Go言語による並行処理」を読んでいます。
前回の記事ではgoroutineにフォーカスして勉強しました。

次にchannelとWaitGroupについて学んでいきます。

channel

channelとは

goroutine間のデータをやりとりするための仲介業者のような存在です。

前回の記事にも書きましたが、Go言語ではMessage-passing communicationによって並行処理をする事を推奨しています。
その通信の役割を担うのがchannelです。

つまりGoroutineAとGoroutineBのやりとりはchannelを用いて行われます。
※A,Bはchannelを参照しているだけなのでお互いのことを知らない。

簡単にですが、このように使います。

func main()  {
    stream := make(chan interface{})
    fmt.Println("main start")
    go func() {
        stream <- "hello"
    }()
    fmt.Println(<- stream)
    fmt.Println("main end")
}

// main start
// hello
// main end

<- streamはstreamに何か値が入るまで次の処理に行かずにブロックします。
※複数streamにデータが入っても、一回値が入ったらその時点でブロックが解除される。

channelの挙動

初期化

それでは詳しくみていきます。

・通常の初期化
stream := make(chan interface{})
これはinterface型のchannelを初期化し生成しています。
型を変える事でその型しか受け取れないchnnelを生成できます。

・バッファサイズを指定した初期化
stream := make(chan int{}, 4)
channelがためこめる最大量(ここでは4)を指定することができます。

<-を付ける位置によって送信受信専用のchannelを生成することができます。

・送信専用
stream := make(chan <- interface{})

・受信専用
stream := make(<- chan interface{})

closeの使い方

以下のようにcloseするだけでchannelを終了することが出来ます。

stream := make(chan interface{})
close(stream)

・close後の書き込みは？

func main()  {
    stream := make(chan interface{})
    fmt.Println("channel終了")
    close(stream)
    go func() {
        fmt.Println("channelに送信")
        stream <- "hello"
    }()
    time.Sleep(time.Second * 4)
}
// channel終了
// panic: send on closed channel

閉じたchannelに対しての書き込みはダメなようです。

・close後の読み込みは？

func main()  {
    stream := make(chan interface{})
    fmt.Println("channel終了")
    go func() {
        fmt.Println("channelに送信")
        stream <- "hello"
    }()
    time.Sleep(time.Second * 2)
    close(stream)
    num, isOpen := <- stream
    fmt.Println(num, isOpen)
}
// channel終了
// channelに送信
// <nil> false

helloを書き込んでいるにも関わらず閉じられた場合は、初期値が返却されます。
また、第二戻り値にbool値が返却されるのでcloseされたchannelか判断できます。

for range

rangeの引数にchannelをとり、channelが閉じた時にループを終了する機能です。

func main()  {
    stream := make(chan interface{})
    go func() {
        defer fmt.Println("close")
        defer close(stream)
        for i := 0; i < 3; i++ {
            stream <- i
        }
    }()

    for num := range stream {
        time.Sleep(time.Second * 1)
        fmt.Println("range開始:",num)
    }
}
// range開始: 0
// range開始: 1
// close
// range開始: 2

closeが呼び出し回数3回に対して1回しか呼ばれていません。
つまりcloseされた時点でrangeのループ処理が終了していることがわかります。

closeされた時に実行

func main() {

    done := make(chan interface{})

    for i := 0; i < 5; i++ {
        go func() {
            <-done
            fmt.Println("end", i)
        }()
    }

    close(done)
    time.Sleep(time.Second * 2)
}
// end 2
// end 3
// end 4
// end 1
// end 0

closeが呼ばれる事で、<-doneに初期値が渡ります。
結果としてcloseされた時にループ処理が実行されます。

WaitGroup

goroutineが処理を終えるまで待ってくれる機能です。
例えば、以下のようなコードがあった場合何も表示されずに終わります。
goroutineが実行される前にmain groutineが終了しているためです。

func main() {
    for i := 0; i < 3; i++ {
        go func() {
            fmt.Println("test")
        }()
    }
}

WaitGroupを使うと、期待する動作になります。
Addする事で、引数に指定したgoroutineの数をカウントします。
Doneする事で、Addされた数からgoroutineの数を-1します。
WaitはAddされたgoroutineの数が0になるまで待機します。

func main() {
    wg := &sync.WaitGroup{}
    for i := 0; i < 3; i++ {
        wg.Add(1)
        go func() {
            defer wg.Done()
            fmt.Println("test")
        }()
    }
    wg.Wait()
}

//test
//test
//test

最後に

今回はchannelとwaitGroupの使い分けについて学びました。

なんとなくgoroutineの使い方や雰囲気はわかりましたが、
まだ実践できるレベルには至っていません。

次はもう少し現場のコードに近い実践的なコードを学んでいきます。

Go言語による並行処理備忘録 ~channel、WaitGroup編~