grpool goroutine池协程管理

Serepta ·

更新时间:2024-05-19

· 1023 次阅读

前言

名词概念

使用示例

踩坑之旅

常犯的错误

分析原因

使用grpool

错误代码

正确代码

总结

前言

goroutine协程非常轻量级，这也是为什么go支持高并发，但是goroutine频繁创建销毁对GC的压力比较大。

grpool的作用就是复用goroutine，减少频繁创建销毁的性能消耗。

名词概念

Pool: goroutine池，用于管理若干可复用的goroutine协程资源

Worker: 池对象中参与任务执行的goroutine，一个worker可以执行若干个job，直到队列中再无等待的job

Job：添加到池对象的任务队列中等待执行的任务，是一个func()方法，一个job同时只能被一个worker获取并执行。

使用示例

使用默认的协程池，限制100个协程执行1000个任务

pool.Size() 获得当前工作的协程数量

pool.Jobs() 获得当前池中待处理的任务数量

package main
import (
   "fmt"
   "github.com/gogf/gf/os/grpool"
   "github.com/gogf/gf/os/gtimer"
   "sync"
   "time"
)
func main() {
   pool := grpool.New(100)
   //添加1千个任务
   for i := 0; i < 1000; i++ {
      _ = pool.Add(job)
   }
   fmt.Println("worker:", pool.Size()) //当前工作的协程数量
   fmt.Println("jobs:", pool.Jobs())   //当前池中待处理的任务数量
   gtimer.SetInterval(time.Second, func() {
      fmt.Println("worker:", pool.Size()) //当前工作的协程数
      fmt.Println("jobs:", pool.Jobs())   //当前池中待处理的任务数
   })
   //阻止进程结束
   select {}
}
//任务方法
func job() {
   time.Sleep(time.Second)
}

打印结果：

是不是灰常简单~

踩坑之旅

一个简单的场景，请使用协程打印0~9。

常犯的错误

大家看下面的代码有没有问题，请预测一下打印结果。

wg := sync.WaitGroup{}
for i := 0; i < 9; i++ {
   wg.Add(1)
   go func() {
      fmt.Println(i)
      wg.Done()
   }()
}
wg.Wait()

不用着急看答案

猜一下打印结果是什么

打印结果：

分析原因

对于异步线程/协程来讲，函数进行异步执行注册时，该函数并未真正开始执行(注册时只在goroutine的栈中保存了变量i的内存地址)，而一旦开始执行时函数才会去读取变量i的值，而这个时候变量i的值已经自增到了9。

正确写法：

wg := sync.WaitGroup{}
for i := 0; i < 9; i++ {
   wg.Add(1)
   go func(v int) {
      fmt.Println(v)
      wg.Done()
   }(i)
}
wg.Wait()

打印结果：

使用grpool

使用grpool和使用go一样，都需要把当前变量i的值赋值给一个不会改变的临时变量，在函数中使用该临时变量而不是直接使用变量i。

错误代码

wg := sync.WaitGroup{}
for i := 0; i < 9; i++ {
   wg.Add(1)
   _ = grpool.Add(func() {
      fmt.Println(i) //打印结果都是9
      wg.Done()
   })
}
wg.Wait()

打印结果：

正确代码

wg := sync.WaitGroup{}
for i := 0; i < 9; i++ {
   wg.Add(1)
   v := i //grpoll.add() 的参数只能是不带参数的匿名函数 因此只能以设置临时变量的方式赋值
   _ = grpool.Add(func() {
      fmt.Println(v)
      wg.Done()
   })
}
wg.Wait()

打印结果：