go中控制goroutine数量

时间:2023-03-09 04:04:38
go中控制goroutine数量

控制goroutine数量

前言

goroutine被无限制的大量创建,造成的后果就不啰嗦了,主要讨论几种如何控制goroutine的方法

控制goroutine的数量

通过channel+sync

var (
// channel长度
poolCount = 5
// 复用的goroutine数量
goroutineCount = 10
) func pool() {
jobsChan := make(chan int, poolCount) // workers
var wg sync.WaitGroup
for i := 0; i < goroutineCount; i++ {
wg.Add(1)
go func() {
defer wg.Done()
for item := range jobsChan {
// ...
fmt.Println(item)
}
}()
} // senders
for i := 0; i < 1000; i++ {
jobsChan <- i
} // 关闭channel,上游的goroutine在读完channel的内容,就会通过wg的done退出
close(jobsChan)
wg.Wait()
}

通过WaitGroup启动指定数量的goroutine,监听channel的通知。发送者推送信息到channel,信息处理完了,关闭channel,等待goroutine依次退出。

使用semaphore

package main

import (
"context"
"fmt"
"sync"
"time" "golang.org/x/sync/semaphore"
) const (
// 同时运行的goroutine上限
Limit = 3
// 信号量的权重
Weight = 1
) func main() {
names := []string{
"小白",
"小红",
"小明",
"小李",
"小花",
} sem := semaphore.NewWeighted(Limit)
var w sync.WaitGroup
for _, name := range names {
w.Add(1)
go func(name string) {
sem.Acquire(context.Background(), Weight)
// ... 具体的业务逻辑
fmt.Println(name, "-吃饭了")
time.Sleep(2 * time.Second)
sem.Release(Weight)
w.Done()
}(name)
}
w.Wait() fmt.Println("ending--------")
}

借助于x包中的semaphore,也可以进行goroutine的数量限制。

线程池

不过原本go中的协程已经是非常轻量了,对于协程池还是要根据具体的场景分析。

对于小场景使用channel+sync就可以,其他复杂的可以考虑使用第三方的协程池库。

几个开源的线程池的设计

fasthttp中的协程池实现

fasthttpnet/http效率高很多倍的重要原因,就是利用了协程池。来看下大佬的设计思路。

1、按需增长goroutine数量,有一个最大值,同时监听channelServer会把accept到的connection放入到channel中,这样监听的goroutine就能处理消费。

2、本地维护了一个待使用的channel列表,当本地channel列表拿不到ch,会在sync.pool中取。

3、如果workersCount没达到上限,则从生成一个workerFunc监听workerChan

4、对于待使用的channel列表,会定期清理掉超过最大空闲时间的workerChan

看下具体实现

// workerPool通过一组工作池服务传入的连接
// 按照FILO(先进后出)的顺序,即最近停止的工作人员将为下一个工作传入的连接。
//
// 这种方案能够保持cpu的缓存保持高效(理论上)
type workerPool struct {
// 这个函数用于server的连接
// It must leave c unclosed.
WorkerFunc ServeHandler // 最大的Workers数量
MaxWorkersCount int LogAllErrors bool MaxIdleWorkerDuration time.Duration Logger Logger lock sync.Mutex
// 当前worker的数量
workersCount int
// worker停止的标识
mustStop bool // 等待使用的workerChan
// 可能会被清理
ready []*workerChan // 用来标识start和stop
stopCh chan struct{} // workerChan的缓存池,通过sync.Pool实现
workerChanPool sync.Pool connState func(net.Conn, ConnState)
} // workerChan的结构
type workerChan struct {
lastUseTime time.Time
ch chan net.Conn
}
Start
func (wp *workerPool) Start() {
// 判断是否已经Start过了
if wp.stopCh != nil {
panic("BUG: workerPool already started")
}
// stopCh塞入值
wp.stopCh = make(chan struct{})
stopCh := wp.stopCh
wp.workerChanPool.New = func() interface{} {
// 如果单核cpu则让workerChan阻塞
// 否则,使用非阻塞,workerChan的长度为1
return &workerChan{
ch: make(chan net.Conn, workerChanCap),
}
}
go func() {
var scratch []*workerChan
for {
wp.clean(&scratch)
select {
// 接收到退出信号,退出
case <-stopCh:
return
default:
time.Sleep(wp.getMaxIdleWorkerDuration())
}
}
}()
} // 如果单核cpu则让workerChan阻塞
// 否则,使用非阻塞,workerChan的长度为1
var workerChanCap = func() int {
// 如果GOMAXPROCS=1,workerChan的长度为0,变成一个阻塞的channel
if runtime.GOMAXPROCS(0) == 1 {
return 0
} // 如果GOMAXPROCS>1则使用非阻塞的workerChan
return 1
}()

