Golang并发，处理批量项目

Question

我正在编写一个程序来处理文本文件中的数百万行，500k需要5秒才能验证文件，我想加快速度。

我想对这些项进行循环，并异步处理其中的x个项，然后等待响应，看看是否应该继续。

我写了一些伪代码，我不确定我写的东西是否有意义，只是看起来相当复杂，有没有更简单、更优雅的方法可以做到这一点。

package main
import (
"fmt"
"sync"
"time"
)
func main() {
// Need an object to loop over
// need a loop to read the response
items := 100000
concurrency := 20
sem := make(chan bool, concurrency)
returnChan := make(chan error)
finChan := make(chan bool)
var wg sync.WaitGroup
go func() {
for x := 0; x < items; x++ {
// loop over all items
// only do maxitems at a time
wg.Add(1)
sem <- true
go delayFunc(x, sem, returnChan, &wg)
}
wg.Wait()
finChan <- true
}()
var err error
finished := false
for {
select {
case err = <-returnChan:
if err != nil {
break
}
case _ = <-finChan:
finished = true
break
default:
continue
}
if err != nil || finished == true {
break
}
}
fmt.Println(err)
}
func delayFunc(x int, sem chan bool, returnChan chan error, wg *sync.WaitGroup) {
//fmt.Printf("PROCESSING (%v)n", x)
time.Sleep(10 * time.Millisecond)
<-sem // release the lock
wg.Done()
if x == 95000 {
returnChan <- fmt.Errorf("Something not right")
} else {
returnChan <- nil
}
}

Answer 1

您的代码看起来不错，实现了Go模式中常用的代码。不利的一面是，你会为每一个项目生成worker goroutine。趁着便宜生孩子可不是免费的。另一种方法是产生N个工人，并通过渠道为他们提供物品。像这样的

package main
import (
"fmt"
"time"
)
func main() {
items := 100
concurrency := 10
in := make(chan int)
ret := make(chan error)
for x := 0; x < concurrency; x++ {
go worker(in, ret)
}
go func() {
for x := 0; x < items; x++ {
// loop over all items
in <- x
}
close(in)
}()
for err := range ret {
if err != nil {
fmt.Println(err.Error())
break
}
}
}
func worker(in chan int, returnChan chan error) {
//fmt.Printf("PROCESSING (%v)n", x)
for x := range in {
if x == 95 {
returnChan <- fmt.Errorf("Something not right")
} else {
returnChan <- nil
}
time.Sleep(10 * time.Millisecond)
}
returnChan <- fmt.Errorf("The End")
}

游乐场