如何用Golang实现fan-in fan-out模式_Golang fan-in fan-out模式实践-Golang-PHP中文网

如何用Golang实现fan-in fan-out模式_Golang fan-in fan-out模式实践

P粉602998670

发布： 2025-11-22 14:51:06

原创

257人浏览过

fan-out通过多个goroutine并行处理任务，fan-in将结果汇总；示例中3个worker并行转字符串为大写，结果经fan-in合并输出，体现Golang并发优势。

如何用golang实现fan-in fan-out模式_golang fan-in fan-out模式实践

在并发编程中，fan-in fan-out 是一种常见的模式，用于提高程序的并行处理能力。Golang 凭借其轻量级的 goroutine 和强大的 channel 机制，非常适合实现这一模式。fan-in 指多个数据源合并到一个通道，fan-out 指将任务分发给多个工作者并行处理。下面通过实际例子说明如何在 Golang 中优雅地实现该模式。

什么是 fan-out（分发任务）

fan-out 的核心是将输入数据分发给多个工作协程，让它们并行处理，从而提升吞吐量。通常适用于 CPU 密集型或 I/O 密集型任务，比如批量请求、文件处理等。

实现方式：

启动多个 worker goroutine，每个都从同一个输入 channel 读取任务
使用独立的输出 channel 返回处理结果
所有 worker 共享输入，但各自写入自己的输出通道

注意：输入 channel 应在所有任务发送完毕后关闭，以便 worker 能正确退出。

什么是 fan-in（合并结果）

fan-in 负责将多个输出 channel 的结果汇聚到一个 channel 中，方便主流程统一处理。这在需要收集所有 worker 结果时非常有用。

立即学习“go语言免费学习笔记（深入）”；

常见做法是启动一个单独的 goroutine，将多个 channel 的数据复制到一个统一的输出 channel 中，直到所有输入 channel 关闭为止。

关键点：

LobeHub

LobeChat brings you the best user experience of ChatGPT, OLLaMA, Gemini, Claude

201

查看详情

使用 sync.WaitGroup 等待所有 worker 完成
每个 worker 完成后关闭自己的输出 channel
fan-in 协程在所有 channel 关闭后关闭汇总 channel

完整示例：并行处理字符串转大写

以下是一个完整的 fan-in fan-out 实现，模拟将一批字符串并行转为大写：

package main
<p>import (
"fmt"
"strings"
"sync"
)</p><p>func worker(in <-chan string, out chan<- string, wg *sync.WaitGroup) {
defer wg.Done()
for s := range in {
out <- strings.ToUpper(s)
}
close(out)
}</p><p>func fanIn(channels []<-chan string) <-chan string {
var wg sync.WaitGroup
merged := make(chan string)</p><pre class="brush:php;toolbar:false;"><pre class="brush:php;toolbar:false;">for _, ch := range channels {
    wg.Add(1)
    go func(c <-chan string) {
        defer wg.Done()
        for val := range c {
            merged <- val
        }
    }(c)
}

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

return merged

登录后复制

}

func main() { inputs := []string{"hello", "world", "golang", "concurrency", "fan", "out"}

in := make(chan string, len(inputs))
numWorkers := 3
outputs := make([]chan string, numWorkers)

for i := 0; i < numWorkers; i++ {
    outputs[i] = make(chan string)
}

var wg sync.WaitGroup

for i := 0; i < numWorkers; i++ {
    wg.Add(1)
    go worker(in, outputs[i], &wg)
}

for _, input := range inputs {
    in <- input
}
close(in)

merged := fanIn(outputs)

for result := range merged {
    fmt.Println(result)
}

wg.Wait()

登录后复制

}

这段代码中，3 个 worker 并行处理输入数据，各自将结果写入独立 channel，再通过 fan-in 将所有结果合并输出。整个过程高效且易于扩展。