梳理下流程:

1、首先判断下stopCh是否为nil,不为nil表示已经started了;

2、初始化wp.stopCh = make(chan struct{})stopCh是一个标识,用了struct{}不用bool,因为空结构体变量的内存占用大小为0,而bool类型内存占用大小为1,这样可以更加最大化利用我们服务器的内存空间;

3、设置workerChanPoolNew函数,然后可以在Get不到东西时,自动创建一个;如果单核cpu则让workerChan阻塞,否则,使用非阻塞,workerChan的长度设置为1;

4、启动一个goroutine,处理clean操作,在接收到退出信号,退出。

Stop
func (wp *workerPool) Stop() {
// 同start,stop也只能触发一次
if wp.stopCh == nil {
panic("BUG: workerPool wasn't started")
}
// 关闭stopCh
close(wp.stopCh)
// 将stopCh置为nil
wp.stopCh = nil // 停止所有的等待获取连接的workers
// 正在运行的workers,不需要等待他们退出,他们会在完成connection或mustStop被设置成true退出
wp.lock.Lock()
ready := wp.ready
// 循环将ready的workerChan置为nil
for i := range ready {
ready[i].ch <- nil
ready[i] = nil
}
wp.ready = ready[:0]
// 设置mustStop为true
wp.mustStop = true
wp.lock.Unlock()
}

梳理下流程:

1、判断stop只能被关闭一次;

2、关闭stopCh,设置stopChnil

3、停止所有的等待获取连接的workers,正在运行的workers,不需要等待他们退出,他们会在完成connectionmustStop被设置成true退出。

clean
func (wp *workerPool) clean(scratch *[]*workerChan) {
maxIdleWorkerDuration := wp.getMaxIdleWorkerDuration() // 清理掉最近最少使用的workers如果他们过了maxIdleWorkerDuration时间没有提供服务
criticalTime := time.Now().Add(-maxIdleWorkerDuration) wp.lock.Lock()
ready := wp.ready
n := len(ready) // 使用二分搜索算法找出最近可以被清除的worker
// 最后使用的workerChan 一定是放回队列尾部的。
l, r, mid := 0, n-1, 0
for l <= r {
mid = (l + r) / 2
if criticalTime.After(wp.ready[mid].lastUseTime) {
l = mid + 1
} else {
r = mid - 1
}
}
i := r
if i == -1 {
wp.lock.Unlock()
return
} // 将ready中i之前的的全部清除
*scratch = append((*scratch)[:0], ready[:i+1]...)
m := copy(ready, ready[i+1:])
for i = m; i < n; i++ {
ready[i] = nil
}
wp.ready = ready[:m]
wp.lock.Unlock() // 通知淘汰的workers停止
// 此通知必须位于wp.lock之外,因为ch.ch
// 如果有很多workers,可能会阻塞并且可能会花费大量时间
// 位于非本地CPU上。
tmp := *scratch
for i := range tmp {
tmp[i].ch <- nil
tmp[i] = nil
}
}

主要是清理掉最近最少使用的workers如果他们过了maxIdleWorkerDuration时间没有提供服务

getCh

获取一个workerChan

func (wp *workerPool) getCh() *workerChan {
var ch *workerChan
createWorker := false wp.lock.Lock()
ready := wp.ready
n := len(ready) - 1
// 如果ready为空
if n < 0 {
if wp.workersCount < wp.MaxWorkersCount {
createWorker = true
wp.workersCount++
}
} else {
// 不为空从ready中取一个
ch = ready[n]
ready[n] = nil
wp.ready = ready[:n]
}
wp.lock.Unlock() // 如果没拿到ch
if ch == nil {
if !createWorker {
return nil
}
// 从缓存中获取一个ch
vch := wp.workerChanPool.Get()
ch = vch.(*workerChan)
go func() {
// 具体的执行函数
wp.workerFunc(ch)
// 再放入到pool中
wp.workerChanPool.Put(vch)
}()
}
return ch
}

梳理下流程:

1、获取一个可执行的workerChan,如果ready中为空,并且workersCount没有达到最大值,增加workersCount数量,并且设置当前操作createWorker = true

2、ready中不为空,直接在ready获取一个;

3、如果没有获取到则在sync.pool中获取一个,之后再放回到pool中;

4、拿到了就启动一个workerFunc监听workerChan,处理具体的业务逻辑。

