高性能协程池ants源码剖析

2023年3月6日 · 阅读需 13 分钟

ahKevinXy

作者

简介

ants是什么

ants是一个高性能的 goroutine 池，实现了对大规模 goroutine 的调度管理、goroutine 复用，允许使用者在开发并发程序的时候限制 goroutine 数量，复用资源，达到更高效执行任务的效果。

功能特点

自动调度海量的 goroutines，复用 goroutines
定期清理过期的 goroutines，进一步节省资源
提供了大量有用的接口：任务提交、获取运行中的 goroutine 数量、动态调整 Pool 大小、释放 Pool、重启 Pool
优雅处理 panic，防止程序崩溃
资源复用，极大节省内存使用量；在大规模批量并发任务场景下比原生 goroutine 并发具有更高的性能
非阻塞机制

ants核心概念

Pool ：协程池
WorkerArray：Pool池中的worker队列，存放所有的goWorker
goWorker：运行任务的实际执行者，它启动一个 goroutine 来接受任务并执行函数调用。

Pool协程池

Pool结构

Ants 提供了两种Pool结构：Pool和PoolWithFunc ；但两者逻辑大致一样，本文着重介绍Pool的结构

 // Pool accepts the tasks from client, it limits the total of goroutines to a given number by recycling goroutines.
type Pool struct {
 // capacity of the pool, a negative value means that the capacity of pool is limitless, an infinite pool is used to
 // avoid potential issue of endless blocking caused by nested usage of a pool: submitting a task to pool
 // which submits a new task to the same pool.
  // 协程池的容量
 capacity int32

 // running is the number of the currently running goroutines.
  // 正在运行的goroutines的数量
 running int32

 // lock for protecting the worker queue.
  // 锁，自旋锁，保护队列
 lock sync.Locker

 // workers is a slice that store the available workers.
  // 存放池中所有的worker,workerArray包含可用workers队列和过期workers队列，只会从可用workers队列中取可用worker
 workers workerArray

 // state is used to notice the pool to closed itself.
  // 记录池子的状态 （关闭、开启）
 state int32

 // cond for waiting to get an idle worker.
  // 条件变量
 cond *sync.Cond

 // workerCache speeds up the obtainment of a usable worker in function:retrieveWorker.
  // worker 对象池
 workerCache sync.Pool

 // waiting is the number of goroutines already been blocked on pool.Submit(), protected by pool.lock
  //阻塞等待的任务量
 waiting int32
 // 清道夫，定时清理workerarray 队列中过期的worker
 purgeDone int32
 stopPurge context.CancelFunc
 // 定时器 更新pool中now的字段
 ticktockDone int32
 stopTicktock context.CancelFunc

 now atomic.Value
 // 需要自定义加载的配置
 options *Options
}

Pool创建

// NewPool generates an instance of ants pool.
func NewPool(size int, options ...Option) (*Pool, error) {
 opts := loadOptions(options...) // 加载自定义的options中的配置

 if size <= 0 {
  size = -1
 }

 if !opts.DisablePurge {// 当 DisablePurge 为 true 时，worker 不会被清除并且是驻留的。
  if expiry := opts.ExpiryDuration; expiry < 0 {
   return nil, ErrInvalidPoolExpiry
  } else if expiry == 0 {
   opts.ExpiryDuration = DefaultCleanIntervalTime // 默认间隔时间1s
  }
 }

 if opts.Logger == nil {
  opts.Logger = defaultLogger
 }

 p := &Pool{
  capacity: int32(size),
  lock:     syncx.NewSpinLock(),//自旋锁
  options:  opts,
 }
 p.workerCache.New = func() interface{} { //sync.pool 初始化
  return &goWorker{
   pool: p,
   task: make(chan func(), workerChanCap),
  }
 }
 if p.options.PreAlloc {
  if size == -1 {
   return nil, ErrInvalidPreAllocSize
  }
  p.workers = newWorkerArray(loopQueueType, size) //循环队列
 } else {
  p.workers = newWorkerArray(stackType, 0) //数组
 }

 p.cond = sync.NewCond(p.lock) // sync.cond初始化

 p.goPurge()
 p.goTicktock()

 return p, nil
}

自旋锁SpinLock(重点)

思考：如何设计一种自旋锁，设计自旋锁时需要注意什么？

spinLock是基于CAS机制和指数退避算法实现的一种自旋锁

package sync

import (
 "runtime"
 "sync"
 "sync/atomic"
)

type spinLock uint32 // 实现sync.Locker接口

const maxBackoff = 16 //最大的回避次数

