概要

並行処理のパターンであるfan-in、fan-outをGoで実装する。

fan-in/fan-outとは

fan-inは、複数の入力を1つにまとめる処理で、fan-outは、1つの入力を複数に分ける処理である。

fan-inはデータを集約させ、fan-outはデータを分散させる。

Goではchannelとgoroutineを使って実現することができる。

実装

ソースコードはgithubにも置いてある。

package main

import (
	"fmt"
	"math/rand"
	"sync"
	"time"
)

func producer(id int, out chan<- int) {
	for i := 0; i < 5; i++ {
		value := rand.Intn(100)
		fmt.Printf("Producer %d: Sending %d\n", id, value)
		out <- value
		time.Sleep(time.Millisecond * time.Duration(rand.Intn(1000)))
	}
	close(out)
}

func fanIn(inputs []<-chan int, out chan<- int) {
	var wg sync.WaitGroup
	wg.Add(len(inputs))

	for _, input := range inputs {
		go func(ch <-chan int) {
			for value := range ch {
				out <- value
			}
			wg.Done()
		}(input)
	}

	go func() {
		wg.Wait()
		close(out)
	}()
}

func main() {
	rand.Seed(time.Now().UnixNano())

	// Fan-Out
	numProducers := 3
	inputs := make([]chan int, numProducers)
	for i := 0; i < numProducers; i++ {
		inputs[i] = make(chan int)
		go producer(i+1, inputs[i])
	}

	// Convert channels to <-chan int
	inputChans := make([]<-chan int, numProducers)
	for i := 0; i < numProducers; i++ {
		inputChans[i] = inputs[i]
	}

	// Fan-In
	result := make(chan int)
	go fanIn(inputChans, result)

	// Consume the merged values
	for value := range result {
		fmt.Printf("Consumer: Received %d\n", value)
	}

	fmt.Println("All done!")
}

fan-outの処理でデータを分散して、fan-inの処理でデータを集約している。

所感

並行処理は自身がないので勉強しないといけない。。。

参考

• ludwig125.hatenablog.com - go言語のpipeline、fan-in、fan-out
• devlights.hatenablog.com - Goメモ-73 (fan-in パターンでデータを集約するチャネル, FanIn)
• devlights.hatenablog.com - Goメモ-79 (指定されたワーカー数でファンアウトさせる関数, FanOut)
• tech-up.hatenablog.com - fan-out、fan-inパターン【Go】
• selfnote.work - [Golang]アルゴリズムにチャレンジ ~ChannelでFan In/Fan Outを実装しよう!~
• go.dev - Go Concurrency Patterns: Pipelines and cancellation
• kapoorrahul.medium.com - Golang Fan-In Fan-Out Concurrency Pattern
• mariocarrion.com - LEARNING GO: FAN-IN/FAN-OUT CONCURRENCY PATTERN
• www.golinuxcloud.com - Go Fan Out Fan In Concurrency Pattern Explained