fanout fanin封装

### Fanout 和 Fanin 的概念 Fanout 是指将数据分发到多个通道或目标的过程，而 Fanin 则是指从多个通道接收数据并将其聚合到单个通道中的过程。这两种操作通常用于并发编程场景中，特别是在 Go 语言的 Goroutine 中非常常见。 以下是基于 Go 语言实现 Fanout 和 Fanin 的封装方法及其示例代码： --- ### 实现 Fanout 和 Fanin 的封装 #### 1. **Fanout 的实现** Fanout 可以通过创建多个 Goroutines 并向它们发送相同的数据流来完成。下面是一个简单的 Fanout 封装函数，其中输入数据被复制到多个输出通道上[^4]。 ```go package main import ( "fmt" ) // Fanout 函数定义 func fanout(input <-chan int, outputs []chan<- int) { for value := range input { for _, output := range outputs { output <- value } } closeAll(outputs) } // 关闭所有的输出通道 func closeAll(channels []chan<- int) { for _, ch := range channels { close(ch) } } ``` 上述代码展示了如何将一个输入通道 `input` 复制到多个输出通道 `outputs` 上。每次接收到新的值时，都会将其广播给所有连接的输出通道。 --- #### 2. **Fanin 的实现** Fanin 负责从多个输入通道读取数据并将这些数据合并成单一的输出通道。以下是一个 Fanin 的简单实现[^5]: ```go // Fanin 函数定义 func fanin(inputs ...<-chan int) chan int { var wg sync.WaitGroup out := make(chan int) output := func(c <-chan int) { defer wg.Done() for n := range c { out <- n } } wg.Add(len(inputs)) for _, in := range inputs { go output(in) } go func() { wg.Wait() close(out) }() return out } ``` 在这个例子中，`fanin` 接收任意数量的输入通道，并启动一个新的 Goroutine 来监听每个输入通道上的数据。当某个输入通道关闭时，对应的 Goroutine 自动退出。最终，在所有输入通道都关闭之后，整个 Fanin 过程也会结束。 --- ### 完整示例代码 以下是一份完整的示例代码，演示了如何组合使用 Fanout 和 Fanin: ```go package main import ( "fmt" "sync" ) // Fanout 函数定义 func fanout(input <-chan int, outputs []chan<- int) { for value := range input { for _, output := range outputs { output <- value } } closeAll(outputs) } // 关闭所有的输出通道 func closeAll(channels []chan<- int) { for _, ch := range channels { close(ch) } } // Fanin 函数定义 func fanin(inputs ...<-chan int) chan int { var wg sync.WaitGroup out := make(chan int) output := func(c <-chan int) { defer wg.Done() for n := range c { out <- n } } wg.Add(len(inputs)) for _, in := range inputs { go output(in) } go func() { wg.Wait() close(out) }() return out } func main() { input := make(chan int) ch1 := make(chan int) ch2 := make(chan int) // 使用 Fanout 分发数据 fanout(input, []chan<-int{ch1, ch2}) // 合并两个通道的结果 result := fanin(ch1, ch2) // 发送数据到输入通道 go func() { for i := 0; i < 10; i++ { input <- i } close(input) }() // 打印结果 for v := range result { fmt.Println(v) } } ``` --- ### 总结 以上实现了 Fanout 和 Fanin 的基本功能。Fanout 主要负责将数据分发到多个消费者，而 Fanin 则专注于收集来自不同生产者的多路数据流并统一处理。这种设计非常适合高并发环境下的任务分配与汇总需求。 ---