登录
首页 >  Golang >  Go教程

go中控制goroutine数量的方法

来源:脚本之家

时间:2023-02-16 15:24:02 126浏览 收藏

IT行业相对于一般传统行业,发展更新速度更快,一旦停止了学习,很快就会被行业所淘汰。所以我们需要踏踏实实的不断学习,精进自己的技术,尤其是初学者。今天golang学习网给大家整理了《go中控制goroutine数量的方法》,聊聊goroutine、go控制、数量,我们一起来看看吧!

前言

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 

通过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就可以,其他复杂的可以考虑使用第三方的协程池库。

panjf2000/ants

go-playground/pool

Jeffail/tunny

几个开源的线程池的设计

fasthttp中的协程池实现

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

1、按需增长goroutine数量,有一个最大值,同时监听channel,Server会把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 1则使用非阻塞的workerChan
 return 1
}()

梳理下流程:

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

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

3、设置workerChanPool的New函数,然后可以在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 

梳理下流程:

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

2、关闭stopCh,设置stopCh为nil;

3、停止所有的等待获取连接的workers,正在运行的workers,不需要等待他们退出,他们会在完成connection或mustStop被设置成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 

主要是清理掉最近最少使用的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 

梳理下流程:

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 

梳理下流程:

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

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

3、当workerFunc拿到的workerChan为nil或wp.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 

示例二

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 

设计思路

整体的设计思路

梳理下思路:

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 

来看下实现

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 

梳理下流程:

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 = 0; i-- {
  b.units[i].Cancel()
 }

 b.m.Unlock()
}

// 输出执行完成的结果集
func (b *batch) 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

到这里,我们也就讲完了《go中控制goroutine数量的方法》的内容了。个人认为,基础知识的学习和巩固,是为了更好的将其运用到项目中,欢迎关注golang学习网公众号,带你了解更多关于golang的知识点!

声明:本文转载于:脚本之家 如有侵犯,请联系study_golang@163.com删除
相关阅读
更多>
最新阅读
更多>
课程推荐
更多>