我正在尝试使用工作池建立通用管道库。我为源,管道和接收器创建了一个接口。您会看到,管道的工作是从输入通道接收数据,对其进行处理,然后将结果输出到通道上。这是其预期的行为:
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
通道已关闭,因此会出现紧急情况。
将等待组放置在其他地方吗?还是有更好的方法等待所有工人完成?
[这些工作的完成速度可能与发送的速度一样快。在这种情况下,即使有更多项目要处理,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
通道,直到清空/关闭为止。似乎没有必要同时执行这两项。
尚不清楚为什么每个工作要一名工人,但是如果这样做,您可以重组外循环设置(请参阅下面未经测试的代码)。首先,这种方式消除了对工作池的需要。
不过,总是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剥离的想法,这可以使事情变得更清洁。但是,如果每个工作人员都打算剥离另一个goroutine,则该工作人员本身也必须使用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
计算可以
考虑使用比较简单(但未经测试):
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,但是要让每个运行例行程序[[one
来完成,并交付其结果,然后再返回阅读下一份工作: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数量,预期作业数量或其他。