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就可以，其他复杂的可以考虑使用第三方的协程池库。
几个开源的线程池的设计
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的知识点！
-
221 收藏
-
433 收藏
-
106 收藏
-
451 收藏
-
331 收藏
-
108 收藏
-
367 收藏
-
419 收藏
-
234 收藏
-
155 收藏
-
457 收藏
-
- 前端进阶之JavaScript设计模式
- 设计模式是开发人员在软件开发过程中面临一般问题时的解决方案，代表了最佳的实践。本课程的主打内容包括JS常见设计模式以及具体应用场景，打造一站式知识长龙服务，适合有JS基础的同学学习。
- 立即学习 542次学习
-
- GO语言核心编程课程
- 本课程采用真实案例，全面具体可落地，从理论到实践，一步一步将GO核心编程技术、编程思想、底层实现融会贯通，使学习者贴近时代脉搏，做IT互联网时代的弄潮儿。
- 立即学习 507次学习
-
- 简单聊聊mysql8与网络通信
- 如有问题加微信：Le-studyg；在课程中，我们将首先介绍MySQL8的新特性，包括性能优化、安全增强、新数据类型等，帮助学生快速熟悉MySQL8的最新功能。接着，我们将深入解析MySQL的网络通信机制，包括协议、连接管理、数据传输等，让
- 立即学习 497次学习
-
- JavaScript正则表达式基础与实战
- 在任何一门编程语言中,正则表达式,都是一项重要的知识,它提供了高效的字符串匹配与捕获机制,可以极大的简化程序设计。
- 立即学习 487次学习
-
- 从零制作响应式网站—Grid布局
- 本系列教程将展示从零制作一个假想的网络科技公司官网，分为导航，轮播，关于我们，成功案例，服务流程，团队介绍，数据部分，公司动态，底部信息等内容区块。网站整体采用CSSGrid布局，支持响应式，有流畅过渡和展现动画。
- 立即学习 484次学习