限制运行的并发任务数

Question

所以我经常遇到这个问题go。假设我有一个包含 100,000 行文本的文本文件。现在我想将所有这些行保存到数据库中。所以我会做这样的事情：

file, _ := iotuil.ReadFile("file.txt")

fileLines := strings.Split(string(file), "\n")

现在我将遍历文件中的所有行：

for _, l := range fileLines{
  saveToDB(l)
}

现在我想saveToDB同时运行这个函数：

var wg sync.WaitGroup

for _, l := range fileLines{
  wg.Add(1)
  go saveToDB(l, &wg)
}

wg.Wait()

我不知道这是否有问题，但那会运行 100,000 个并发函数。有什么办法可以说嘿，运行 100 个并发函数，等待所有这些函数完成，然后再运行 100 个。

for i, _ := range fileLine {
  for t = 0; t < 100; t++{
    wg.Add(1)
    go saveToDB(fileLine[i], &wg)
  }
  wg.Wait()
}

我需要做类似的事情还是有更干净的方法来解决这个问题？或者我运行 100,000 个并发任务不是问题？

Answer 1

我认为最好的方法是保留一个工作 goroutine 池，在通道中为它们分派工作，然后关闭通道以便它们退出。

像这样：

// create a channel for work "tasks"
ch := make(chan string)

wg := sync.WaitGroup{}

// start the workers
for t := 0; t < 100; t++{
    wg.Add(1)
    go saveToDB(ch, &wg)
}

// push the lines to the queue channel for processing
for _, line := range fileline {
    ch <- line
}

// this will cause the workers to stop and exit their receive loop
close(ch)

// make sure they all exit
wg.Wait()

然后 saveFunction 看起来像这样：

func saveToDB(ch chan string, wg *sync.WaitGroup) {
    // cnosume a line
    for line := range ch {
        // do work
        actuallySaveToDB(line)
    }
    // we've exited the loop when the dispatcher closed the channel, 
    // so now we can just signal the workGroup we're done
    wg.Done()
}