Go语言虽然有着高效的GMP调度模型，理论上支持成千上万的goroutine，但是goroutine过多，对调度，gc以及系统内存都会造成压力，这样会使我们的服务性能不升反降。常用做法可以用池化技术，构造一个协程池，把进程中的协程控制在一定的数量，防止系统中goroutine过多，影响服务性能。
sem := make(chan struct{}, 10)
for _, task := range tasks {
sem <- struct{}{}
go func(task Task) {
defer func() { <-sem }()
task()
}(task)
}
package main
import (
"fmt"
"sync"
"sync/atomic"
"time"
)
type Task struct {
f func() error // 具体的任务逻辑
}
func NewTask(funcArg func() error) *Task {
return &Task{
f: funcArg,
}
}
type Pool struct {
RunningWorkers int64 // 运行着的worker数量
Capacity int64 // 协程池worker容量
JobCh chan *Task // 用于worker取任务
sync.Mutex
}
func NewPool(capacity int64, taskNum int) *Pool {
return &Pool{
Capacity: capacity,
JobCh: make(chan *Task, taskNum),
}
}
func (p *Pool) GetCap() int64 {
return p.Capacity
}
func (p *Pool) incRunning() { // runningWorkers + 1
atomic.AddInt64(&p.RunningWorkers, 1)
}
func (p *Pool) decRunning() { // runningWorkers - 1
atomic.AddInt64(&p.RunningWorkers, -1)
}
func (p *Pool) GetRunningWorkers() int64 {
return atomic.LoadInt64(&p.RunningWorkers)
}
func (p *Pool) run() {
p.incRunning()
go func() {
defer func() {
p.decRunning()
}()
for task := range p.JobCh {
task.f()
}
}()
}
// AddTask 往协程池添加任务
func (p *Pool) AddTask(task *Task) {
// 加锁防止启动多个 worker
p.Lock()
defer p.Unlock()
if p.GetRunningWorkers() < p.GetCap() { // 如果任务池满, 则不再创建 worker
// 创建启动一个 worker
p.run()
}
// 将任务推入队列, 等待消费
p.JobCh <- task
}
func main() {
// 创建任务池
pool := NewPool(3, 10)
for i := 0; i < 20; i++ {
// 任务放入池中
pool.AddTask(NewTask(func() error {
fmt.Printf("I am Task\n")
return nil
}))
}
time.Sleep(1e9) // 等待执行
}