Day 030 — 並行処理パターン②（Fan-in/Fan-out）

📚 背景知識（読んでから問題へ）

Day 029のWorker Poolは「1つのjobschannelを複数workerが奪い合いながら処理し、結果を1つのresultschannelに書き込む」という構成でした。実はこれ自体がすでにFan-out（分配）の一種です。今日はこの考え方を明示的に2つの部品に分解し、単独で再利用できる形にします。

• Fan-out（分配）: 1つの入力channelを、複数のgoroutine（＝複数の出力channel）に分散して処理させること。「1つの仕事の流れを、複数の実行系統に広げる」イメージです。
• Fan-in（集約）: 複数のchannelから届く値を、1つのchannelにまとめること。「複数の実行系統の結果を、1つの流れに戻す」イメージです。

Worker Poolでは「複数workerが同じresultschannelに直接書き込む」ことでFan-inを暗黙に行っていましたが、これはworkerの数だけresults<-という書き込み箇所が存在するということでもあります。Fan-out/Fan-inパターンを明示的に分離すると、次のように役割がはっきりします。

📋 コピーouts := fanOut(jobs, numWorkers, process) // 1つのjobsを複数の出力channelに分配
merged := fanIn(outs...)                   // 複数の出力channelを1つに集約

こう分離する利点は、fanOutとfanInをそれぞれ独立にテスト・再利用できることです。実務では「fanOutの段数を増やす」「fanInの前段にフィルタ処理を挟む」といった変更が、部品が分かれているだけで局所化できます。これはPhase 3後半で扱うPipelineパターン（Day 031予定）の前段階でもあります。

Fan-inの実装で最も重要な注意点は、「複数の送信元がある1つのchannelを、いつcloseするか」です。Day 029で学んだ「closeできるのは送信側、それも全送信元の完了を知っている場所だけ」という原則が、Fan-inではそのまま「sync.WaitGroupで全入力channelの読み切りを待ってから、集約先channelをcloseする」という形で登場します。

📝 問題

商品価格を複数の外部ベンダーAPI（ここではモック関数で模擬）から並行して取得する処理を、Fan-out/Fan-inパターンで実装してください。

要件

1. 以下の型を使用すること。

   📋 コピーtype Job struct {
   	ID  int
   	SKU string
   }
   
   type PriceResult struct {
   	JobID    int
   	SKU      string
   	Price    int
   	WorkerID int
   }

2. fanOut関数を実装すること。シグネチャは次の通り。

   📋 コピーfunc fanOut(jobs <-chan Job, numWorkers int, process func(workerID int, job Job) PriceResult) []<-chan PriceResult

   • jobsから受け取ったジョブをnumWorkers個のgoroutineに分配し、それぞれが自分専用の出力channelにprocessの結果を書き込む
   • 戻り値はnumWorkers個の受信専用channelのスライス
   • 各goroutineはjobsがcloseされて空になったら、自分の出力channelをcloseして終了する
3. fanIn関数を実装すること。シグネチャは次の通り。

   📋 コピーfunc fanIn(channels ...<-chan PriceResult) <-chan PriceResult

   • 可変長の入力channel群から届く値を、1つの出力channelにまとめる
   • 全ての入力channelがcloseされ、かつ中身を読み切ったら、出力channelもcloseする
   • 出力channelへの書き込みは複数goroutineから行われるため、closeは「全員の完了を検知できる1箇所」でのみ行うこと（Day 029の原則の再利用）
4. mainで、5件のジョブ（SKU: "A001"〜"A005"）を投入し、numWorkers=3でfanOut→fanInを実行し、全結果をIDでソートして表示すること。
5. runFanOutFanIn(jobs []Job, numWorkers int, process func(int, Job) PriceResult) []PriceResultという統合関数を用意し、テストしやすい形にすること。
6. テーブル駆動テストを書き、次を検証すること。
   • ジョブ件数分の結果が過不足なく返る（JobIDでソートして期待値と一致することを確認）
   • ジョブが0件のとき、パニックせず空スライスが返る
   • numWorkersがジョブ件数より多くても正しく全件処理される
7. go test -raceでデータ競合が検出されないことを確認すること。

go run . / go test -race -v ./...でそのまま動作する完全なコードとして提示すること。

🔍 ヒント（段階的開示）

📋 コピーfunc fanOut(jobs <-chan Job, numWorkers int, process func(int, Job) PriceResult) []<-chan PriceResult {
	outs := make([]<-chan PriceResult, numWorkers)
	for i := 0; i < numWorkers; i++ {
		out := make(chan PriceResult)
		outs[i] = out
		go func(workerID int, out chan<- PriceResult) {
			defer close(out)
			for job := range jobs {
				out <- process(workerID, job)
			}
		}(i, out)
	}
	return outs
}

