在现代编程语言中，并发已经成为一种非常明确的需求。如今几乎每种编程语言都有一些并发的方法。有些语言具有丰富的结构体，可以将负载分配到操作系统的多线程上执行，例如Java，类似的还有Ruby。
Golang具有非常强大的并发模型，称为CSP(communicating sequential processes)，它将一个问题分解成更小的顺序进程，然后调度这些进程的几个实例，称为Goroutines。这些goroutines之间的通信是通过通道传递不可变的消息进行的。
本文，我们将探讨如何利用golang中的并发，以及如何通过协程池限制goroutine的使用。
一个简单的例子
假设我们有一个外部API调用，它需要大约100毫秒才能完成。如果我们有1000个这样的调用，并且我们同步地调用，大约需要100s才能完成。
//// model/data.go
package model
type SimpleData struct {
ID int
}
//// basic/basic.go
package basic
import (
"fmt"
"github.com/Joker666/goworkerpool/model"
"time"
)
func Work(allData []model.SimpleData) {
start := time.Now()
for i, _ := range allData {
Process(allData[i])
}
elapsed := time.Since(start)
fmt.Printf("Took ===============> %s\n", elapsed)
}
func Process(data model.SimpleData) {
fmt.Printf("Start processing %d\n", data.ID)
time.Sleep(100 * time.Millisecond)
fmt.Printf("Finish processing %d\n", data.ID)
}
//// main.go
package main
import (
"fmt"
"github.com/Joker666/goworkerpool/basic"
"github.com/Joker666/goworkerpool/model"
"github.com/Joker666/goworkerpool/worker"
)
func main() {
// 准备数据
var allData []model.SimpleData
for i := 0; i < 1000; i++ {
data := model.SimpleData{ ID: i }
allData = append(allData, data)
}
fmt.Printf("Start processing all work \n")
//处理数据
basic.Work(allData)
}
这里，我们有一个简单的模型，它包含一个只有整数值的数据结构。我们同步处理这结构体数组。这显然不是最佳解决方案，因为这些任务可以并发处理。让我们使用goroutines和通道将其转换为异步过程。
异步处理
////未使用协程池
func NotPooledWork(allData []model.SimpleData) {
start := time.Now()
var wg sync.WaitGroup
dataCh := make(chan model.SimpleData, 100)
wg.Add(1)
go func() {
defer wg.Done()
for data := range dataCh {
wg.Add(1)
go func(data model.SimpleData) {
defer wg.Done()
basic.Process(data)
}(data)
}
}()
for i, _ := range allData {
dataCh <- allData[i]
}
close(dataCh)
wg.Wait()
elapsed := time.Since(start)
fmt.Printf("Took ===============> %s\n", elapsed)
}
//// main.go
// Process
worker.NotPooledWork(allData)
在这里，我们创建了一个100的缓冲通道，并将传递给NoPooledWork函数的所有数据添加到该通道。由于它是一个缓冲通道，因此在数据被处理之前，它不能输入超过100个数据。这些都在一个goroutine中完成。从通道中每提取一个数据，添加一个goroutine，然后处理。在这里创建的goroutine数量是没有限制的。它可以处理传递的所有任务。理论上，在给定所需资源的情况下，处理尽可能多的数据是可能的。运行以上代码，我们将在100毫秒左右完成1000个任务。
存在问题
除非我们有足够的资源，否则我们在一段时间内所能做的资源分配是有限的。goroutine对象的最小大小是2K，但可以达到1GB。上面的代码并发地运行所有任务，假设有一百万个这样的任务，它会很快耗尽机器的内存和CPU。我们要么升级机器，要么找别的解决办法。
计算机科学家很久以前就已经考虑过这个问题，并提出了一个聪明的解决方案，称为线程池或WorkerPool。这个想法是为了让有限的计算资源来处理这些任务。一旦一个woker完成了一个任务，它就开始处理下一个任务。因此，任务一直等待被处理。这减少了CPU和内存的突发增长，随着时间的推移分配任务。
WorkerPool解决方案
/// worker/pooled.go
func PooledWork(allData []model.SimpleData) {
start := time.Now()
var wg sync.WaitGroup
workerPoolSize := 100
dataCh := make(chan model.SimpleData, workerPoolSize)
for i := 0; i < workerPoolSize; i++ {
wg.Add(1)
go func() {
defer wg.Done()
for data := range dataCh {
basic.Process(data)
}
}()
}
for i, _ := range allData {
dataCh <- allData[i]
}
close(dataCh)
wg.Wait()
elapsed := time.Since(start)
fmt.Printf("Took ===============> %s\n", elapsed)
}
//// main.go
// Process
worker.PooledWork(allData)
这里使用100个worker，创建100个goroutine来处理任务。我们可以把通道看作队列，每个goroutine都是一个消费者。多个goroutines可以监听同一个channel，但是channel上的每个数据只被处理一次。
这是一个不错的解决方案，执行上面的代码可以在1s完成所有的任务。
错误处理
我们还没有完全完成。上面的代码看起来像一个完整的解决方案，但其实不是。在这里未处理错误。我们创建一个需要处理错误的场景，看看如何处理。
//// worker/pooledError.go
func PooledWorkError(allData []model.SimpleData) {
start := time.Now()
var wg sync.WaitGroup
workerPoolSize := 100
dataCh := make(chan model.SimpleData, workerPoolSize)
errors := make(chan error, 1000)
for i := 0; i < workerPoolSize; i++ {
wg.Add(1)
go func() {
defer wg.Done()
for data := range dataCh {
process(data, errors)
}
}()
}
for i, _ := range allData {
dataCh <- allData[i]
}
close(dataCh)
wg.Add(1)
go func() {
defer wg.Done()
for {
select {
case err := <-errors:
fmt.Println("finished with error:", err.Error())
case <-time.After(time.Second * 1):
fmt.Println("Timeout: errors finished")
return
}
}
}()
defer close(errors)
wg.Wait()
elapsed := time.Since(start)
fmt.Printf("Took ===============> %s\n", elapsed)
}
func process(data model.SimpleData, errors chan<- error) {
fmt.Printf("Start processing %d\n", data.ID)
time.Sleep(100 * time.Millisecond)
if data.ID % 29 == 0 {
errors <- fmt.Errorf("error on job %v", data.ID)
} else {
fmt.Printf("Finish processing %d\n", data.ID)
}
}
//// main.go
// Process
worker.PooledWorkError(allData)
我们修改了process方法来处理一些随机错误。因此，为了在并发模型中处理错误，我们需要错误通道来保存错误数据。在所有任务完成处理后，我们检查errors通道以查找错误。error对象保存任务的ID，以便在需要时可以再次处理这些ID。
这比完全不处理错误的解决方案更好。但它还不是最完整的，在下一篇文章，我们将讨论如何创建一个健壮的专用WorkerPool，并在有限的goroutine数量下处理并发任务。
练习
如果多个任务处理失败，我们并不总是希望继续处理任务。表明我们有其他问题要先解决。作为练习，您可以尝试改进最后一个解决方案，当有两个以上的任务导致错误时，停止处理任务。
