限制运行的并发任务数

Question

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

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

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

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

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

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

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()
}