Published on

Go语言中的fan-out与fan-in模式

何为fan-out, fan-in

fan-out

Multiple functions can read from the same channel until that channel is closed; this is called fan-out. This provides a way to distribute work amongst a group of workers to parallelize CPU use and I/O.

译：多个函数可以从同一个channel读取数据，直到该channel关闭；这称为扇出(fan-out)。 这提供了一种在一组worker之间分配工作以并行化CPU使用和I/O的途径。

fan-in

A function can read from multiple inputs and proceed until all are closed by multiplexing the input channels onto a single channel that’s closed when all the inputs are closed. This is called fan-in.

译：一个函数可以通过将多个输入通道（input channels）多路复用（multiplexing）到一个单一的通道上，从而从多个输入中读取数据，并持续处理直到所有输入通道都被关闭。当所有输入通道关闭时，这个单一通道也会被关闭。这称为扇入(fan-in)。

示例程序

fan-in、fan-out代码

https://github.com/apuppy/go-play/blob/main/demo/concurrency_pattern/fan.go
package concurrency_pattern

import (
	"fmt"
	"sync"
)

// gen sends the values in nums on the returned channel, then closes it.
func gen(nums ...int) <-chan int {
	out := make(chan int)
	go func() {
		for _, n := range nums {
			out <- n
		}
		close(out)
	}()
	return out
}

// sq receives values from in, squares them, and sends them on the returned
// channel, until in is closed.  Then sq closes the returned channel.
func sq(in <-chan int) <-chan int {
	out := make(chan int)
	go func() {
		for n := range in {
			out <- n * n
		}
		close(out)
	}()
	return out
}

// merge receives values from each input channel and sends them on the returned
// channel.  merge closes the returned channel after all the input values have
// been sent.
func merge(cs ...<-chan int) <-chan int {
	var wg sync.WaitGroup // HL
	out := make(chan int)

	// Start an output goroutine for each input channel in cs.  output
	// copies values from c to out until c is closed, then calls wg.Done.
	output := func(c <-chan int) {
		for n := range c {
			out <- n
		}
		wg.Done() // HL
	}
	wg.Add(len(cs)) // HL
	for _, c := range cs {
		go output(c)
	}

	// Start a goroutine to close out once all the output goroutines are
	// done.  This must start after the wg.Add call.
	go func() {
		wg.Wait() // HL
		close(out)
	}()
	return out
}

func FanOutFanIn() {
	in := gen(2, 3)

	// Distribute the sq work across two goroutines that both read from in.
	c1 := sq(in)
	c2 := sq(in)
	// ------ The code above is the fan-out part of the pattern.

	// ------ The code bellow is the fan-in part of the pattern.
	// Consume the merged output from c1 and c2.
	for n := range merge(c1, c2) {
		fmt.Println(n) // 4 then 9, or 9 then 4
	}
}

单元测试

https://github.com/apuppy/go-play/blob/main/tests/con_pattern_test.go
package tests

import (
	"github.com/apuppy/go-play/demo/concurrency_pattern"
	"testing"
)

func TestPatternFanOutFanIn(t *testing.T) {
	concurrency_pattern.FanOutFanIn()
}

运行测试

$ go test -v github.com/apuppy/go-play/tests -run TestPatternFanOutFanIn
# github.com/apuppy/go-play/tests.test
ld: warning: -no_pie is deprecated when targeting new OS versions
=== RUN   TestPatternFanOutFanIn
9
4
--- PASS: TestPatternFanOutFanIn (0.00s)
PASS
ok      github.com/apuppy/go-play/tests 1.041s

参考