通道关闭责任划分与多生产者退出的设计方法

多生产者场景里，通道关闭最容易出错的地方不是 close 语法，而是“谁拥有关闭权”没有被写进设计。一个消费者看到结果够了就关闭共享输出，其他生产者仍在发送，就会触发 send on closed channel；如果消费者提前返回却没有通知上游，生产者又可能永久阻塞。

比较稳定的划分是：任务发送方关闭 jobs，多个 worker 只接收任务并发送结果，等待所有 worker 完成的协调者关闭 results，消费者只负责接收；当消费者决定提前结束时，通过 context 广播取消，而不是去关闭自己不拥有的业务通道。

• 输入通道：由最后一个任务发送者负责关闭，worker 不抢着关闭。
• 共享输出：由唯一的汇聚协调者在全部生产者退出后关闭。
• 退出信号：由创建这组 goroutine 的控制者取消，生产者在发送点响应。

先把每条通道的所有权写清楚

我在改造并发流水线时，会先画一张很小的责任表，而不是从 close(results) 开始写。判断方法只有一个：谁是这个通道的发送者，谁就负责在“不会再发送”时关闭它；接收者通常只负责读到关闭状态。

这里的“发送者关闭”不是绝对禁止封装层代关，而是要求关闭者能够证明所有发送已经结束。多生产者共享一个输出时，单个 worker 没有这个证明，因此它不能直接关闭 results。图 1 用静态关系图表示三条通道、worker 集合和关闭责任；它是说明图，不是运行截图。

用协调者等待所有生产者再关闭结果

共享输出的关闭动作要从 worker 中抽出来。下面的 fanOut 启动固定数量的生产者，使用 WaitGroup 记录它们；单独的协调 goroutine 等待计数归零后执行唯一一次 close(results)。

package pipeline

import (
    "context"
    "sync"
)

type Job struct {
    ID int
}

type Result struct {
    JobID int
}

// fanOut 让多个 worker 共享 jobs，并把结果汇聚到一个只读输出通道。
func fanOut(ctx context.Context, workers int, jobs 

注意 wg.Add 必须发生在启动等待 goroutine 之前，否则协调者可能在计数建立前就观察到零值。worker 的 defer wg.Done() 也不能省略，否则某个退出分支遗漏计数，消费者会一直等不到结果通道关闭。

输入 jobs 的关闭仍由发送任务的一方负责，例如主流程把全部任务写完后执行 close(jobs)。worker 是接收者，不应因为某个任务失败就关闭输入，否则其他任务发送者可能继续向已关闭通道写入。

下游提前结束时用取消广播退出

“读够一条结果就返回”会改变流水线的终止条件。此时仅等待 results 自然关闭不够，因为 worker 可能还在等待任务或阻塞发送。控制者应创建可取消的 context，在确定不再消费时调用 cancel；每个生产者同时监听取消信号和自身的输入、输出操作。

package main

import (
    "context"
    "fmt"
)

// firstResult 只取一个结果，随后取消整条流水线，避免剩余 worker 继续发送。
func firstResult(parent context.Context, jobs chan Job) (Result, error) {
    ctx, cancel := context.WithCancel(parent)
    defer cancel() // 所有返回路径都广播退出，包含结果处理失败。

    results := fanOut(ctx, 4, jobs)
    result, ok := 

这里的取消只负责让 goroutine 退出，不替代输入输出通道的关闭责任。若任务分发方还在发送 jobs，它也要在发送处监听同一个 ctx.Done()，否则发送方仍可能在无人接收时卡住。对于不需要提前返回的完整消费场景，正常关闭 jobs 并等待 results 结束即可，不必额外制造取消分支。

并发退出的代价：关闭、取消和等待不能混成一个动作

通道关闭表达“以后不会再有值”，取消表达“请尽快停止当前工作”，WaitGroup 表达“这组 goroutine 是否已经退出”。三者语义不同，混用时最容易出现下面几种反模式：

如果 worker 持有文件、连接或临时资源，退出函数还要把清理放在 defer 中，并确保取消路径也能走到清理逻辑。不要把关闭结果通道当成资源回收信号：结果关闭只说明发送端退出，资源回收仍由各自的 goroutine 负责。

上线前重点检查哪些退出边界

我会把测试重点放在“谁先结束”而不是某次输出顺序。至少覆盖：输入正常关闭、某个 worker 处理失败、消费者只取一条结果后取消、多个 worker 同时返回，以及任务发送方在取消时停止发送。用 go test -race ./... 检查共享状态时，还要确认测试真的走到了取消和关闭分支。

如果测试偶发卡住，先看 WaitGroup 是否有对应的 Done，再看所有可能阻塞的发送是否有取消分支，最后检查是不是存在第二个关闭者。不要通过增加缓冲区容量来掩盖所有权错误；缓冲区只能延迟阻塞，不能改变通道关闭语义。

常见问题

多个 worker 能不能各自判断自己是最后一个，然后关闭结果通道？

除非另有可靠的单一协调机制，否则不建议。多个 worker 可能同时认为自己接近结束，关闭动作应交给等待计数归零的协调者，所有权更容易审查。

接收者什么时候可以关闭通道？

只有接收者同时拥有“不会再有任何发送者”的事实，并且这个事实被设计明确时才可以。普通多生产者流水线里，消费者通常没有这个信息，因此不应关闭共享输出。

取消后还需要关闭 jobs 吗？

如果发送方已经停止并且通道只用于本次流水线，关闭输入有助于让仍在排空的 worker 正常结束；但不能把关闭输入当成取消替代品。发送方必须先停止发送，关闭动作也必须由它的所有权方完成。