常见模式之四：工作池/协程池模式
定义
顾名思义，就是有固定数量的工人（协程），去执行批量的任务
使用场景
-
适用于需要限制并发执行任务数量的情况
-
创建一个固定大小的 goroutine 池，将任务分发给池中的 goroutine 并等待它们完成，使用带缓冲的通道来接收任务，以避免阻塞主线程
示例
有生产需求，建议使用大佬写的 ants库 ，以下是模拟协程池的简单示例
假设，我们有固定数量（2个）工人执行批量（4个）任务
package main
import (
"fmt"
"sync"
)
type Task struct {
TaskFunc func() interface{}
}
type WorkerPool struct {
Size int
Wg *sync.WaitGroup
Tasks chan Task
Results chan Result
}
type Result struct {
ID int
Res interface{}
}
func NewWorkerPool(workerNum, taskBufSize int) *WorkerPool {
return &WorkerPool{
Size: workerNum,
Wg: &sync.WaitGroup{},
Tasks: make(chan Task, taskBufSize),
Results: make(chan Result, taskBufSize),
}
}
func (w *WorkerPool) AddTask(task Task) {
w.Tasks <- task
}
func (w *WorkerPool) Run() {
for i := 1; i <= w.Size; i++ {
w.Wg.Add(1)
go func(id int) {
defer w.Wg.Done()
w.Work(id, w.Tasks, w.Results)
}(i)
}
}
func (w *WorkerPool) Work(Id int, tasks chan Task, results chan Result) {
for task := range tasks {
results <- Result{
ID: Id,
Res: task.TaskFunc(),
}
}
}
func main() {
pool := NewWorkerPool(3, 10)
pool.Run()
pool.AddTask(Task{TaskFunc: func() interface{} {
return 2 * 3
}})
pool.AddTask(Task{TaskFunc: func() interface{} {
return 4 * 5
}})
pool.AddTask(Task{TaskFunc: func() interface{} {
return 6 * 7
}})
pool.AddTask(Task{TaskFunc: func() interface{} {
return 8 * 9
}})
close(pool.Tasks)
go func() {
pool.Wg.Wait()
close(pool.Results)
}()
for v := range pool.Results {
fmt.Println(v.ID, v.Res)
}
}