func (sl *spinLock) Lock() {
 backoff := 1
  // 基于CAS机制，尝试获取锁，且使用指数退避算法来提供获取锁的成功率
 for !atomic.CompareAndSwapUint32((*uint32)(sl), 0, 1) {
  // Leverage the exponential backoff algorithm, see https://en.wikipedia.org/wiki/Exponential_backoff.
  for i := 0; i < backoff; i++ {
     //runtime.Gosched()函数功能：使当前goroutine让出CPU时间片（“回避”），让其他的goroutine获得执行的机会。当前的goroutine会在未来的某个时间点继续运行。
    //注意：当一个goroutine发生阻塞，Go会自动地把与该goroutine处于同一系统线程的其他goroutines转移到另一个系统线程上去，以使这些goroutines不阻塞（从GMP模型角度来说，就是当与P绑定的M发生阻塞，P就与其解绑，然后与另一个空闲的M进行绑定 或者 去创建一个M进行绑定）。
   runtime.Gosched()
  }
  if backoff < maxBackoff {
   backoff <<= 1
  }
 }
}

func (sl *spinLock) Unlock() {
  //原子操作，并发安全
 atomic.StoreUint32((*uint32)(sl), 0)
}

// NewSpinLock instantiates a spin-lock.
func NewSpinLock() sync.Locker {
 return new(spinLock)
}

sync.Locker

设计锁时必须实现该接口中的方法

// A Locker represents an object that can be locked and unlocked.
type Locker interface {
 Lock()
 Unlock()
}

关键知识点

sync.Locker接口
指数退避算法
atomic 原子包中的方法了解
runtime.Gosched()

任务执行

goWorker的结构

// goWorker is the actual executor who runs the tasks,
// it starts a goroutine that accepts tasks and
// performs function calls.
//goWorker 是运行任务的实际执行者，它启动一个 goroutine 来接受任务并执行函数调用。
type goWorker struct {
 // pool who owns this worker.
 pool *Pool // 拥有worker的协议池

 // task is a job should be done.
 task chan func() // 需要执行的任务，注意：该chan 可能是缓存区或者非缓存区，如果是多核的话，缓存区的大小是1

 // recycleTime will be updated when putting a worker back into queue.
 recycleTime time.Time // 回收时间
}

goWoker的初始化

goWorker 是sync.pool 对象池创建的；初始化在Pool创建时

p.workerCache.New = func() interface{} {
  return &goWorker{
   pool: p,
   task: make(chan func(), workerChanCap),
  }
 }

workerChanCap的值

// workerChanCap determines whether the channel of a worker should be a buffered channel
 // to get the best performance. Inspired by fasthttp at
 // https://github.com/valyala/fasthttp/blob/master/workerpool.go#L139
 workerChanCap = func() int {
  // Use blocking channel if GOMAXPROCS=1.
  // This switches context from sender to receiver immediately,
  // which results in higher performance (under go1.5 at least).
  if runtime.GOMAXPROCS(0) == 1 {
   return 0
  }

  // Use non-blocking workerChan if GOMAXPROCS>1,
  // since otherwise the sender might be dragged down if the receiver is CPU-bound.
  return 1
 }()

Task 任务提交

func (p *Pool) Submit(task func()) error {
 if p.IsClosed() { // 前置检查 协程池是否关闭
  return ErrPoolClosed
 }
 var w *goWorker
 if w = p.retrieveWorker(); w == nil { //获取一个可用的worker取执行任务
  return ErrPoolOverload
 }
 w.task <- task
 return nil
}

retrieveWorker （可用worker获取）

// retrieveWorker returns an available worker to run the tasks.
func (p *Pool) retrieveWorker() (w *goWorker) {
 spawnWorker := func() { // 使用sync.pool 创建worker
  w = p.workerCache.Get().(*goWorker)
  w.run()
 }

 p.lock.Lock() // 自旋锁 加锁

 w = p.workers.detach() // 尝试从worker池子中获取可用的worker，注：任务执行完后，会回收worker 以便下次使用
 if w != nil { // first try to fetch the worker from the queue  获取到返回
  p.lock.Unlock()
 } else if capacity := p.Cap(); capacity == -1 || capacity > p.Running() { // 判断正在运行的worker 是否超过 pool协程词设置的容量大小，注 capacity=-1 表示池子容量无限大
  // if the worker queue is empty and we don't run out of the pool capacity,
  // then just spawn a new worker goroutine.
  p.lock.Unlock()
  spawnWorker() // 创建worker
 } else { // otherwise, we'll have to keep them blocked and wait for at least one worker to be put back into pool.
  if p.options.Nonblocking { // 判断协程池是否 是非阻塞模式 ，如果是非阻塞模式下就直接返回
   p.lock.Unlock()
   return
  }
 retry: // 阻塞模式下的逻辑
    // 判断阻塞的任务数量是否超过了设置的最大阈值，如果超过直接返回
  if p.options.MaxBlockingTasks != 0 && p.Waiting() >= p.options.MaxBlockingTasks {
   p.lock.Unlock()
   return
  }
  p.addWaiting(1) // 阻塞任务数量加1
  p.cond.Wait() // block and wait for an available worker
  p.addWaiting(-1) // 获取到可用的worker后，阻塞数量减一

  if p.IsClosed() { // check pool池是否关闭
   p.lock.Unlock()
   return
  }

  var nw int
  if nw = p.Running(); nw == 0 { // 如果正在执行的worker数量为0时，需要重新创建woker
   p.lock.Unlock()
   spawnWorker()
   return
  }
  if w = p.workers.detach(); w == nil { //从workerArray中获取可用的worker
   if nw < p.Cap() { // 获取不到，判断正在运行的goroutines的数量是否超过协层池的容量，没有就创建
    p.lock.Unlock()
    spawnWorker()
    return
   }
   goto retry // goto 重试阻塞模式下获取可用worker的逻辑
  }
  p.lock.Unlock()
 }
 return
}

