在完成之前去常规结束

Question

我想用多个 go 例程异步执行一些事情。我传入用于异步处理文件的“线程”数量。这些文件是要处理的字符串数组。

queue := make(chan string)

threadCount := c.Int("threads")

if c.Int("threads") < len(files) {
    threadCount = len(files)
} 

log.Infof("Starting %i processes", c.Int("threads"))

for i := 0; i < threadCount; i++ {
    go renderGoRoutine(queue)
}

for _, f := range files {
    queue <- f
}
close(queue)

而例程本身看起来是这样的：

func renderGoRoutine(queue chan string) {
    for file := range queue { 
        // do some heavy lifting stuff with the file
    }
}

只要我只使用一个线程，它就可以正常工作。只要我多取一个，它就会在完成所有 go 例程之前退出/离开范围。

如何让它处理所有内容？

上一个问题：Using a channel for dispatching tasks to go routine

Answer 1

使用 WaitGroups 是一种选择。

一开始，您将任务数添加到 WaitGroup 中，并在每个任务完成后递减 WaitGroup 中的计数器。等到所有任务在代码流结束时完成。

查看示例：https://godoc.org/sync#WaitGroup

您的代码将如下所示：

queue := make(chan string)

wg := sync.WaitGroup{}
wg.Add(len(files))

threadCount := c.Int("threads")

if c.Int("threads") < len(files) {
    threadCount = len(files)
}

log.Infof("Starting %i processes", c.Int("threads"))

for i := 0; i < threadCount; i++ {
    go renderGoRoutine(queue)
}


for _, f := range files {
    queue <- f
}

close(queue)
wg.Wait()

渲染GoRoutine：

func renderGoRoutine(queue chan string) {
    for file := range queue {
        // do some heavy lifting stuff with the file
        // decrement the waitGroup counter
        wg.Done()
    }
}

Answer 2

您正在使用该频道发布要完成的工作。一旦从队列中取出最后一项（未完成处理），您的程序就会退出。

您可以在renderGoRoutine 的末尾使用写入通道来表示处理结束。

在顶部：

sync := make(chan bool)

在renderGoRoutine末尾（假设在同一个文件中）：

sync <- true

在底部：

for f := range sync {
    <- sync
}

现在您的程序将等待处理完文件数。

或者有一个完整的例子：

queue := make(chan string)
sync := make(chan bool)

threadCount := c.Int("threads")

if c.Int("threads") < len(files) {
    threadCount = len(files)
} 

log.Infof("Starting %i processes", c.Int("threads"))

for i := 0; i < threadCount; i++ {
    go renderGoRoutine(queue)
}

for _, f := range files {
    queue <- f
}
close(queue)

for f := range sync {
    <- sync
}

并且应该像这样更改例程：

func renderGoRoutine(queue chan string) {
    for file := range queue { 
        // do some heavy lifting stuff with the file
        sync <- true
    }
}

Answer 3

我确实忘记等待所有任务完成。这可以通过等待所有循环结束来简单地完成。由于close(channel) 确实结束了for range channel，因此可以像这样使用与频道的简单同步：

sync := make(chan bool)
queue := make(chan string)

threadCount := c.Int("threads")

if c.Int("threads") < len(files) {
    threadCount = len(files)
} 

log.Infof("Starting %i processes", c.Int("threads"))

for i := 0; i < threadCount; i++ {
    go renderGoRoutine(queue)
}

for _, f := range files {
    queue <- f
}
close(queue)

for i := 0; i < threadCount; i++ {
    <- sync
}

最后但并非最不重要的一点是，每当迭代停止时写入通道。

func renderGoRoutine(queue chan string) {
    for file := range queue { //whatever is done here
    }
    sync <- true
}