func fanIn(channels ...<-chan PriceResult) <-chan PriceResult {
	merged := make(chan PriceResult)
	var wg sync.WaitGroup
	wg.Add(len(channels))
	for _, c := range channels {
		go func(c <-chan PriceResult) {
			defer wg.Done()
			for v := range c {
				merged <- v
			}
		}(c)
	}
	go func() {
		wg.Wait()
		close(merged)
	}()
	return merged
}

✅ 模範解答

ファイル: main.go

📋 コピーpackage main

import (
	"fmt"
	"sort"
	"sync"
)

// Job は価格照会対象の商品ジョブ。
type Job struct {
	ID  int
	SKU string
}

// PriceResult はfanOutのworkerが返す価格照会結果。
type PriceResult struct {
	JobID    int
	SKU      string
	Price    int
	WorkerID int
}

// fanOut はjobsをnumWorkers個のgoroutineに分配し、それぞれ専用の出力channelへ書き込む。
// 各goroutineはjobsがcloseされて空になったら自分の出力channelをcloseする（自分の出力の所有者は自分自身）。
func fanOut(jobs <-chan Job, numWorkers int, process func(workerID int, job Job) PriceResult) []<-chan PriceResult {
	outs := make([]<-chan PriceResult, numWorkers)
	for i := 0; i < numWorkers; i++ {
		out := make(chan PriceResult)
		outs[i] = out
		go func(workerID int, out chan<- PriceResult) {
			defer close(out)
			for job := range jobs {
				out <- process(workerID, job)
			}
		}(i, out)
	}
	return outs
}

// fanIn は複数の入力channelから届く値を1つの出力channelに集約する。
// closeは「全入力channelの読み切りを検知できる唯一の場所」であるwg.Wait()の直後でのみ行う。
func fanIn(channels ...<-chan PriceResult) <-chan PriceResult {
	merged := make(chan PriceResult)
	var wg sync.WaitGroup
	wg.Add(len(channels))
	for _, c := range channels {
		go func(c <-chan PriceResult) {
			defer wg.Done()
			for v := range c {
				merged <- v
			}
		}(c)
	}
	go func() {
		wg.Wait()
		close(merged)
	}()
	return merged
}

// runFanOutFanIn はjobsをfanOut→fanInで並行処理し、全結果をスライスで返す。
// 結果の順序は並行実行のため保証されない。
func runFanOutFanIn(jobs []Job, numWorkers int, process func(int, Job) PriceResult) []PriceResult {
	jobsCh := make(chan Job, len(jobs))
	for _, j := range jobs {
		jobsCh <- j
	}
	close(jobsCh)

	outs := fanOut(jobsCh, numWorkers, process)
	merged := fanIn(outs...)

	var results []PriceResult
	for r := range merged {
		results = append(results, r)
	}
	return results
}

// checkPrice は外部ベンダーAPIへの価格照会を模擬する（ここではSKUの文字数×100円+workerIDで疑似計算）。
func checkPrice(workerID int, job Job) PriceResult {
	price := len(job.SKU)*100 + workerID
	return PriceResult{JobID: job.ID, SKU: job.SKU, Price: price, WorkerID: workerID}
}

func main() {
	jobs := []Job{
		{ID: 1, SKU: "A001"},
		{ID: 2, SKU: "A002"},
		{ID: 3, SKU: "A003"},
		{ID: 4, SKU: "A004"},
		{ID: 5, SKU: "A005"},
	}

	results := runFanOutFanIn(jobs, 3, checkPrice)
	sort.Slice(results, func(i, j int) bool { return results[i].JobID < results[j].JobID })

	for _, r := range results {
		fmt.Printf("job=%d sku=%s price=%d (worker=%d)\n", r.JobID, r.SKU, r.Price, r.WorkerID)
	}
}

ファイル: main_test.go

📋 コピーpackage main

import (
	"sort"
	"testing"
)

func TestRunFanOutFanIn(t *testing.T) {
	tests := []struct {
		name       string
		jobs       []Job
		numWorkers int
	}{
		{
			name: "5件・3worker",
			jobs: []Job{
				{ID: 1, SKU: "A001"},
				{ID: 2, SKU: "A002"},
				{ID: 3, SKU: "A003"},
				{ID: 4, SKU: "A004"},
				{ID: 5, SKU: "A005"},
			},
			numWorkers: 3,
		},
		{
			name:       "0件",
			jobs:       nil,
			numWorkers: 3,
		},
		{
			name:       "worker数がジョブ件数より多い",
			jobs:       []Job{{ID: 1, SKU: "B001"}},
			numWorkers: 10,
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			results := runFanOutFanIn(tt.jobs, tt.numWorkers, checkPrice)

			if len(results) != len(tt.jobs) {
				t.Fatalf("len(results) = %d, want %d", len(results), len(tt.jobs))
			}

			sort.Slice(results, func(i, j int) bool { return results[i].JobID < results[j].JobID })
			wantJobs := append([]Job(nil), tt.jobs...)
			sort.Slice(wantJobs, func(i, j int) bool { return wantJobs[i].ID < wantJobs[j].ID })

			for i, r := range results {
				if r.JobID != wantJobs[i].ID {
					t.Fatalf("results[%d].JobID = %d, want %d", i, r.JobID, wantJobs[i].ID)
				}
				if r.SKU != wantJobs[i].SKU {
					t.Fatalf("results[%d].SKU = %s, want %s", i, r.SKU, wantJobs[i].SKU)
				}
			}
		})
	}
}

