multithreading - Go 中的简单工作池

Question

我正在尝试在 go 中实现一个简单的工作线程池，但一直遇到问题。我只想让一定数量的 worker 完成一定数量的工作，然后再做更多的工作。我使用的代码类似于:

    jobs := make(chan imageMessage, 1)
    results := make(chan imageMessage, 1)

    for w := 0; w < 2; w++ {
        go worker(jobs, results)
    }

    for j := 0; j < len(images); j++ {
        jobs <- imageMessage{path: paths[j], img: images[j]}
    }
    close(jobs)

    for r := 0; r < len(images); r++ {
        <-results
    }
}

func worker(jobs <-chan imageMessage, results chan<- imageMessage) {
    for j := range jobs {
        processImage(j.path, j.img)
        results <- j
    }
}

我的理解是，这应该创建 2 个 worker ，他们可以一次做 1 件“事情”，并且在他们完成那 1 件事情时会继续获得更多工作，直到没有其他事情可做。但是，我得到了 fatal error: all goroutines are sleep - deadlock!

如果我将缓冲区设置为 100 之类的大值，这会起作用，但我希望能够限制一次完成的工作。

我觉得我很接近，但显然遗漏了一些东西。

Answer 1

问题在于，一旦您成功发送了jobs channel 上的所有作业，您就只会开始“耗尽”results channel 。但是为了能够发送所有作业，jobs channel 必须有足够大的缓冲区，或者 worker goroutines 必须能够从中使用作业。

但是一个 worker goroutines 在消费一个工作时，在它可以接受下一个工作之前，将结果发送到 results channel 。如果 results channel 的缓冲区已满，将阻塞发送结果。

但是最后一部分——在发送结果时被阻塞的工作协程——只能通过从 results channel 接收来“解除阻塞”——直到你可以发送所有的作业。如果 jobs channel 和 results channel 的缓冲区不能容纳您的所有作业，则会出现死锁。这也解释了为什么如果将缓冲区大小增加到一个大值它会起作用:如果作业可以放入缓冲区，则不会发生死锁，并且在成功发送所有作业后，您的最终循环将耗尽 results channel 。

解决方案？在自己的 goroutine 中运行生成和发送作业，因此您可以“立即”开始从 results channel 接收数据，而不必等待发送所有作业，这意味着 worker goroutines 不会永远被阻塞尝试发送结果:

go func() {
    for j := 0; j < len(images); j++ {
        jobs <- imageMessage{path: paths[j], img: images[j]}
    }
    close(jobs)
}()

在 Go Playground 上试试.

另请查看 Is this an idiomatic worker thread pool in Go? 中的类似实现