WorkerArray 工作池的结构

type workerArray interface {
 len() int // 长度
 isEmpty() bool // 是否为空
 insert(worker *goWorker) error // 插入
 detach() *goWorker // 从WorkerArray获取可用的worker
 retrieveExpiry(duration time.Duration) []*goWorker  //清道夫调用pool.worker中的此方法来清理pool.workers中的过期worker
 reset() // 重置，清空WorkerArray中所有的Worker
}

workerArray 接口的实现workerStack 和loopQueue

任务执行

// run starts a goroutine to repeat the process
// that performs the function calls.
func (w *goWorker) run() {
   w.pool.addRunning(1) // pool的running 加 一
   go func() {
      defer func() {
         w.pool.addRunning(-1)
         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 {
         if f == nil {
            return
         }
         f() // 任务执行
         if ok := w.pool.revertWorker(w); !ok { // 回收woker
            return
         }
      }
   }()
}

revertWorkerworker回收

逻辑简单：1、往workerArray 队列中插入；2、通知正在阻塞获取worker的goroutines

// revertWorker puts a worker back into free pool, recycling the goroutines.
func (p *Pool) revertWorker(worker *goWorker) bool {
 if capacity := p.Cap(); (capacity > 0 && p.Running() > capacity) || p.IsClosed() {
  p.cond.Broadcast()
  return false
 }
 worker.recycleTime = p.nowTime()
 p.lock.Lock()

 // To avoid memory leaks, add a double check in the lock scope.
 // Issue: https://github.com/panjf2000/ants/issues/113
 if p.IsClosed() {
  p.lock.Unlock()
  return false
 }

 err := p.workers.insert(worker)
 if err != nil {
  p.lock.Unlock()
  return false
 }

 // Notify the invoker stuck in 'retrieveWorker()' of there is an available worker in the worker queue.
 p.cond.Signal()
 p.lock.Unlock()
 return true
}

定时清理过期的worker

func (p *Pool) goPurge() {
 if p.options.DisablePurge {
  return
 }

 // Start a goroutine to clean up expired workers periodically.
 var ctx context.Context
 ctx, p.stopPurge = context.WithCancel(context.Background())
 go p.purgeStaleWorkers(ctx)
}

// purgeStaleWorkers clears stale workers periodically, it runs in an individual goroutine, as a scavenger.
func (p *Pool) purgeStaleWorkers(ctx context.Context) {
 ticker := time.NewTicker(p.options.ExpiryDuration)

 defer func() {
  ticker.Stop()
  atomic.StoreInt32(&p.purgeDone, 1)
 }()

 for {
  select {
  case <-ctx.Done():
   return
  case <-ticker.C:
  }

  if p.IsClosed() {
   break
  }

  p.lock.Lock()
  expiredWorkers := p.workers.retrieveExpiry(p.options.ExpiryDuration)
  p.lock.Unlock()

  // Notify obsolete workers to stop.
  // This notification must be outside the p.lock, since w.task
  // may be blocking and may consume a lot of time if many workers
  // are located on non-local CPUs.
  for i := range expiredWorkers {
   expiredWorkers[i].task <- nil
   expiredWorkers[i] = nil
  }

  // There might be a situation where all workers have been cleaned up(no worker is running),
  // or another case where the pool capacity has been Tuned up,
  // while some invokers still get stuck in "p.cond.Wait()",
  // then it ought to wake all those invokers.
  if p.Running() == 0 || (p.Waiting() > 0 && p.Free() > 0) {
   p.cond.Broadcast()
  }
 }
}

ants源码自旋锁的设计、sync.cond 条件变量、sync.pool 对象池，原子操作，channel通信等

简介​

ants是什么​

功能特点​

ants核心概念​

Pool协程池​

Pool结构​

Pool创建​

自旋锁SpinLock(重点)​

关键知识点​

任务执行​

goWorker的结构​

goWoker的初始化​

workerChanCap的值​

Task 任务提交​

retrieveWorker （可用worker获取）​

WorkerArray 工作池的结构​

任务执行​

定时清理过期的worker​

简介

ants是什么

功能特点

ants核心概念

Pool协程池

Pool结构

Pool创建

自旋锁SpinLock(重点)

关键知识点

任务执行

goWorker的结构

goWoker的初始化

workerChanCap的值

Task 任务提交

retrieveWorker （可用worker获取）

WorkerArray 工作池的结构

任务执行

定时清理过期的worker