workerFunc
func (wp *workerPool) workerFunc(ch *workerChan) {
var c net.Conn var err error
// 监听workerChan
for c = range ch.ch {
if c == nil {
break
} // 具体的业务逻辑
...
c = nil // 释放workerChan
// 在mustStop的时候将会跳出循环
if !wp.release(ch) {
break
}
} wp.lock.Lock()
wp.workersCount--
wp.lock.Unlock()
} // 把Conn放入到channel中
func (wp *workerPool) Serve(c net.Conn) bool {
ch := wp.getCh()
if ch == nil {
return false
}
ch.ch <- c
return true
} func (wp *workerPool) release(ch *workerChan) bool {
// 修改 ch.lastUseTime
ch.lastUseTime = time.Now()
wp.lock.Lock()
// 如果需要停止,直接返回
if wp.mustStop {
wp.lock.Unlock()
return false
}
// 将ch放到ready中
wp.ready = append(wp.ready, ch)
wp.lock.Unlock()
return true
}

梳理下流程:

1、workerFunc会监听workerChan,并且在使用完workerChan归还到ready中;

2、Serve会把connection放入到workerChan中,这样workerFunc就能通过workerChan拿到需要处理的连接请求;

3、当workerFunc拿到的workerChannilwp.mustStop被设为了true,就跳出for循环。

panjf2000/ants

先看下示例

示例一

package main

import (
"fmt"
"sync"
"sync/atomic"
"time" "github.com/panjf2000/ants"
) func demoFunc() {
time.Sleep(10 * time.Millisecond)
fmt.Println("Hello World!")
} func main() {
defer ants.Release() runTimes := 1000 var wg sync.WaitGroup
syncCalculateSum := func() {
demoFunc()
wg.Done()
}
for i := 0; i < runTimes; i++ {
wg.Add(1)
_ = ants.Submit(syncCalculateSum)
}
wg.Wait()
fmt.Printf("running goroutines: %d\n", ants.Running())
fmt.Printf("finish all tasks.\n")
}

示例二

package main

import (
"fmt"
"sync"
"sync/atomic"
"time" "github.com/panjf2000/ants"
) var sum int32 func myFunc(i interface{}) {
n := i.(int32)
atomic.AddInt32(&sum, n)
fmt.Printf("run with %d\n", n)
} func main() {
var wg sync.WaitGroup
runTimes := 1000 // Use the pool with a method,
// set 10 to the capacity of goroutine pool and 1 second for expired duration.
p, _ := ants.NewPoolWithFunc(10, func(i interface{}) {
myFunc(i)
wg.Done()
})
defer p.Release()
// Submit tasks one by one.
for i := 0; i < runTimes; i++ {
wg.Add(1)
_ = p.Invoke(int32(i))
}
wg.Wait()
fmt.Printf("running goroutines: %d\n", p.Running())
fmt.Printf("finish all tasks, result is %d\n", sum)
if sum != 499500 {
panic("the final result is wrong!!!")
}
}
设计思路

整体的设计思路

梳理下思路:

1、先初始化缓存池的大小,然后处理任务事件的时候,一个task分配一个goWorker

2、在拿goWorker的过程中会存在下面集中情况;

  • 本地的缓存中有空闲的goWorker,直接取出;

  • 本地缓存没有就去sync.Pool,拿一个goWorker

3、如果缓存池满了,非阻塞模式直接返回nil,阻塞模式就循环去拿直到成功拿出一个;

4、同时也会定期清理掉过期的goWorker,通过sync.Cond唤醒其的阻塞等待;

5、对于使用完成的goWorker在使用完成之后重新归还到pool

具体的设计细节可参考,作者的文章Goroutine 并发调度模型深度解析之手撸一个高性能 goroutine 池

go-playground/pool

go-playground/pool会在一开始就启动

先放几个使用的demo

Per Unit Work

package main

import (
"fmt"
"time" "gopkg.in/go-playground/pool.v3"
) func main() { p := pool.NewLimited(10)
defer p.Close() user := p.Queue(getUser(13))
other := p.Queue(getOtherInfo(13)) user.Wait()
if err := user.Error(); err != nil {
// handle error
} // do stuff with user
username := user.Value().(string)
fmt.Println(username) other.Wait()
if err := other.Error(); err != nil {
// handle error
} // do stuff with other
otherInfo := other.Value().(string)
fmt.Println(otherInfo)
} func getUser(id int) pool.WorkFunc { return func(wu pool.WorkUnit) (interface{}, error) { // simulate waiting for something, like TCP connection to be established
// or connection from pool grabbed
time.Sleep(time.Second * 1) if wu.IsCancelled() {
// return values not used
return nil, nil
} // ready for processing... return "Joeybloggs", nil
}
} func getOtherInfo(id int) pool.WorkFunc { return func(wu pool.WorkUnit) (interface{}, error) { // simulate waiting for something, like TCP connection to be established
// or connection from pool grabbed
time.Sleep(time.Second * 1) if wu.IsCancelled() {
// return values not used
return nil, nil
} // ready for processing... return "Other Info", nil
}
}

