go中控制goroutine数量的方法_Golang_

 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() } 

 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--------") } 

 // 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 } 

 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 }() 

 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() } 

 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 } } 

 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 } 

 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 } 

 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!!!") } } 

 package main import ( "fmt" "time" "gopkg.in/go-playground/pool.v3" ) func main() { p := pool.NewLimited(10) defer p.Close() user := p.Queu
                
提示：
                    本文由神整理自网络，如有侵权请联系本站删除！
                    

                    本站声明： 

                    1、本站所有资源均来源于互联网，不保证100%完整、不提供任何技术支持； 

                    2、本站所发布的文章以及附件仅限用于学习和研究目的;不得将用于商业或者非法用途；否则由此产生的法律后果，本站概不负责！
                
                

                

                                            
上一篇：解决go在函数退出后子协程的退出问题_Golang_
                                                                
下一篇：go语言通过反射创建结构体、赋值、并调用对应的操作_Golang_