[Golang] チャネル

概要

今回のブログポストではGolangでチャネル(Channel)を使ってゴルーチン間メッセージをやり取りする方法について説明します。このブログで紹介するソースコードは下記のリンクで確認できます。

GitHub: https://github.com/dev-yakuza/study-golang/tree/main/channel

ゴルーチンに関して詳しく知りたい方は以前のブログポストを参考してください。

[Golang] ゴルーチン

チャネル

チャネルはゴルーチン間メッセージをやり取りするためのメッセージキュー(Thread-safe queue)です。

Golangでは次のようにmake()でチャネルインスタンスを生成することができます。

// var VARIABLE_NAME chan MESSAGE_TYPE = make(chan MESSAGE_TYPE)
var messages chan string = make(chan string)

チャネルを生成したら<-演算子を使ってチャネルにデータを渡すことができます。

// CHANNEL <- DATA
messages <- "This is a message"

逆に<-演算子を使ってチャネルからデータを取得することもできます。

// var VARIABLE <- CHANNEL
var msg string = <- messages

これを確認するため、main.goファイルを生成して次のように修正します。

package main

import (
  "fmt"
  "sync"
  "time"
)

func main() {
  var wg sync.WaitGroup
  ch := make(chan int)

  wg.Add(1)
  go square(&wg, ch)
  time.Sleep(time.Second)

  ch <- 3
  wg.Wait()
}

func square(wg *sync.WaitGroup, ch chan int) {
  fmt.Println("Square start")
  n := <-ch
  fmt.Println("Square:", n*n)
  wg.Done()
  fmt.Println("Square end")
}

これを実行すると次のような結果が表示されます。

# go run main.go
Square start
Square: 9
Square end

ソースコードをもっと詳しく見ると、まず、WaitGroupを使って1つのサブゴルーチンが終了されるまでメインゴルーチンを待機するようにしました。

func main() {
  var wg sync.WaitGroup
  ...
  wg.Add(1)
  ...
  wg.Wait()
}

そして、make関数を使ってチャネルを生成して、goキーワードを使ってsquare関数をサブゴルーチンで実行しました。この時、WaitGroupとチャネルを一緒に渡しました。

func main() {
  ...
  ch := make(chan int)
  ...
  go square(&wg, ch)
  ...
}

square関数を見るとチャネルで値を取得して計算をして画面に表示した後、WaitGroupのDoneを使って、ゴルーチンが終了されたことを知らせます。

func square(wg *sync.WaitGroup, ch chan int) {
  ...
  n := <-ch
  ...
  wg.Done()
  ...
}

n := <-chコードが実行されると、チャネルからデータを取得します。しかし、現在チャネルにデータがないので、データを取得することができず、ゴルーチンは待機されることになります。

この時、メインゴルーチンでch <- 3でデータを送ると、サブゴルーチンはチャネルからデータを取得することができるので、サブゴルーチンが実行され、wg.Done()が実行されるので、メインゴルーチンはサブゴルーチンを待つ必要がないので、プログラムが終了されます。

チャネルのサイズ

チャネルの基本サイズは0です。従って、次のようにプログラムを作成すると、ゴルーチンはチャネルにあるデータを取得されるまで待機されます。

package main

import (
  "fmt"
  "time"
)

func main() {
  ch := make(chan int)

  go square()
  ch <- 9
  fmt.Println("Never Print")
}

func square() {
  for {
    time.Sleep(2 * time.Second)
    fmt.Println("sleep")
  }
}

チャネルのサイズが0なので、ゴルーチンはチャネルにデータを追加することがでません。データを取得するコードがあるば、すぐ渡すことができるので、問題ありませんが、データを取得するコードを追加しないと、ゴルーチンはデータを追加することができないので、ずっと待機されます。

今度は次のようにチャネルのサイズを指定して、チャネルを生成して見ましょう。

package main

import (
  "fmt"
  "time"
)

func main() {
  // ch := make(chan int)
  ch := make(chan int, 1)

  go square()
  ch <- 9
  fmt.Println("Never Print")
}

func square() {
  for {
    time.Sleep(2 * time.Second)
    fmt.Println("sleep")
  }
}

チャネルにサイズが指定されると、そのサイズだけのデータを追加することができます。従ってゴルーチンはチャネルにデータを取得するコードがなくても、チャネルにデータを保存することができるので、チャネルにデータを保存して、続けて進むことができます。

チャネルでデータ待機

次のようにチャネルを使うと、データが追加されるまでゴルーチンを待機させることができます。

for n := range ch {
  ...
}

これを確認するため、main.goファイルを次のように修正します。

package main

import (
  "fmt"
  "sync"
  "time"
)

func main() {
  var wg sync.WaitGroup
  ch := make(chan int)

  wg.Add(1)
  go square(&wg, ch)

  for i := 0; i < 10; i++ {
    ch <- i * 2
  }

  wg.Wait()
}

func square(wg *sync.WaitGroup, ch chan int) {
  for n := range ch {
    fmt.Println("Square:", n*n)
    time.Sleep(time.Second)
  }
  fmt.Println("Done")
  wg.Done()
}

これを実行すると次のような結果が表示されます。

# go run main.go
Square: 100
Square: 144
Square: 196
Square: 256
Square: 324
fatal error: all goroutines are asleep - deadlock!

