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

func Process(data model.SimpleData) {
    fmt.Printf("Start processing %dn", data.ID)
    time.Sleep(100 * time.Millisecond)
    fmt.Printf("Finish processing %dn", 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 

这里,我们有一个简单的模型,它包含一个只有整数值的数据结构。我们同步处理这结构体数组。这显然不是最佳解决方案,因为这些任务可以并发处理。让我们使用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  %sn", 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  %sn", 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  %sn", elapsed)
}

func process(data model.SimpleData, errors chan

我们修改了process方法来处理一些随机错误。因此,为了在并发模型中处理错误,我们需要错误通道来保存错误数据。在所有任务完成处理后,我们检查errors通道以查找错误。error对象保存任务的ID,以便在需要时可以再次处理这些ID。

这比完全不处理错误的解决方案更好。但它还不是最完整的,在下一篇文章,我们将讨论如何创建一个健壮的专用WorkerPool,并在有限的goroutine数量下处理并发任务。

练习

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

由于版权原因,本站共享资源只供云盘资源,版权均属于影片公司所有,请在下载后24小时删除,切勿用于商业用途。本站所有资源信息均从互联网搜索而来,本站不对显示的内容承担责任,如您认为本站页面信息侵犯了您的权益,请附上版权证明邮件并发送到[email protected]告知,我们会在收到邮件后72小时内删除。
想开点 » Go并发和协程池