Skip to main content
 首页 » 编程设计

go中控制goroutine数量

2022年07月19日109java哥

控制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 版权协议,转载请附上原文出处链接和本声明。


本文参考链接:https://www.cnblogs.com/ricklz/p/14656691.html