如何在 Go 中实现数据流的扇入扇出处理-Golang-PHP中文网

Go中扇入扇出核心靠channel+goroutine，需控制并发、防泄漏、正确关通道：扇出要限流、传参防闭包陷阱、统一关输出通道；扇入用中转goroutine合并多通道，必监听done；组合时须用context超时/取消，缓冲区大小需按场景权衡。

如何在 go 中实现数据流的扇入扇出处理

Go 里做扇入（fan-in）和扇出（fan-out）不靠第三方库，核心就靠 channel + goroutine 的组合，关键在控制并发粒度、避免 goroutine 泄漏、以及正确关闭通道。

扇出：用 goroutine 并发调用多个数据源

扇出本质是把一个输入分发给多个并行任务。常见于批量请求下游服务（如多个 API、DB 分片、文件读取）。

别直接为每个任务起 goroutine 而不设限——容易打爆资源；用带缓冲的 semaphore channel 或 sync.WaitGroup 控制并发数
每个 goroutine 必须有自己的输入副本，否则闭包捕获循环变量会导致所有 goroutine 处理同一个值（经典陷阱）
输出 channel 不要由每个 goroutine 单独 close，只由发起方或专用 collector 关闭

示例：向 3 个 URL 并发发起 HTTP 请求

urls := []string{"https://a.com", "https://b.com", "https://c.com"}
out := make(chan string, len(urls))
var wg sync.WaitGroup
sem := make(chan struct{}, 2) // 限制最多 2 个并发
<p>for <em>, url := range urls {
wg.Add(1)
go func(u string) {
defer wg.Done()
sem  := http.Get(u)
out <p>go func() {
wg.Wait()
close(out)
}()</p>
<p>// 消费结果
for status := range out {
fmt.Println(status)
}</p></em></p>

扇入：合并多个 channel 的输出到一个 channel

扇入是把多个 producer 的 channel 统一收口，典型场景是等待多个异步操作完成、或聚合结果。

不要用 select 轮询多个 channel（无法预知数量），改用 reflect.Select 或更稳妥的「goroutine 中转」模式
中转 goroutine 必须监听 done channel 防止泄漏，尤其当上游 channel 可能永不关闭时
如果上游 channel 可能发送零值或错误，扇入后需额外标记来源，否则无法区分 “空响应” 和 “通道已关”

推荐做法：为每个输入 channel 启一个转发 goroutine

func fanIn(chs ...for _, ch := range chs {
    wg.Add(1)
    go func(c <p>}</p><div class="aritcle_card flexRow artxards">
											<div class="artcardd flexRow">
												<a class="aritcle_card_img" rel="nofollow" href="/xiazai/skill8661" title="Golang Uber Dig"><img
														src="https://img.php.cn/upload/skill/000/000/081/179215391842073.jpg" alt="Golang Uber Dig" onerror="this.onerror='';this.src='/static/lhimages/moren/morentu.png'" ></a>
												<div class="aritcle_card_info flexColumn">
													<a rel="nofollow" href="/xiazai/skill8661" title="Golang Uber Dig" class="overflowclass">Golang Uber Dig</a>
													<p class="overflowclass">使用 uber-go/dig 在 Golang 中实现依赖注入——基于反射容器，提供 Provide/Invoke、dig.In/dig.Out 参数与结果对象，支持命名依赖等</p>
												</div>
												<a rel="nofollow" href="/xiazai/skill8661" title="Golang Uber Dig" class="aritcle_card_btn flexRow flexcenter"><b></b><span>下载</span>
												</a>
											</div>
										</div><h3>扇入扇出组合时，done channel 是必须项</h3><p>真实业务中，扇出任务可能超时、失败或被取消，此时必须中断所有未完成的 goroutine，否则会卡死或泄漏。</p>

所有扇出 goroutine 都要监听 ，并在退出前清理资源（如关闭 HTTP 连接、释放锁）
扇入端不能只等 close(out)，要配合 ctx.Done() 提前退出 range 循环
别用 context.WithCancel 后忘记调用 cancel 函数——泄漏 context 和 goroutine

例如，在扇出中加入上下文控制：

ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
<p>for _, url := range urls {
go func(u string) {
select {
case out </p><h3>注意 buffer size 和阻塞风险</h3><p>channel 缓冲区大小不是越大越好，它直接影响内存占用和背压行为。</p>