バッファあり Channel とバッファなし Channel

CoddyのGOジャーニー「オブジェクト指向プログラミング」セクションの一部。レッスン 61/107。

これまで使用してきたチャネルはunbufferedです。値を保持するcapacityがありません。送信操作は別のgoroutineが受信するまでブロックし、その逆も同様です。Buffered channelsは内部ストレージを追加し、即時の受信側がいなくても送信を完了できるようにします。

make に capacity を渡して、バッファ付き channel を作成します。

// アンバッファード - 容量 0
unbuffered := make(chan int)

// バッファード - 容量 3
buffered := make(chan int, 3)

buffered channel では、送信は buffer が full のときだけ block し、受信は buffer が empty のときだけ block します。

ch := make(chan string, 2)

ch <- "first"   // ブロックしない - バッファに空きがある
ch <- "second"  // ブロックしない - バッファに空きがある
// ch <- "third" はブロックする - バッファがいっぱい

fmt.Println(<-ch)  // "first"
fmt.Println(<-ch)  // "second"

バッファに現在いくつの項目があるかを確認するにはlen(ch)を使用し、合計容量にはcap(ch)を使用します。

それぞれの使用時期: バッファなしチャネルは緊密な同期を提供します。送信者は受信者が値を受け取ったことを把握できます。Buffered チャネルは送信者と受信者のタイミングを切り離すため、生産者と消費者の処理速度が異なる場合に役立ちます。ただし、Buffered チャネルは同期に関するバグを見えにくくする可能性があるため、バッファリングを行う明確な理由がない限り、バッファなしチャネルを優先してください。

チャレンジ

簡単

buffered と unbuffered channel の違いを示す batch processor を構築しましょう。生産者が processor に items を送信し、buffered channel を使用してそれらのタイミングを分離し、生産者が先行して処理できるシステムを作成します。

コードを2つのファイルに分けて整理します。

processor.go: channel を使用した batch processing ロジックを定義します。
ID（int）フィールドと Value（string）フィールドを持つ Item struct を作成します。

2つの関数を Implement します。
- Producer(items []Item, out chan Item) - 各 item を output channel に送信します。各 item の送信後、Produced item [ID] を出力します。すべての items の送信後、len(out) を使用して buffer で待機している items の現在の number を、Buffer has [count] items の形式で出力します。その後、channel を close します。
- Consumer(in chan Item) []string - input channel から items を受信し、formatted string の slice に収集します。受信した各 item について、string は Consumed: [ID] - [Value] とします。channel が close されたら slice を返します。
main.go: buffered channel を設定し、生産者と Consumer を調整します。
buffer capacity、次に number of items、続いて各 item の ID と value を読み取ります。指定された capacity で buffered channel を作成します。まず Producer を（goroutine としてではなく）実行して buffer を fill し、その後 Consumer を実行してすべての items を処理します。各 consumed result を別々の line に出力します。

次の input が提供されます。

line 1: Buffer capacity（integer）
line 2: Number of items（integer）
following lines: 各 item について2つの line - item ID（integer）、次にその value（string）

たとえば、次の入力が与えられた場合：

3
3
1
apple
2
banana
3
cherry

output は次のようになります。

Produced item 1
Produced item 2
Produced item 3
Buffer has 3 items
Consumed: 1 - apple
Consumed: 2 - banana
Consumed: 3 - cherry

capacity が3の buffer では、Producer が blocking されずに3つすべての items を送信でき、Consumer が開始する前に buffer に3つの items が待機していることがわかります。buffer が items の number より小さい場合、Producer は空き space を待つために block します。

自分で試してみよう

package main

import (
	"bufio"
	"fmt"
	"os"
	"strconv"
	"strings"
)

func main() {
	reader := bufio.NewReader(os.Stdin)

	// バッファ容量を読み取る
	line, _ := reader.ReadString('\n')
	bufferCapacity, _ := strconv.Atoi(strings.TrimSpace(line))

	// アイテム数を読み取る
	line, _ = reader.ReadString('\n')
	numItems, _ := strconv.Atoi(strings.TrimSpace(line))

	// 各アイテムのIDと値を読み取る
	items := make([]Item, numItems)
	for i := 0; i < numItems; i++ {
		line, _ = reader.ReadString('\n')
		id, _ := strconv.Atoi(strings.TrimSpace(line))
		line, _ = reader.ReadString('\n')
		value := strings.TrimSpace(line)
		items[i] = Item{ID: id, Value: value}
	}

	// TODO: 指定された容量のバッファ付きチャネルを作成する

	// TODO: Producerを（goroutineとしてではなく）実行してバッファを埋める

	// TODO: Consumerを実行してすべてのアイテムを処理する

	// TODO: 消費された各結果を別々の行に出力する
}