func TestFanIn_EmptyChannels(t *testing.T) {
	merged := fanIn() // channelを1つも渡さない場合、即座にcloseされる必要がある
	count := 0
	for range merged {
		count++
	}
	if count != 0 {
		t.Fatalf("count = %d, want 0", count)
	}
}

$ go vet ./... && go build ./...
（エラーなし）

$ go test -race -v ./...
=== RUN   TestRunFanOutFanIn
=== RUN   TestRunFanOutFanIn/5件・3worker
--- PASS: TestRunFanOutFanIn/5件・3worker (0.00s)
=== RUN   TestRunFanOutFanIn/0件
--- PASS: TestRunFanOutFanIn/0件 (0.00s)
=== RUN   TestRunFanOutFanIn/worker数がジョブ件数より多い
--- PASS: TestRunFanOutFanIn/worker数がジョブ件数より多い (0.00s)
--- PASS: TestRunFanOutFanIn (0.00s)
=== RUN   TestFanIn_EmptyChannels
--- PASS: TestFanIn_EmptyChannels (0.00s)
PASS
ok  	example.com/day030	1.021s

$ go run .
job=1 sku=A001 price=402 (worker=2)
job=2 sku=A002 price=402 (worker=2)
job=3 sku=A003 price=400 (worker=0)
job=4 sku=A004 price=400 (worker=0)
job=5 sku=A005 price=401 (worker=1)

go.mod・main.go・main_test.goを1モジュール（module example.com/day030）として配置し、go vet / go build / go test -race -v ./... / go run .で実際に検証済みです。

🪜 Step-by-Step 解説

📋 コピーgo func(workerID int, out chan<- PriceResult) {
	defer close(out)
	for job := range jobs {
		out <- process(workerID, job)
	}
}(i, out)

📋 コピーgo func() {
	wg.Wait()
	close(merged)
}()

📋 コピーouts := make([]<-chan PriceResult, numWorkers)
outs[i] = out // out は chan PriceResult 型だが、代入先が <-chan なので自動的に受信専用に変換される

📋 コピーmerged := fanIn(outs...)

💡 設計思想・なぜこう書くのか

🌐 他言語との比較

Goのchannelベースのfan-in最大の特徴は、「全部揃うまで待つ一括取得」ではなく「終わったものから順に流れてくるストリーム」として自然に扱える点です。Java/Python/Node.jsのallOf/gather/Promise.allは基本的に「全完了を待ってから配列で受け取る」設計であり、Goのように結果をfor rangeでその場から処理し始める形にするには追加の工夫（as_completed等）が必要です。

🏆 実務での使いどころ

• 複数ベンダーAPIからの並行価格照会: 今日の問題そのものが典型例です。1つの商品IDに対して複数の外部ベンダーへ並行問い合わせを行い、結果を1つのchannelに集約して「最安値を選ぶ」「全ベンダーの結果をログに残す」といった処理につなげられます
• 複数ソースのログ集約: 複数のログファイル・複数のKafkaパーティションなど、異なるソースから届くイベントを1つの処理パイプラインに合流させる場面でfan-inが使われます
• CPU負荷の高い処理をコア数で分散: 画像リサイズや暗号化処理などをfan-outで複数goroutineに分配し、結果をfan-inで集約してから後続処理（DB書き込み等）に渡す構成は頻出パターンです
• Pipelineパターンへの接続: Day 031で扱う予定のPipelineパターンは、今日のfan-out/fan-inを複数のステージに直列でつなげたものとして理解できます

⚠️ よくある誤解・ミス

🚀 次のステップ

• 発展: 今日のfanOutのprocess関数の中で、意図的にpanicを起こすジョブを1件混ぜてみてください。現在の実装ではworkerのgoroutineがpanicでクラッシュし、fanIn側は該当channelが二度とcloseされないため、結果を読み切れず永久にブロックします。recoverを使ってこの状況を防ぐにはprocessの呼び出しをどう変更すればよいか考えてみましょう
• 次回予告: Day 031 — 並行処理パターン③（Pipeline）（実装）。今日のfan-out/fan-inをステージとして複数連結し、「生成 → 変換 → フィルタ」のようなストリーム処理パイプラインを構築します

🎯 自己評価

自分の回答

気づき・メモ