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

今日不肯埋头，明日何以抬头！每日一句努力自己的话哈哈~哈喽，今天我将给大家带来一篇《关闭和发送到通道之间的竞争条件》，主要内容是讲解等等，感兴趣的朋友可以收藏或者有更好的建议在评论提出，我都会认真看的！大家一起进步，一起学习！

我正在尝试使用工作池构建通用管道库。我为源、管道和接收器创建了一个接口。您会看到，管道的工作是从输入通道接收数据，对其进行处理，然后将结果输出到通道上。这是它的预期行为：

1. 从输入通道接收数据。
2. 将数据委托给可用的工作人员。
3. worker 将结果发送到输出通道。
4. 所有工作人员完成后关闭输出通道。

func (p *pipe) Process(in chan interface{}) (out chan interface{}) {
    var wg sync.WaitGroup
    out = make(chan interface{}, 100)
    Go func() {
        for i := 1; i <= 100; i++ {
            go p.work(in, out, &wg)
        }
        wg.Wait()
        close(out)
    }()

    return
}

func (p *pipe) work(jobs <-chan interface{}, out chan<- interface{}, wg *sync.WaitGroup) {
    for j := range jobs {
        func(j Job) {
            defer wg.Done()
            wg.Add(1)

            res := doSomethingWith(j)

            out <- res
        }(j)
    }
}

但是，运行它可能会退出而不处理所有输入，或者因 send on closed channel 消息而出现恐慌。使用 -race 标志构建源会在 close(out) 和 out <-res 之间发出数据争用警告。

这就是我认为可能发生的情况。一旦许多工人完成了工作，wg 的计数器就会瞬间归零。因此，wg.wait() 已完成，程序继续执行 close(out)。与此同时，作业通道尚未完成数据生成，这意味着一些工作人员仍在另一个 goroutine 中运行。由于 out 通道已经关闭，因此会导致恐慌。

等待组应该放在其他地方吗？或者有没有更好的方法来等待所有工人完成？

解决方案

目前尚不清楚为什么每个工作需要一名工作人员，但如果您这样做，您可以重组您的外循环设置（请参阅下面未经测试的代码）。这种方式从一开始就消除了对工作池的需求。

不过，在分拆任何工作人员之前，请务必执行 wg.add 。在这里，您正好剥离了 100 名员工：

var wg sync.waitgroup
out = 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.waitgroup
out = 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

也就是说，每个匿名函数都是通道的另一个用户，因此在分离新的 goroutine 之前增加通道用户计数 (wg.add(1))。当您完成读取输入通道 jobs 后，调用 wg.done() （可能通过早期的 defer，但我在此处的末尾展示了它）。

思考这个问题的关键是 wg 计算此时可以写入通道的活动 goroutines 的数量。仅当没有 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 数量、预期作业数量或其他什么。

作业可能正在按如下方式完成：发送速度很快。在这种情况下，即使有更多的项目需要处理，waitgroup 也会在零附近浮动。

解决此问题的一个方法是在发送作业之前添加一个，并在发送所有作业后减少该作业，有效地将发送者视为“作业”之一。在这种情况下，最好在发送方中执行 wg.add：

func (p *pipe) Process(in chan interface{}) (out chan interface{}) {
    var wg sync.WaitGroup
    out = make(chan interface{}, 100)
    go func() {
        for i := 1; i <= 100; i++ {
            wg.Add(1)
            go p.work(in, out, &wg)
        }
        wg.Wait()
        close(out)
    }()

    return
}

func (p *pipe) work(jobs <-chan interface{}, out chan<- interface{}, wg *sync.WaitGroup) {
    for j := range jobs {
        func(j Job) {
            res := doSomethingWith(j)

            out <- res
            wg.Done()
        }(j)
    }
}

我在代码中注意到的一件事是，每个作业都会启动一个 goroutine。同时，每个作业循环处理 jobs 通道，直到为空/关闭。似乎没有必要两者都做。

终于介绍完啦！小伙伴们，这篇关于《关闭和发送到通道之间的竞争条件》的介绍应该让你收获多多了吧！欢迎大家收藏或分享给更多需要学习的朋友吧~编程网公众号也会发布golang相关知识，快来关注吧！