メインゴルーチンではチャネルにデータを追加します。しかし、チャネルのサイズが0なので、チャネルからデータを取得するコードがないとずっと待機されることになります。この時、サブゴルーチンで実行されたsquare関数がチャネルからデータがある場合データを取得するコードを持っていますので、データが取得されます。

だから、メイン関数ではデータを連続的に10回追加したいけど、チャネルのサイズが0なので待機されます。この時、square関数にデータがある場合、データを取得するコードがあるので、メイン関数がデータを1回入れて、square関数がそのデータを取得して表示することになります。その後、またメイン関数はチャネルにデータを入れて、square関すはまたこれを取得します。

これを10回繰り返したら、メイン関数はチャネルにデータを追加しません。しかし、square関数はチャネルにデータを続けて取得するようにしてるので、サブゴルーチンが終了されないことになります。そのせいで、どんな動作もしたいことになるのでデッドロック(Deadlock)が発生することが確認できます。

チャネルのクローズ

このようにゴルーチンでチャネルのデータをずっと待つコードを使う場合、ゴルーチンが無限待機するゾンビゴルーチンまたはゴルーチンリーク(Leak)が発生されます。

これを防ぐためには使ったチャネルをclose関数を使ってクローズする必要があります。

これを確認するためmain.goファイルを次のように修正します。

package main

import (
  "fmt"
  "sync"
  "time"
)

func main() {
  var wg sync.WaitGroup
  ch := make(chan int)

  wg.Add(1)
  go square(&wg, ch)

  for i := 0; i < 10; i++ {
    ch <- i * 2
  }
  close(ch)

  wg.Wait()
}

func square(wg *sync.WaitGroup, ch chan int) {
  for n := range ch {
    fmt.Println("Square:", n*n)
    time.Sleep(time.Second)
  }
  fmt.Println("Done")
  wg.Done()
}

これを実行すると次のような結果が表示されます。

# go run main.go
Square: 100
Square: 144
Square: 196
Square: 256
Square: 324
Done

以前とは違ってデッドロックが発生しないことが確認できます。メイン関数でチャネルに全てのメッセージを追加した後、close(ch)を使ってチャネルをクローズします。そしたらsquare関数はもうチャネルからデータを取得するため待機する必要がないのでサブゴルーチンが正常に終了されます。従って、以前とは違ってデッドロックが発生しません。

select文

次のようにselect文を使うと複数のチャネルを同時に待ってデータを取得することができます。

select {
  case n := <- ch1:
    ...
  case n := <- ch2:
    ...
  case ...
}

これを確認するためmain.goファイルを次のように修正します。

package main

import (
  "fmt"
  "sync"
  "time"
)

func main() {
  var wg sync.WaitGroup
  ch := make(chan int)

  wg.Add(1)
  go square(&wg, ch)

  for i := 0; i < 10; i++ {
    ch <- i * 2
  }
  close(ch)

  wg.Wait()
}

func square(wg *sync.WaitGroup, ch chan int) {
  tick := time.Tick(time.Second)
  terminate := time.After(10 * time.Second)

  for {
    select {
    case <-tick:
      fmt.Println("Tick")
    case <-terminate:
      fmt.Println("Terminate")
      wg.Done()
      return
    case n := <-ch:
      fmt.Println("Square:", n*n)
      time.Sleep(time.Second)
    }
  }
}

これを実行すると次のような結果が表示されます。

# go run main.go
Square: 0
Square: 4
Square: 16
Tick
Square: 36
Square: 64
Tick
Square: 100
Tick
Square: 144
Tick
Square: 196
Tick
Square: 256
Square: 324
Square: 0
Terminate

timeパッケージのTick()は一定間隔でシグナルを送信するチャンネルをリターンします。また、After()は一定時間を待機して一回シグナルを送信するチャネルをリターンします。ここでselect文を使ってsquare関数が複数のチャネルのデータを待って動作するように実装しました。

一方向チャネル

Golangでは次のように一方向チャネルを生成することができます。

// データを入れることしかできないチャネル
chan <- string
// データを取得することしかできないチャネル
<- chan string

これを確認して次のようにmain.goファイルを修正します。

package main

import (
  "fmt"
  "sync"
  "time"
)

func main() {
  var wg sync.WaitGroup
  ch := make(chan int)

  wg.Add(1)
  go square(&wg, ch)

  for i := 0; i < 10; i++ {
    ch <- i * 2
  }
  close(ch)

  wg.Wait()
}

func square(wg *sync.WaitGroup, ch <-chan int) {
  for n := range ch {
    fmt.Println("Square:", n*n)
    time.Sleep(time.Second)
  }
  fmt.Println("Done")
  wg.Done()
}

これを実行すると次のような結果が表示されます。

# go run main.go
Square: 0
Square: 4
Square: 16
Square: 36
Square: 64
Square: 100
Square: 144
Square: 196
Square: 256
Square: 324
Done

一方向チャネルはチャネルの役割を明確するために使います。例題ではsquare関数のチャネルは単純にデータを取得することしかしません。この時一方向チャネルを使って明確にこの役割を指定することができます。そしたら、間違ってsquare関数でチャネルにデータを追加するコードを作成してもコンパイルエラーが発生するので安全にチャネルを使うことができます。