Batch Work

package main

import (
"fmt"
"time" "gopkg.in/go-playground/pool.v3"
) func main() { p := pool.NewLimited(10)
defer p.Close() batch := p.Batch() // for max speed Queue in another goroutine
// but it is not required, just can't start reading results
// until all items are Queued. go func() {
for i := 0; i < 10; i++ {
batch.Queue(sendEmail("email content"))
} // DO NOT FORGET THIS OR GOROUTINES WILL DEADLOCK
// if calling Cancel() it calles QueueComplete() internally
batch.QueueComplete()
}() for email := range batch.Results() { if err := email.Error(); err != nil {
// handle error
// maybe call batch.Cancel()
} // use return value
fmt.Println(email.Value().(bool))
}
} func sendEmail(email string) pool.WorkFunc {
return func(wu pool.WorkUnit) (interface{}, error) { // simulate waiting for something, like TCP connection to be established
// or connection from pool grabbed
time.Sleep(time.Second * 1) if wu.IsCancelled() {
// return values not used
return nil, nil
} // ready for processing... return true, nil // everything ok, send nil, error if not
}
}

来看下实现

workUnit

workUnit作为channel信息进行传递,用来给work传递当前需要执行的任务信息。

// WorkUnit contains a single uint of works values
type WorkUnit interface { // 阻塞直到当前任务被完成或被取消
Wait() // 执行函数返回的结果
Value() interface{} // Error returns the Work Unit's error
Error() error // 取消当前的可执行任务
Cancel() // 判断当前的可执行单元是否被取消了
IsCancelled() bool
} var _ WorkUnit = new(workUnit) // workUnit contains a single unit of works values
type workUnit struct {
// 任务执行的结果
value interface{}
// 错误信息
err error
// 通知任务完成
done chan struct{}
// 需要执行的任务函数
fn WorkFunc
// 任务是会否被取消
cancelled atomic.Value
// 是否正在取消任务
cancelling atomic.Value
// 任务是否正在执行
writing atomic.Value
}
limitedPool
var _ Pool = new(limitedPool)

// limitedPool contains all information for a limited pool instance.
type limitedPool struct {
// 并发量
workers uint
// work的channel
work chan *workUnit
// 通知结束的channel
cancel chan struct{}
// 是否关闭的标识
closed bool
// 读写锁
m sync.RWMutex
} // 初始化一个pool
func NewLimited(workers uint) Pool { if workers == 0 {
panic("invalid workers '0'")
}
// 初始化pool的work数量
p := &limitedPool{
workers: workers,
}
// 初始化pool的操作
p.initialize() return p
} func (p *limitedPool) initialize() {
// channel的长度为work数量的两倍
p.work = make(chan *workUnit, p.workers*2)
p.cancel = make(chan struct{})
p.closed = false // fire up workers here
for i := 0; i < int(p.workers); i++ {
p.newWorker(p.work, p.cancel)
}
} // 将工作传递并取消频道到newWorker()以避免任何潜在的竞争状况
// 在p.work读写之间
func (p *limitedPool) newWorker(work chan *workUnit, cancel chan struct{}) {
go func(p *limitedPool) { var wu *workUnit defer func(p *limitedPool) {
// 捕获异常,结束掉异常的工作单元,并将其再次作为新的任务启动
if err := recover(); err != nil { trace := make([]byte, 1<<16)
n := runtime.Stack(trace, true) s := fmt.Sprintf(errRecovery, err, string(trace[:int(math.Min(float64(n), float64(7000)))])) iwu := wu
iwu.err = &ErrRecovery{s: s}
close(iwu.done) // 重新启动
p.newWorker(p.work, p.cancel)
}
}(p) var value interface{}
var err error
// 监听channel,读取内容
for {
select {
// channel中取出数据
case wu = <-work: // 防止channel 被关闭后读取到零值
if wu == nil {
continue
} // 单个和批量的cancellation这个都支持
if wu.cancelled.Load() == nil {
// 执行我们的业务函数
value, err = wu.fn(wu) wu.writing.Store(struct{}{}) // 如果WorkFunc取消了此工作单元,则需要再次检查
// 防止产生竞争条件
if wu.cancelled.Load() == nil && wu.cancelling.Load() == nil {
wu.value, wu.err = value, err // 执行完成,关闭当前channel
close(wu.done)
}
}
// 如果取消了,就退出
case <-cancel:
return
}
} }(p)
} // 放置一个执行的task到channel,并返回channel
func (p *limitedPool) Queue(fn WorkFunc) WorkUnit {
// 初始化一个workUnit类型的channel
w := &workUnit{
done: make(chan struct{}),
// 具体的执行函数
fn: fn,
} go func() {
p.m.RLock()
// 如果pool关闭的时候通知channel关闭
if p.closed {
w.err = &ErrPoolClosed{s: errClosed}
if w.cancelled.Load() == nil {
close(w.done)
}
p.m.RUnlock()
return
}
// 将channel传递给pool的work
p.work <- w p.m.RUnlock()
}() return w
}

