Go Worker pool 工作池实现详解

Z时代
2024-01-10
分类：综合

Channel缓冲的重要用途之一是实现工作池。

通常，工作池是等待分配给它们的任务的线程的集合。一旦他们完成分配的任务，他们就会再次为下一个任务提供服务。

我们将使用缓冲通道实现一个工作池。我们的工作池将执行计算输入数字的数字总和的任务。例如，如果传递了 234，则输出将为 9 (2 + 3 + 4)。工作池的输入将是一个伪随机整数列表。

下面是我们的工作池的核心功能

创建一个 Goroutines 池，它侦听输入缓冲通道，等待分配作业

将作业添加到输入缓冲通道

作业完成后将结果写入输出缓冲通道

从输出缓冲通道读取和打印结果

我们将逐步实现这个程序，以使其更易于理解。

第一步将是创建表示作业和结果的结构体


type Job struct {  
    id       int
    randomno int
}
type Result struct {  
    job         Job
    sumofdigits int
}

每个 Job 结构体都有一个 id 和一个 randomno ，必须为其计算各个数字的总和。

Result 结构体有一个 job 字段，该字段是在 sumofdigits 字段中保存结果（单个数字的总和）的作业。

下一步是创建用于接收作业和写入输出的缓冲通道。


var jobs = make(chan Job, 10)  
var results = make(chan Result, 10)

Worker Goroutines 在 jobs 缓冲通道上监听新任务。任务完成后，将结果写入 results 缓冲通道。

下面我们定义一个 digits 函数，该函数的主要功能是计算各个数字的总和并返回它。我们将给这个函数添加一个 2 秒的睡眠，只是为了模拟这个函数需要一些时间来计算结果的事实。


funcdigits(number int)int {  
    sum := 0
    no := number
for no != 0 {
        digit := no % 10
        sum += digit
        no /= 10
    }
    time.Sleep(2 * time.Second)
return sum
}

接下来我们定义一个 worker 函数。


funcworker(wg *sync.WaitGroup) {  
for job := range jobs {
        output := Result{job, digits(job.randomno)}
        results <- output
    }
    wg.Done()
}

上述函数创建一个从 jobs 通道读取数据的工作协程，使用当前作业和 digits 函数的返回值创建一个 Result 结构体，然后将结果写入 results 缓冲通道。此函数将 WaitGroup wg 作为参数，当所有作业完成后，它将调用 Done() 方法。

createWorkerPool 函数将创建一个 worker Goroutines 池。


funccreateWorkerPool(noOfWorkers int) {  
var wg sync.WaitGroup
for i := 0; i < noOfWorkers; i++ {
        wg.Add(1)
go worker(&wg)
    }
    wg.Wait()
close(results)
}

上面的函数将要创建的 worker 数量作为参数。它在创建 Goroutine 之前调用 wg.Add(1) 来增加 WaitGroup 计数器。然后它通过将 WaitGroup wg 的指针传递给 worker 函数来创建 worker Goroutines。在创建所需的 Goroutines 后，它通过调用 wg.Wait() 等待所有 Goroutines 完成它们的执行。在所有 Goroutines 执行完毕后，它关闭 results 通道，因为所有 Goroutines 都已完成它们的执行，并且没有其他的协程会进一步写入 results 通道。

现在我们已经准备好工作池，让我们继续定义一个用来进行分配的函数。


funcallocate(noOfJobs int) {  
for i := 0; i < noOfJobs; i++ {
        randomno := rand.Intn(999)
        job := Job{i, randomno}
        jobs <- job
    }
close(jobs)
}

上面的 allocate 函数以要创建的任务的数量作为输入参数，生成最大值为998的伪随机数，以随机数和for循环计数器i为id创建Job结构体，然后将它们写入 jobs 通道。它在写入所有作业后关闭 jobs 通道。

下一步是创建读取 results 通道并打印输出的函数。


funcresult(done chanbool) {  
for result := range results {
        fmt.Printf("Job id %d, input random no %d , sum of digits %d\n", result.job.id, result.job.randomno, result.sumofdigits)
    }
    done <- true
}

result 函数读取 results 通道并打印 Job ID，输入的随机数，以及随机数的数字总和。 result 函数还将接收done 通道作为参数，一旦它打印了所有结果，它就会写入该通道。

我们现在已经准备好了一切。让我们继续完成最后一步，编写 main() 函数调用所有这些函数。


funcmain() {  
    startTime := time.Now()
    noOfJobs := 100
go allocate(noOfJobs)
    done := make(chanbool)
go result(done)
    noOfWorkers := 10
    createWorkerPool(noOfWorkers)
    <-done
    endTime := time.Now()
    diff := endTime.Sub(startTime)
    fmt.Println("total time taken ", diff.Seconds(), "seconds")
}

我们首先保存程序的执行开始时间，在最后我们计算 endTime 和 startTime 之间的时间差并显示程序花费的总时间。这是必要的，因为我们将通过改变 Goroutines 的数量来做一些基准测试。

noOfJobs 设置为 100，然后调用 allocate 将作业添加到 jobs 通道。

然后创建 done 通道并将其传递给 result Goroutine，以便它可以开始打印输出并在打印完所有内容后通知主协程。

最后，通过调用 createWorkerPool 函数创建了一个包含 10 个 job Goroutines 的池，然后 main 在 done 通道上等待所有要打印的结果。

下面是完整的程序。


package main
import (  
"fmt"
"math/rand"
"sync"
"time"
)
type Job struct {  
    id       int
    randomno int
}
type Result struct {  
    job         Job
    sumofdigits int
}
var jobs = make(chan Job, 10)  
var results = make(chan Result, 10)
funcdigits(number int)int {  
    sum := 0
    no := number
for no != 0 {
        digit := no % 10
        sum += digit
        no /= 10
    }
    time.Sleep(2 * time.Second)
return sum
}
funcworker(wg *sync.WaitGroup) {  
for job := range jobs {
        output := Result{job, digits(job.randomno)}
        results <- output
    }
    wg.Done()
}
funccreateWorkerPool(noOfWorkers int) {  
var wg sync.WaitGroup
for i := 0; i < noOfWorkers; i++ {
        wg.Add(1)
go worker(&wg)
    }
    wg.Wait()
close(results)
}
funcallocate(noOfJobs int) {  
for i := 0; i < noOfJobs; i++ {
        randomno := rand.Intn(999)
        job := Job{i, randomno}
        jobs <- job
    }
close(jobs)
}
funcresult(done chanbool) {  
for result := range results {
        fmt.Printf("Job id %d, input random no %d , sum of digits %d\n", result.job.id, result.job.randomno, result.sumofdigits)
    }
    done <- true
}
funcmain() {  
    startTime := time.Now()
    noOfJobs := 100
go allocate(noOfJobs)
    done := make(chanbool)
go result(done)
    noOfWorkers := 10
    createWorkerPool(noOfWorkers)
    <-done
    endTime := time.Now()
    diff := endTime.Sub(startTime)
    fmt.Println("total time taken ", diff.Seconds(), "seconds")
}