关闭和发送到通道之间的竞争条件

目前尚不清楚为什么每个工作需要一名工人，但如果你这样做了，你可以重组你的外循环设置（见下面未经测试的代码）。这种方式首先消除了对工作池的需求。但是，总是在剥离任何工人wg.Add 之前做一个。在这里，你正在剥离 100 个工人：var wg sync.WaitGroupout = make(chan interface{}, 100)go func() {    for i := 1; i <= 100; i++ {        go p.work(in, out, &wg)    }    wg.Wait()    close(out)}()因此，您可以这样做：var wg sync.WaitGroupout = make(chan interface{}, 100)go func() {    wg.Add(100)  // ADDED - count the 100 workers    for i := 1; i <= 100; i++ {        go p.work(in, out, &wg)    }    wg.Wait()    close(out)}()请注意，您现在可以将wg自己向下移动到剥离工作人员的 goroutine 中。如果你放弃让每个工作人员将工作作为新的 goroutine 分拆的想法，这可以让事情变得更干净。但是，如果每个 worker 都将衍生出另一个 goroutine，那么该 worker 本身也必须使用wg.Add，如下所示：for j := range jobs {    wg.Add(1)  // ADDED - count the spun-off goroutines    func(j Job) {        res := doSomethingWith(j)        out <- res        wg.Done()  // MOVED (for illustration only, can defer as before)    }(j)}wg.Done() // ADDED - our work in `p.work` is now done也就是说，每个匿名函数都是通道的另一个用户，因此在启动wg.Add(1)新的 goroutine 之前增加用户的通道计数 ()。当你读完输入通道jobs后，调用wg.Done()（可能是通过之前的defer，但我在最后展示了它）。考虑这一点的关键是计算此时可以写入通道wg的活动 goroutine 的数量。只有当没有goroutine 打算再写时，它才会变为零。这样可以安全地关闭通道。考虑使用更简单的（但未经测试）：func (p *pipe) Process(in chan interface{}) (out chan interface{}) {    out = make(chan interface{})    var wg sync.WaitGroup    go func() {        defer close(out)        for j := range in {            wg.Add(1)            go func(j Job) {                res := doSomethingWith(j)                out <- res                wg.Done()            }(j)        }        wg.Wait()    }()    return out}您现在有一个 goroutine 正在in尽可能快地读取通道，并在运行时分拆作业。每个传入的工作都会得到一个 goroutine，除非他们提前完成工作。没有池，每个作业只有一个工人（与您的代码相同，只是我们淘汰了没有做任何有用的池）。或者，因为只有一些可用的 CPU，所以在开始时像以前一样分拆一些 goroutine，但让每个 goroutine 运行一个作业以完成，并交付其结果，然后返回阅读下一个作业：func (p *pipe) Process(in chan interface{}) (out chan interface{}) {    out = make(chan interface{})    go func() {        defer close(out)        var wg sync.WaitGroup        ncpu := runtime.NumCPU()  // or something fancier if you like        wg.Add(ncpu)        for i := 0; i < ncpu; i++ {            go func() {                defer wg.Done()                for j := range in {                    out <- doSomethingWith(j)                }            }()        }        wg.Wait()    }    return out}通过使用runtime.NumCPU()，我们只能获得与运行作业的 CPU 一样多的工作人员读取作业。那些是游泳池，他们一次只做一项工作。如果输出通道读取器结构良好（即，不会导致管道阻塞），通常不需要缓冲输出通道。如果不是，则此处的缓冲深度会限制您可以“提前”完成多少工作，而无论谁正在使用结果。根据“提前工作”执行此操作的有用程度来设置它 - 不一定是 CPU 的数量，或预期作业的数量，或其他任何东西。

关闭和发送到通道之间的竞争条件

2回答