Go并发和协程池

在现代编程语言中，并发已经成为一种非常明确的需求。如今几乎每种编程语言都有一些并发的方法。有些语言具有丰富的结构体，可以将负载分配到操作系统的多线程上执行，例如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数量下处理并发任务。

练习

如果多个任务处理失败，我们并不总是希望继续处理任务。表明我们有其他问题要先解决。作为练习，您可以尝试改进最后一个解决方案，当有两个以上的任务导致错误时，停止处理任务。