当前位置：首页 > 文章列表 > Golang > Go教程 > Golang实现并发模式详解

Golang实现并发模式详解

2025-12-02 19:42:47 0浏览收藏

**Golang实现Fan-In Fan-Out模式详解：提升并发处理能力** 在Golang并发编程中，Fan-In Fan-Out是一种强大的模式，能显著提升程序并行处理能力。本文深入解析Fan-Out（任务分发）和Fan-In（结果合并）机制，通过示例展示如何利用Golang的goroutine和channel优雅实现。示例代码演示了3个worker并行将字符串转为大写，并通过Fan-In合并结果，充分体现Golang的并发优势。掌握这种模式，能有效解决CPU密集型和I/O密集型任务，如批量请求和文件处理，从而优化程序性能。此外，文章还提供优化建议和注意事项，助您在实际项目中灵活应用Fan-In Fan-Out模式。

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 结果时非常有用。

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

关键点：

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

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

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

package main
import (
"fmt"
"strings"
"sync"
)
func worker(in <-chan string, out chan<- string, wg *sync.WaitGroup) {
defer wg.Done()
for s := range in {
out <- strings.ToUpper(s)
}
close(out)
}
func fanIn(channels []<-chan string) <-chan string {
var wg sync.WaitGroup
merged := make(chan string)
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 将所有结果合并输出。整个过程高效且易于扩展。