golang ants 源码分析
结构图
poolwithfunc与pool相差不大，这里我们只分析ants默认pool的流程
| 文件 | 作用 |
|---|---|
| ants.go | 定义常量、errors显示、默认建一个大小为2147483647的goroutine池、封装一些方便用户操作查看goroutine池的函数 |
| options.go | goroutine池的相关配置 |
| pool.go | 普通pool(不绑定特定函数)的创建以及对pool相关的操作 |
| pool_func.go | 创建绑定某个特定函数的pool以及对pool相关的操作 |
| worker.go | goworker的struct(其他语言中的类)、run(其他语言中的方法) |
| worker_array.go | 一个worker_array的接口和一个能返回实现该接口的函数 |
| worker_func.go | 略 |
| worker_loop_queue.go | 略 |
| worker_stack.go | workerStack(struct)实现worker_array中的所有接口 |
| spinlock.go | 锁相关 |
关键结构
type Pool struct
type Pool struct {
capacity int32 // 容量
running int32 // 正在运行的数量
lock sync.Locker //定义一个锁 用以支持 Pool 的同步操作
workers workerArray // workers 一个接口 存放可循环利用的Work(goroutine)的相关信息
// type workerArray interface {
// len() int
// isEmpty() bool
// insert(worker *goWorker) error
// detach() *goWorker
// retrieveExpiry(duration time.Duration) []*goWorker
// reset()
// }
state int32 //记录池子的状态(关闭，开启)
cond *sync.Cond // 条件变量
workerCache sync.Pool // golang原始池子 使用sync.Pool对象池管理和创建worker对象，提升性能
blockingNum int // 阻塞等待的任务数量；
stopHeartbeat chan struct{} //一个空结构体的通道,仅用于接收标志
options *Options // 用于配置pool的options指针
}
- func (p *Pool) purgePeriodically() //定期清理过期worker任务
- func (p *Pool) Submit(task func()) error //提交func任务与worker绑定进行运行
- func (p *Pool) Running() int //有多少个运行的worker
- func (p *Pool) Free() int //返回空闲的worker数量
- func (p *Pool) Cap() int // 返回pool的容量
- ......
- func (p *Pool) retrieveWorker() (w *goWorker) //返回一个worker
workerArray
type workerArray interface {
len() int // worker的数量
isEmpty() bool // worker是否为0
insert(worker *goWorker) error //将执行完的worker(goroutine)放回
detach() *goWorker // 获取worker
retrieveExpiry(duration time.Duration) []*goWorker //取出所有的过期 worker；
reset() // 重置
}
workerStack
type workerStack struct {
items []*goWorker //空闲的worker
expiry []*goWorker //过期的worker
size int
}
下面是对接口workerArray的实现
func (wq *workerStack) len() int {
return len(wq.items)
}
func (wq *workerStack) isEmpty() bool {
return len(wq.items) == 0
}
func (wq *workerStack) insert(worker *goWorker) error {
wq.items = append(wq.items, worker)
return nil
}
//返回items中最后一个worker
func (wq *workerStack) detach() *goWorker {
l := wq.len()
if l == 0 {
return nil
}
w := wq.items[l-1]
wq.items[l-1] = nil // avoid memory leaks
wq.items = wq.items[:l-1]
return w
}
func (wq *workerStack) retrieveExpiry(duration time.Duration) []*goWorker {
n := wq.len()
if n == 0 {
return nil
}
expiryTime := time.Now().Add(-duration) //过期时间=现在的时间-1s
index := wq.binarySearch(0, n-1, expiryTime)
wq.expiry = wq.expiry[:0]
if index != -1 {
wq.expiry = append(wq.expiry, wq.items[:index+1]...) //因为以后进先出的模式去worker 所有过期的woker这样wq.items[:index+1]取
m := copy(wq.items, wq.items[index+1:])
for i := m; i < n; i++ { //m是存活的数量 下标为m之后的元素全部置为nil
wq.items[i] = nil
}
wq.items = wq.items[:m] //抹除后面多余的内容
}
return wq.expiry
}
// 二分法查询过期的worker
func (wq *workerStack) binarySearch(l, r int, expiryTime time.Time) int {
var mid int
for l <= r {
mid = (l + r) / 2
if expiryTime.Before(wq.items[mid].recycleTime) {
r = mid - 1
} else {
l = mid + 1
}
}
return r
}
func (wq *workerStack) reset() {
for i := 0; i < wq.len(); i++ {
wq.items[i].task <- nil //worker的任务置为nil
wq.items[i] = nil //worker置为nil
}
wq.items = wq.items[:0] //items置0
}
流程分析
创建pool
func NewPool(size int, options ...Option) (*Pool, error) {
opts := loadOptions(options...) // 导入配置
根据不同项进行配置此处省略
p := &Pool{
capacity: int32(size),
lock: internal.NewSpinLock(),
stopHeartbeat: make(chan struct{}, 1), //开一个通道用于接收一个停止标志
options: opts,
}
p.workerCache.New = func() interface{} {
return &goWorker{
pool: p,
task: make(chan func(), workerChanCap),
}
}
p.workers = newWorkerArray(stackType, 0)
p.cond = sync.NewCond(p.lock)
go p.purgePeriodically()
return p, nil
}
提交任务(将worker于func绑定)
func (p *Pool) retrieveWorker() (w *goWorker) {
spawnWorker := func() {
w = p.workerCache.Get().(*goWorker)
w.run()
}
p.lock.Lock()
w = p.workers.detach() // 获取列表中最后一个worker
if w != nil { // 取出来的话直接解锁
p.lock.Unlock()
} else if capacity := p.Cap(); capacity == -1 || capacity > p.Running() { //没取到但是容量为无限大或者容量未满
p.lock.Unlock()
spawnWorker() //开一个新的worker
} else { // 没取到 而且容量已经满了
if p.options.Nonblocking { //默认为False
p.lock.Unlock()
return
}
retry:
xxxx
goto retry
xxxx
p.lock.Unlock()
}
return
}
goworker的运行
func (w *goWorker) run() {
w.pool.incRunning() //增加正在运行的worker数量
go func() {
defer func() {
w.pool.decRunning()
w.pool.workerCache.Put(w)
if p := recover(); p != nil {
if ph := w.pool.options.PanicHandler; ph != nil {
ph(p)
} else {
w.pool.options.Logger.Printf("worker exits from a panic: %v\n", p)
var buf [4096]byte
n := runtime.Stack(buf[:], false)
w.pool.options.Logger.Printf("worker exits from panic: %s\n", string(buf[:n]))
}
}
// Call Signal() here in case there are goroutines waiting for available workers.
w.pool.cond.Signal()
}()
for f := range w.task { //阻塞接受task
if f == nil {
return
}
f() //执行函数
if ok := w.pool.revertWorker(w); !ok { // 将goworker放回items中
return
}
}
}()
}
__EOF__