梳理下流程:

1、首先初始化pool的大小;

2、然后根据pool的大小启动对应数量的worker,阻塞等待channel被塞入可执行函数;

3、然后可执行函数会被放入workUnit,然后通过channel传递给阻塞的worker

同样这里也提供了批量执行的方法

batch
// batch contains all information for a batch run of WorkUnits
type batch struct {
pool Pool
m sync.Mutex
// WorkUnit的切片
units []WorkUnit
// 结果集,执行完后的workUnit会更新其value,error,可以从结果集channel中读取
results chan WorkUnit
// 通知batch是否完成
done chan struct{}
closed bool
wg *sync.WaitGroup
} // 初始化Batch
func newBatch(p Pool) Batch {
return &batch{
pool: p,
units: make([]WorkUnit, 0, 4),
results: make(chan WorkUnit),
done: make(chan struct{}),
wg: new(sync.WaitGroup),
}
} // 将WorkFunc放入到WorkUnit中并保留取消和输出结果的参考。
func (b *batch) Queue(fn WorkFunc) { b.m.Lock() if b.closed {
b.m.Unlock()
return
}
// 返回一个WorkUnit
wu := b.pool.Queue(fn) // 放到WorkUnit的切片中
b.units = append(b.units, wu)
// 通过waitgroup进行goroutine的执行控制
b.wg.Add(1)
b.m.Unlock() // 执行任务
go func(b *batch, wu WorkUnit) {
wu.Wait()
// 将执行的结果写入到results中
b.results <- wu
b.wg.Done()
}(b, wu)
} // QueueComplete让批处理知道不再有排队的工作单元
// 以便在所有工作完成后可以关闭结果渠道。
// 警告:如果未调用此函数,则结果通道将永远不会耗尽,
// 但会永远阻止以获取更多结果。
func (b *batch) QueueComplete() {
b.m.Lock()
b.closed = true
close(b.done)
b.m.Unlock()
} // 取消批次的任务
func (b *batch) Cancel() { b.QueueComplete() b.m.Lock() // 一个个取消units,倒叙的取消
for i := len(b.units) - 1; i >= 0; i-- {
b.units[i].Cancel()
} b.m.Unlock()
} // 输出执行完成的结果集
func (b *batch) Results() <-chan WorkUnit {
// 启动一个协程监听完成的通知
// waitgroup阻塞直到所有的worker都完成退出
// 最后关闭channel
go func(b *batch) {
<-b.done
b.m.Lock()
// 阻塞直到上面waitgroup中的goroutine一个个执行完成退出
b.wg.Wait()
b.m.Unlock()
// 关闭channel
close(b.results)
}(b) return b.results
}

梳理下流程:

1、首先初始化Batch的大小;

2、然后Queue将一个个WorkFunc放入到WorkUnit中,执行,并将结果写入到results中,全部执行完成,调用QueueComplete,发送执行完成的通知;

3、Results会打印出所有的结果集,同时监听所有的worker执行完成,关闭channel,退出。

总结

控制goroutine数量一般使用两种方式:

  • 简单的场景使用sync+channel就可以了;

  • 复杂的场景可以使用goroutine pool

参考

【Golang 开发需要协程池吗?】https://www.zhihu.com/question/302981392

【来,控制一下 Goroutine 的并发数量】https://segmentfault.com/a/1190000017956396

【golang协程池设计】https://segmentfault.com/a/1190000018193161

【fasthttp中的协程池实现】https://segmentfault.com/a/1190000009133154

【panjf2000/ants】https://github.com/panjf2000/ants

【golang协程池设计】https://segmentfault.com/a/1190000018193161

本文作者:liz

本文链接https://boilingfrog.github.io/2021/04/14/控制goroutine的数量/

版权声明:本文为博主原创文章,遵循 CC 4.0 BY-SA 版权协议,转载请附上原文出处链接和本声明。