Ants

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

功能

• 自动调度海量的 goroutines，复用 goroutines
• 定期清理过期的 goroutines，进一步节省资源
• 提供了大量实用的接口：任务提交、获取运行中的 goroutine 数量、动态调整 Pool 大小、释放 Pool、重启 Pool 等
• 优雅处理 panic，防止程序崩溃
• 资源复用，极大节省内存使用量；在大规模批量并发任务场景下甚至可能比 Go 语言的无限制 goroutine 并发具有更高的性能
• 非阻塞机制
• 预分配内存 (环形队列，可选)

工作流程

▼
mermaid
复制代码
flowchart TD
    A[初始化 goroutine 池] -->|提交一个任务| B[工作池]
    
    C{池中是否有可用的 worker?} -->|是| G[取出 worker 来执行任务]
    C -->|否| D{工作池容量是否用完}
    
    B --> C
    
    D -->|是| E{工作池是否为非阻塞模式?}
    D -->|否| F[新启动一个 worker 来执行任务]
    
    E -->|是| H[直接返回 nil]
    E -->|否| I[阻塞等待可用的 worker]
    
    I --> G
    
    F --> B
    G --> |完成任务后将 worker 放回工作池| B

快速开始

安装

▼
bash
复制代码
go get -u github.com/panjf2000/ants/v2

使用

• Default Pool

▼
go
复制代码
package main

import (
    "fmt"
    "sync"
    
    "github.com/panjf2000/ants/v2"
)

func main() {
    var wg sync.WaitGroup
    
    // Submit tasks to the default pool
    for i := 0; i < 10; i++ {
        wg.Add(1)
        ants.Submit(func() {
            // Your task here
            fmt.Println("Task is running")
            wg.Done()
        })
    }
    
    wg.Wait()
    fmt.Printf("Running goroutines: %d\n", ants.Running())
    
    // Release the default pool when finished
    ants.Release()
}

• Basic Pool

▼
go
复制代码
package main

import (
    "fmt"
    "sync"
    
    "github.com/panjf2000/ants/v2"
)

func main() {
    var wg sync.WaitGroup
    
    // Create a pool with capacity of 10 workers
    pool, _ := ants.NewPool(10)
    defer pool.Release()
    
    // Submit tasks
    for i := 0; i < 20; i++ {
        wg.Add(1)
        pool.Submit(func() {
            fmt.Println("Task is running")
            wg.Done()
        })
    }
    
    wg.Wait()
    fmt.Printf("Running goroutines: %d\n", pool.Running())
}

• Pool with Function

▼
go
复制代码
package main

import (
    "fmt"
    "sync"
    
    "github.com/panjf2000/ants/v2"
)

func main() {
    var wg sync.WaitGroup
    
    // Create a pool with capacity of 10 workers
    // that runs the same function for all tasks
    pool, _ := ants.NewPoolWithFunc(10, func(i interface{}) {
        n := i.(int)
        fmt.Printf("Processing task: %d\n", n)
        wg.Done()
    })
    defer pool.Release()
    
    // Submit tasks
    for i := 1; i <= 20; i++ {
        wg.Add(1)
        pool.Invoke(i) // Pass the argument to the function
    }
    
    wg.Wait()
    fmt.Printf("Running goroutines: %d\n", pool.Running())
}

• Type-Safe Generic Pool

▼
go
复制代码
package main

import (
    "fmt"
    "sync"
    
    "github.com/panjf2000/ants/v2"
)

func main() {
    var wg sync.WaitGroup
    
    // Create a generic pool with capacity of 10 workers
    // that runs the same function for all tasks of type int
    pool, _ := ants.NewPoolWithFuncGeneric(10, func(n int) {
        fmt.Printf("Processing task: %d\n", n)
        wg.Done()
    })
    defer pool.Release()
    
    // Submit tasks
    for i := 1; i <= 20; i++ {
        wg.Add(1)
        pool.Invoke(i) // Type-safe argument
    }
    
    wg.Wait()
    fmt.Printf("Running goroutines: %d\n", pool.Running())
}

核心概念

池化

Go 的 goroutine 轻量级，但创建和销毁大量 goroutine 仍然可能影响性能和内存使用。goroutine 池通过以下方式解决这个问题：

• 管理和限制活动 goroutine 的数量
• 通过回收 goroutine 减少内存分配开销
• 提供受控的并发机制
• 提升高吞吐量场景下的性能

▼
mermaid
复制代码
classDiagram
	direction TB
class poolCommon {
    +int32 capacity
    +int32 running
    +sync.Locker lock
    +workerQueue workers
    +int32 state
    +*sync.Cond cond
    +chan struct allDone
    +*sync.Once once
    +sync.Pool workerCache
    +int32 waiting
    +int32 purgeDone
    +context.Context purgeCtx
    +context.CancelFunc stopPurge
    +atomic.Value now
    +*Options options

    +purgeStaleWorkers()
    +ticktock()
    +goPurge()
    +goTicktock()
    +nowTime() time.Time
    +Running() int
    +Free() int
    +Waiting() int
    +Cap() int
    +Tune(size int)
    +IsClosed()
    +Release()
    +ReleaseTimeout(timeout time.Duration) error
    +Reboot()
    +addRunning(delta int) int
    +addWaiting(delta int)
    +retrieveWorker() (worker, error)
    +revertWorker(worker) bool
}

class Pool {
    +Submit(task func()) error
}

Pool --|> poolCommon : embeds

class MultiPool {
	- []*Pool pools
	- uint32 index
	- int32 state
	- LoadBalancingStrategy lbs
	+ Submit(task func()) error
	+ Running() int
	+ RunningByIndex(idx int) (int, error)
	+ Free() int
	+ FreeByIndex(idx int) (int, error)
	+ Waiting() int
	+ WaitingByIndex(idx int) (int, error)
	+ Cap() int
	+ Tune(size int)
	+ IsClosed() bool
	+ ReleaseTimeout(timeout time.Duration) error
	+ Reboot()
}

class Pool {
	+ Submit(task func()) error
	+ Running() int
	+ Free() int
	+ Waiting() int
	+ Cap() int
	+ Tune(size int)
	+ ReleaseTimeout(timeout time.Duration) error
	+ Reboot()
}

MultiPool --> Pool : contains multiple

Pool

在 ants 中，提供了 Basic Pool 与 Multi Pool 两种实现

• Basic Pool

  保持对 goroutine 最大数量的控制的同时支持并发调用任意函数，主要通过 goroutine 的复用来减少对 goroutine 创建与销毁时的开销

  • Running() int 返回 Running 状态的 goroutine 的数量
  • Free() int 返回可用的 goroutine 数量
  • Cap() int 返回 Pool 的容量
  • Tune(size int) 动态扩缩 Pool 的容量
  • IsClosed() bool Pool 是否关闭
  • Release() 关闭并释放资源
  • ReleaseTimeout(timeout time.Duration) error 等待后释放
  • Reboot() 重启 Pool

▼
mermaid
复制代码
stateDiagram-v2
    [*] --> NewPool
    NewPool --> Created
    Created --> Running : Submit tasks
    Running --> Running : Submit more tasks
    Running --> Released : Release()
    Released --> [*] : Resources freed
    Released --> Rebooted : Reboot()
    Rebooted --> Running : Submit tasks

• Multi Pool
  • Running() int 返回所有 Pool 中 Running 状态的 goroutine 的数量
  • Free() int 返回所有 Pool 中可用的 goroutine 数量
  • FreeByIndex(idx int) (int, error) 返回指定 Pool 中 Free 状态的 goroutine 的数量
  • Waiting() (int, error) 返回所有 Pool 中等待任务的总数
  • WaitingByIndex(idx int) (int, error) 返回指定 Pool 中等待任务的数量
  • Cap() int 返回所有 Pool 的容量
  • Tune(size int) 动态扩缩 Pool 的容量
  • IsClosed() Pool 是否关闭
  • ReleaseTimeout(timeout time.Duration) error 延迟一定时间关闭
  • Reboot() 重启 Pool

▼
mermaid
复制代码
flowchart TD
    SubmitTask --> CheckStrategy

    CheckStrategy -->|if strategy == RoundRobin| RoundRobin
    RoundRobin --> IncrementCounter[Increment counter and get next pool]
    IncrementCounter --> SelectNextPool
    SelectNextPool --> SubmitToPool

    CheckStrategy -->|if strategy == LeastTasks| LeastTasks
    LeastTasks --> ScanPools[Scan all pools]
    ScanPools --> FindPoolWithFewestTasks
    FindPoolWithFewestTasks --> SubmitToPool

    SubmitToPool --> TrySubmit
    TrySubmit -->|if pool accepts task| Success
    TrySubmit -->|if ErrPoolOverload| Failure

    Success --> End1([★ Success End])

    Failure -->|if already using LeastTasks| ReturnError
    ReturnError --> End2([★ Failure End])

    Failure -->|if RoundRobin was used initially| FallbackStrategy
    FallbackStrategy --> UseFallback[Use LeastTasks as fallback]
    UseFallback --> FindPoolWithFewestTasks

状态流转

▼
mermaid
复制代码
stateDiagram-v2
    [*] --> NewPool
    NewPool --> Created
    Created --> Running : worker.run()
    Running --> Processing
    Running --> [*] : Pool closed
    Receive --> Idle : Task completed
    Idle --> [*] : Purged after ExpiryDuration
    Idle --> [*] : Pool closed
    Idle --> Processing

源码解读

创建流程

▼
mermaid
复制代码
sequenceDiagram
    participant Caller
    participant ants as ants.NewPool
    participant Common as newPool
    participant Pool as Pool
    participant WorkerCache as pool.workerCache.New

    Caller->>ants: NewPool(size, options...)
    ants->>Common: newPool(size, options...)
    Common-->>ants: poolCommon, err
    alt if err == nil
        ants->>Pool: 创建 Pool 实例
        Pool->>WorkerCache: 设置 workerCache.New
        WorkerCache-->>Pool: 返回 goWorker 实例创建函数
    end
    ants-->>Caller: *Pool, error

▼
go
复制代码
// NewPool instantiates a Pool with customized options.
func NewPool(size int, options ...Option) (*Pool, error) {
	pc, err := newPool(size, options...)
	if err != nil {
		return nil, err
	}

	pool := &Pool{poolCommon: pc}
	pool.workerCache.New = func() any {
		return &goWorker{
			pool: pool,
			task: make(chan func(), workerChanCap),
		}
	}

	return pool, nil
}

• 创建 poolCommon

  ▼
  go
  复制代码
  func newPool(size int, options ...Option) (*poolCommon, error) {
  	if size <= 0 {
  		size = -1
  	}
  
  	opts := loadOptions(options...)
  
  	if !opts.DisablePurge {
  		if expiry := opts.ExpiryDuration; expiry < 0 {
  			return nil, ErrInvalidPoolExpiry
  		} else if expiry == 0 {
  			opts.ExpiryDuration = DefaultCleanIntervalTime
  		}
  	}
  
  	if opts.Logger == nil {
  		opts.Logger = defaultLogger
  	}
  
  	p := &poolCommon{
  		capacity: int32(size),
  		allDone:  make(chan struct{}),
  		lock:     syncx.NewSpinLock(),
  		once:     &sync.Once{},
  		options:  opts,
  	}
  	if p.options.PreAlloc {
  		if size == -1 {
  			return nil, ErrInvalidPreAllocSize
  		}
  		p.workers = newWorkerQueue(queueTypeLoopQueue, size)
  	} else {
  		p.workers = newWorkerQueue(queueTypeStack, 0)
  	}
  
  	p.cond = sync.NewCond(p.lock)
  
  	p.goPurge()
  	p.goTicktock()
  
  	return p, nil
  }
  

  在当前部分的代码中主要完成了对于公共参数的初始化、清理流程与定时流程创建

• 创建 任务队列

  ▼
  go
  复制代码
  	pool.workerCache.New = func() any {
  		return &goWorker{
  			pool: pool,
  			task: make(chan func(), workerChanCap),
  		}
  	}
  

任务提交

▼
go
复制代码
func (p *Pool) Submit(task func()) error {
	if p.IsClosed() {
		return ErrPoolClosed
	}

	w, err := p.retrieveWorker()
	if w != nil {
		w.inputFunc(task)
	}
	return err
}

• 判断当前 Pool 是否被关闭

• 获取 goWorker

• 将任务放入任务队列 w.inputFunc(task)

  ▼
  go
  复制代码
  func (w *goWorker) inputFunc(fn func()) {
  	w.task <- fn
  }
  

任务执行

▼
go
复制代码
// run starts a goroutine to repeat the process
// that performs the function calls.
func (w *goWorker) run() {
	w.pool.addRunning(1)
	go func() {
		defer func() {
			if w.pool.addRunning(-1) == 0 && w.pool.IsClosed() {
				w.pool.once.Do(func() {
					close(w.pool.allDone)
				})
			}
			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 panic: %v\n%s\n", p, debug.Stack())
				}
			}
			// Call Signal() here in case there are goroutines waiting for available workers.
			w.pool.cond.Signal()
		}()

		for fn := range w.task {
			if fn == nil {
				return
			}
			fn()
			if ok := w.pool.revertWorker(w); !ok {
				return
			}
		}
	}()
}

• 启动工作协程

  w.pool.addRunning(1) 增加了当前运行的工作协程计数，表示有一个新的工作协程启动。随后，通过 go func() 启动一个新的协程来处理任务。

  ▼
  go
  复制代码
  w.pool.addRunning(1)
  go func() {
      defer func() { ... }()
      for fn := range w.task { ... }
  }()
  

• 清理工作协程的状态与错误捕获

  • 清理工作状态

    减少运行中的工作协程计数，并检查如果池已关闭且没有运行的协程，则触发 allDone 通道的关闭信号。接着，将当前的 goWorker 对象放回 workerCache 缓存池中以供复用。

    ▼
    go
    复制代码
    defer func() {
        if w.pool.addRunning(-1) == 0 && w.pool.IsClosed() {
            w.pool.once.Do(func() {
                close(w.pool.allDone)
            })
        }
        w.pool.workerCache.Put(w)
        ...
    }()
    

  • 捕获异常状态

    如果发生异常，会调用用户定义的 PanicHandler，或者记录日志以便调试。

    ▼
    go
    复制代码
    if p := recover(); p != nil {
    	if ph := w.pool.options.PanicHandler; ph != nil {
    		ph(p)
    	} else {
    		w.pool.options.Logger.Printf("worker exits from panic: %v\n%s\n", p, debug.Stack())
    	}
    }
    

  • 状态同步

    借助 sync.Cond 同步当前状态

    ▼
    go
    复制代码
    	w.pool.cond.Signal()
    

• 执行任务并归还当前协程

  ▼
  go
  复制代码
  for fn := range w.task {
  	if fn == nil {
  		return
  	}
  	fn()
  	if ok := w.pool.revertWorker(w); !ok {
  		return
  	}
  }
  

  • 任务执行 fn()
  • 归还协程 revertWorker(w *goWorker) bool

负载均衡

▼
go
复制代码
func (mp *MultiPool) next(lbs LoadBalancingStrategy) (idx int) {
	switch lbs {
	case RoundRobin:
		return int(atomic.AddUint32(&mp.index, 1) % uint32(len(mp.pools)))
	case LeastTasks:
		leastTasks := 1<<31 - 1
		for i, pool := range mp.pools {
			if n := pool.Running(); n < leastTasks {
				leastTasks = n
				idx = i
			}
		}
		return
	}
	return -1
}

• RoundRobin
  • 将任务依次分配到每个池中，适合任务负载较为均匀的场景
  • 通过原子操作递增 mp.index，并对池的数量取模，确保索引在有效范围内循环
• LeastTasks
  • 适合任务负载不均匀的场景，可以动态选择最空闲的池
  • 遍历所有池，比较每个池中当前运行的任务数 pool.Running()，选择任务数最少的池的索引

▼
mermaid
复制代码
flowchart TD
    Start((●)) --> SubmitTask[SubmitTask]
    SubmitTask --> CheckStrategy[CheckStrategy]
    
    CheckStrategy -->|if strategy == RoundRobin| RoundRobin[RoundRobin]
    CheckStrategy -->|if strategy == LeastTasks| LeastTasks[LeastTasks]
    
    RoundRobin --> IncrementCounter["Increment counter and get\nnext pool"]
    IncrementCounter --> SelectNextPool[SelectNextPool]
    SelectNextPool --> SubmitToPool[SubmitToPool]
    
    LeastTasks --> ScanAllPools["Scan all pools"]
    ScanAllPools --> FindPoolWithFewestTasks[FindPoolWithFewestTasks]
    FindPoolWithFewestTasks --> SubmitToPool
    
    SubmitToPool --> TrySubmit[TrySubmit]
    
    TrySubmit -->|if ErrPoolOverload| Failure[Failure]
    TrySubmit -->|if pool accepts task| Success[Success]
    
    Failure -->|if RoundRobin was used\ninitially| FallbackStrategy[FallbackStrategy]
    Failure -->|if already using LeastTasks| ReturnError[ReturnError]
    
    FallbackStrategy --> UseLeastTasks["Use LeastTasks as fallback"]
    UseLeastTasks --> FindPoolWithFewestTasks
    
    Success --> End((●))
    Success --> |retry| TrySubmit

工作队列

▼
mermaid
复制代码
classDiagram
	direction TB
	
class Worker {
	<<interface>>
	-len() int
	-isEmpty() bool
	-insert(worker) error
	-detach() worker
	-refresh(time.Duration) []worker
	-reset()
}

class workerStack {
	-items []worker
	-expiry []worker
}

class workerLoopQueue {
	-items []worker
	-expiry []worker
	-head int
	-tail int
	-size int
	-isFull bool
}

Worker <|.. workerStack : implements
Worker <|.. workerLoopQueue : implements

• len() 返回队列中元素的数量
• isEmpty() 检查队列是否为空
• insert(worker) 向队列中添加一个元素
• detach() 从队列中移除元素
• refresh(time.Duration) 根据过期时间移除过期元素
• reset() 重置队列

构造函数

提供了一个工厂函数来创建适当的队列类型：

▼
go
复制代码
func newWorkerQueue(qType queueType, size int) workerQueue

队列类型

ants 定义了两种队列类型作为常量：

▼
go
复制代码
const (
    queueTypeStack queueType = 1 << iota
    queueTypeLoopQueue
)

• 工作栈实现

  工作栈是基于动态切片实现的先进后出（LIFO）队列。

  ▼
  go
  复制代码
  type workerStackstruct {
      items  []worker // Slice to store workers
      expiry []worker // Temporary storage for expired workers
  }
  

  • 插入操作

    向栈末尾添加一个工作线程，具有 $O(1)$ 平摊复杂度：

    ▼
    go
    复制代码
    func (ws *workerStack) insert(w worker)error {
      ws.items = append(ws.items, w)
    	return nil
    }
    

    由于工作栈没有大小限制，此操作永远不会失败。

  • 弹出操作

    移除并返回最近添加的工作者，复杂度为 $O(1)$：

    ▼
    go
    复制代码
    func (ws *workerStack) detach() worker {
      l := ws.len()
    	if l == 0 {
    		return nil
      }
    
      w := ws.items[l-1]
      ws.items[l-1] = nil // avoid memory leaks
      ws.items = ws.items[:l-1]
    
    	return w
    }
    

  • 刷新操作

    使用二分搜索来识别和移除超过指定持续时间的空闲工蚁：

    ▼
    go
    复制代码
    func (ws *workerStack) refresh(duration time.Duration) []worker {
    	n := ws.len()
    	if n == 0 {
    		return nil
    	}
    
    	expiryTime := time.Now().Add(-duration)
    	index := ws.binarySearch(0, n-1, expiryTime)
    
    	ws.expiry = ws.expiry[:0]
    	if index != -1 {
    		ws.expiry = append(ws.expiry, ws.items[:index+1]...)
    		m := copy(ws.items, ws.items[index+1:])
    		for i := m; i < n; i++ {
    			ws.items[i] = nil
    		}
    		ws.items = ws.items[:m]
    	}
    	return ws.expiry
    }
    
    func (ws *workerStack) binarySearch(l, r int, expiryTime time.Time) int {
    	for l <= r {
    		mid := l + ((r - l) >> 1) // avoid overflow when computing mid
    		if expiryTime.Before(ws.items[mid].lastUsedTime()) {
    			r = mid - 1
    		} else {
    			l = mid + 1
    		}
    	}
    	return r
    }
    

    二分搜索有效地找到陈旧和新鲜工作者的边界，具有 $O(\log n)$ 的复杂度。

• 工作循环队列实现

  工作循环队列是基于环形固定大小数组的先进先出（FIFO）队列。

  ▼
  go
  复制代码
  type loopQueuestruct {
      items  []worker // Fixed-size array for workers
      expiry []worker // Temporary storage for expired workers
      headint // Index for the first element
      tailint // Index for the next insertion
      sizeint // Fixed size of the queue
      isFullbool // Flag indicating if the queue is full
  }
  

  • 插入操作

    添加一个工作节点到尾部，具有 $O(1)$ 复杂度：

    ▼
    go
    复制代码
    func (wq *loopQueue)insert(w worker)error {
    	if wq.isFull {
    		return errQueueIsFull
      }
      wq.items[wq.tail] = w
      wq.tail = (wq.tail + 1) % wq.size
    
    	if wq.tail == wq.head {
    		wq.isFull = true
      }
    
    	return nil
    }
    

    由于循环队列具有固定大小，如果队列已满，此操作可能会失败并显示 errQueueIsFull or 陷入等待。

  • 删除操作

    移除并返回头位置的最早工作进程，具有 $O(1)$ 复杂度：

    ▼
    go
    复制代码
    func (wq *loopQueue) detach() worker {
    	if wq.isEmpty() {
    		return nil
      }
    
      w := wq.items[wq.head]
      wq.items[wq.head] = nil
      wq.head = (wq.head + 1) % wq.size
    
      wq.isFull = false
    
    	return w
    }
    

    实现了先进先出（FIFO）行为，为工作者提供公平的调度。

  • 刷新操作

    使用二分搜索算法来识别和删除循环数组中的过时工作节点：

    ▼
    go
    复制代码
    func (wq *loopQueue) refresh(duration time.Duration) []worker {
      expiryTime := time.Now().Add(-duration)
      index := wq.binarySearch(expiryTime)
    	// Remove and return expired workers from the circular array
    	if index == -1 {
    		return nil
    	}
    	wq.expiry = wq.expiry[:0]
    
    	if wq.head <= index {
    		wq.expiry = append(wq.expiry, wq.items[wq.head:index+1]...)
    		for i := wq.head; i < index+1; i++ {
    			wq.items[i] = nil
    		}
    	} else {
    		wq.expiry = append(wq.expiry, wq.items[0:index+1]...)
    		wq.expiry = append(wq.expiry, wq.items[wq.head:]...)
    		for i := 0; i < index+1; i++ {
    			wq.items[i] = nil
    		}
    		for i := wq.head; i < wq.size; i++ {
    			wq.items[i] = nil
    		}
    	}
    	head := (index + 1) % wq.size
    	wq.head = head
    	if len(wq.expiry) > 0 {
    		wq.isFull = false
    	}
    
    	return wq.expiry
    }
    

Focus Case

过期机制

两个队列实现都提供了一个 refresh 方法，用于在一段时间后移除空闲的工作者。这对于内存管理至关重要，尤其是在长时间运行的应用程序中。

刷新机制特别有趣，因为它使用二分搜索高效地找到过期边界，工作进程按最后使用时间排序。我们可以看一下其在 worker loop queue 中的实现

▼
go
复制代码
func (wq *loopQueue) binarySearch(expiryTime time.Time) int {
	var mid, nlen, basel, tmid int
	nlen = len(wq.items)

	// if no need to remove work, return -1
	if wq.isEmpty() || expiryTime.Before(wq.items[wq.head].lastUsedTime()) {
		return -1
	}

	// example
	// size = 8, head = 7, tail = 4
	// [ 2, 3, 4, 5, nil, nil, nil,  1]  true position
	//   0  1  2  3    4   5     6   7
	//              tail          head
	//
	//   1  2  3  4  nil nil   nil   0   mapped position
	//            r                  l

	// base algorithm is a copy from worker_stack
	// map head and tail to effective left and right
	r := (wq.tail - 1 - wq.head + nlen) % nlen
	basel = wq.head
	l := 0
	for l <= r {
		mid = l + ((r - l) >> 1) // avoid overflow when computing mid
		// calculate true mid position from mapped mid position
		tmid = (mid + basel + nlen) % nlen
		if expiryTime.Before(wq.items[tmid].lastUsedTime()) {
			r = mid - 1
		} else {
			l = mid + 1
		}
	}
	// return true position from mapped position
	return (r + basel + nlen) % nlen
}

1. 首先进行边界检查

   ▼
   go
   复制代码
   if wq.isEmpty() || expiryTime.Before(wq.items[wq.head].lastUsedTime()) {
       return -1
   }
   

2. 将循环队列的逻辑索引映射为线性索引

   ▼
   go
   复制代码
   r := (wq.tail - 1 - wq.head + nlen) % nlen
   basel = wq.head
   l := 0
   

   将循环队列的逻辑索引映射为线性索引，以便进行二分查找。通过计算尾部和头部的相对位置，确定有效的查找范围，并将队列头部作为基准索引。这一点的设计真的非常有意思🤔

3. 计算中间值

   ▼
   go
   复制代码
   mid = l + ((r - l) >> 1)
   tmid = (mid + basel + nlen) % nlen
   if expiryTime.Before(wq.items[tmid].lastUsedTime()) {
       r = mid - 1
   } else {
       l = mid + 1
   }
   

   在二分查找的循环中，方法计算中间位置的逻辑索引，并将其映射回实际的队列索引。然后比较中间位置任务的最后使用时间与过期时间，调整查找范围。如果中间任务的时间早于过期时间，则移动左边界，否则移动右边界。

   在这一步中，我们常遇到的问题也就是左开右闭和左闭右开带来的常规问题了。

   往往在迭代问题中，我们将问题拆解，将复杂问题转换为：

   • 确定边界
   • 确定迭代策略

4. 返回索引位置

   ▼
   go
   复制代码
   return (r + basel + nlen) % nlen
   

同步机制

在展开这部分之前，我们首先要阐明一下什么是同步机制、同步机制在这里具体是指什么。其实所谓的同步机制，就是指确保安全并发访问和池组件之间的信号同步。

1. 锁机制 Spinlock

   ants 库使用自定义的自旋锁实现，而不是标准 sync.Mutex ，以在高冲突场景中提高性能。在 Pool 中主要使用锁来保护对工作队列的并发安全。

   ▼
   mermaid
   复制代码
   classDiagram
   direction TB
   
   class Locker {
   <<interface>>
   +Lock()
   +Unlock()
   }
   
   class spinLock {
   -uint32 value
   }
   
   Locker <|.. spinLock : implements
   

   • 核心流程

     ▼
     mermaid
     复制代码
     flowchart TD
         A[Start Lock] --> B{Try CAS
         0 → 1}
         B -- Success --> C[Lock Acquired]
         B -- Failure --> D[Backoff
         initial = 1]
         D --> E[Yield CPU scheduler
         runtime.Gosched]
         E --> F{Backoff < maxBackoff}
         F -- Yes --> G[Double backoff]
         G --> B
         F -- No --> H[Keep current backoff]
         H --> B
     
         subgraph Unlock
             U[Start Unlock] --> V[Atomic Store
             value = 0] --> W[Lock Released]
         end
     

   • 源码走读

     ▼
     go
     复制代码
     package sync
     
     import (
     	"runtime"
     	"sync"
     	"sync/atomic"
     )
     
     type spinLock uint32
     
     const maxBackoff = 16
     
     func (sl *spinLock) Lock() {
     	backoff := 1
     	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()
     		}
     		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)
     }
     

     • Lock()

       ▼
       go
       复制代码
       for !atomic.CompareAndSwapUint32((*uint32)(sl), 0, 1) {
           for i := 0; i < backoff; i++ {
               runtime.Gosched()
           }
           if backoff < maxBackoff {
               backoff <<= 1
           }
       }
       

       • 使用 atomic.CompareAndSwapUint32 尝试将锁的状态从 0（未锁定）设置为 1（已锁定）
       • 如果获取失败，进入自旋等待。自旋过程中，采用指数退避算法逐步增加等待时间，以减少对 CPU 的占用
       • 在自旋等待中，runtime.Gosched() 会让出当前协程的执行权，允许其他协程运行，从而避免长时间占用 CPU。

     • UnLock()

       ▼
       go
       复制代码
       atomic.StoreUint32((*uint32)(sl), 0)
       

       • 通过 atomic.StoreUint32 将锁的状态重置为 0，表示锁已释放

   Exponential backoff  指数退避

   指数退避是一种算法，它使用反馈来乘性降低某些过程的速率，以逐渐找到可接受的速率。这些算法在广泛的系统和过程中得到应用，尤其是在无线网络和计算机网络中尤为突出。

   指数退避算法是一种闭环控制系统，它根据不利事件降低受控过程的速率。例如，如果智能手机应用程序无法连接到其服务器，它可能会在 1 秒后再次尝试，如果再次失败，则会在 2 秒后尝试，然后是 4 秒，等等。每次暂停都会乘以一个固定的量（在这种情况下是 2）。在这种情况下，不利事件是连接到服务器的失败。其他不利事件示例包括网络流量的冲突、来自服务的错误响应或显式请求降低速率（即退避）。

   速率降低可以建模为指数函数：

   $$ t = b^c $$

   或者

   $$ f = \frac{1}{b^c} $$

   在这里，t 是动作之间的时间延迟，b 是乘法因子或基数，c 是观察到的负面事件数量，f 是过程频率（或速率）（即单位时间内的动作数量）。每当观察到负面事件时，c 的值就会增加，从而导致延迟呈指数增长，因此速率成反比。b = 2 的指数退避算法被称为二进制指数退避算法。

   • 基准测试

     Lock Type	Performance (ns/op)
     Mutex  互斥锁	111.1
     Original Spinlock  原始自旋锁	18.01
     Backoff Spinlock  回退自旋锁	10.81

2. 原子操作

   Field	Type	Purpose	Operations
   capacity	int32	Pool maximum size	LoadInt32, StoreInt32
   running	int32	Active worker count	AddInt32, LoadInt32
   state	int32	Pool state (OPENED/CLOSED)	LoadInt32, CompareAndSwapInt32
   waiting	int32	Blocked submitters count	AddInt32, LoadInt32
   purgeDone	int32	Purge goroutine state	StoreInt32
   ticktockDone	int32	Ticktock goroutine state	StoreInt32
   now	atomic.Value	Cached current time	Store, Load

3. 信号同步

   ▼
   mermaid
   复制代码
   sequenceDiagram
       participant Client
       participant Pool
       participant sync.Cond
       participant WorkerQueue as "Worker Queue"
   
       Client->>Pool: Submit(task)
       Pool->>Pool: retrieveWorker()
       Pool->>WorkerQueue: detach()
       WorkerQueue-->>Pool: nil (no workers)
           alt Pool at capacity
               Pool->>Pool: addWaiting(+1)
               Pool->>sync.Cond: Wait()
               Note right of sync.Cond: Block until worker available
               loop Until worker available or pool closed
                   Pool-->>sync.Cond: (blocked)
               end
               Pool->>Pool: addWaiting(-1)
               Pool->>WorkerQueue: detach()
               WorkerQueue-->>Pool: worker
           end
      Pool-->>Client: Success/Error
   

   • 获取 worker

     ▼
     go
     复制代码
     // retrieveWorker returns an available worker to run the tasks.
     func (p *poolCommon) retrieveWorker() (w worker, err error) {
     	p.lock.Lock()
     
     retry:
     	// First try to fetch the worker from the queue.
     	if w = p.workers.detach(); w != nil {
     		p.lock.Unlock()
     		return
     	}
     
     	// If the worker queue is empty, and we don't run out of the pool capacity,
     	// then just spawn a new worker goroutine.
     	if capacity := p.Cap(); capacity == -1 || capacity > p.Running() {
     		p.lock.Unlock()
     		w = p.workerCache.Get().(worker)
     		w.run()
     		return
     	}
     
     	// Bail out early if it's in nonblocking mode or the number of pending callers reaches the maximum limit value.
     	if p.options.Nonblocking || (p.options.MaxBlockingTasks != 0 && p.Waiting() >= p.options.MaxBlockingTasks) {
     		p.lock.Unlock()
     		return nil, ErrPoolOverload
     	}
     
     	// Otherwise, we'll have to keep them blocked and wait for at least one worker to be put back into pool.
     	p.addWaiting(1)
     	p.cond.Wait() // block and wait for an available worker
     	p.addWaiting(-1)
     
     	if p.IsClosed() {
     		p.lock.Unlock()
     		return nil, ErrPoolClosed
     	}
     
     	goto retry
     }
     

     • 通过加锁机制保护共享资源，防止并发访问问题。

     • 进入 retry 标签后，尝试从 workers 队列中分离出一个可用的 worker。如果成功获取，则解锁并返回该 worker

       ▼
       go
       复制代码
       if w = p.workers.detach(); w != nil {
       	p.lock.Unlock()
       	return
       }
       

     • 如果队列为空且当前运行的 worker 数量未达到池的容量限制，则从 workerCache 缓存中获取一个新的 worker 并启动它。

       ▼
       go
       复制代码
       if capacity := p.Cap(); capacity == -1 || capacity > p.Running() {
           p.lock.Unlock()
           w = p.workerCache.Get().(worker)
           w.run()
           return
       }
       

     • 在非阻塞模式下，或者当等待的任务数已达到最大限制时，方法会直接返回错误 ErrPoolOverload，表示无法获取 worker

       ▼
       go
       复制代码
       if p.options.Nonblocking || (p.options.MaxBlockingTasks != 0 && p.Waiting() >= p.options.MaxBlockingTasks) {
           p.lock.Unlock()
           return nil, ErrPoolOverload
       }
       

     • 如果以上条件均不满足，方法会将当前调用者标记为等待状态，并通过 p.cond.Wait() 阻塞，直到有新的 worker 被放回池中。解锁前会检查池是否已关闭，若关闭则返回 ErrPoolClosed

       ▼
       go
       复制代码
       p.addWaiting(1)
       p.cond.Wait()
       p.addWaiting(-1)
       
       if p.IsClosed() {
           p.lock.Unlock()
           return nil, ErrPoolClosed
       }
       

     • 通过 goto retry 标签重新尝试获取 worker，确保在资源可用时能够成功分配。

   • 释放 worker

     ▼
     go
     复制代码
     // revertWorker puts a worker back into free pool, recycling the goroutines.
     func (p *poolCommon) revertWorker(worker worker) bool {
     	if capacity := p.Cap(); (capacity > 0 && p.Running() > capacity) || p.IsClosed() {
     		p.cond.Broadcast()
     		return false
     	}
     
     	worker.setLastUsedTime(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
     	}
     	if err := p.workers.insert(worker); 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
     }
     

     • 检查当前池的容量是否已满，或者池是否已关闭。如果满足这些条件，则通过 p.cond.Broadcast() 唤醒可能等待的调用者，并返回 false，表示无法将该 worker 放回池中

       ▼
       go
       复制代码
       if capacity := p.Cap(); (capacity > 0 && p.Running() > capacity) || p.IsClosed() {
           p.cond.Broadcast()
           return false
       }
       

     • 更新该 worker 的最后使用时间为当前时间，以便后续清理过期 worker 时使用

       ▼
       go
       复制代码
       worker.setLastUsedTime(p.nowTime())
       

     • 加锁以保护共享资源，避免并发问题。

     • 在锁的作用域内，再次检查池是否已关闭。如果池已关闭，则解锁并返回 false，防止内存泄漏

       ▼
       go
       复制代码
       if p.IsClosed() {
       	p.lock.Unlock()
       	return false
       }
       

     • 通知卡在 retrieveWorker() 中的调用者，在 worker 队列中有一个可用的 worker。

       ▼
       go
       复制代码
       	p.cond.Signal()
       

操作	同步机制
创建	锁和条件变量的常规初始化
任务提交	锁、原子计数器、条件变量
工作者复用	队列访问锁、状态原子操作
池大小调整	容量原子更新、条件广播
池关闭	状态原子变更、上下文取消、条件广播
池重启	状态原子变更、上下文重建

调优策略

初始参数

1. 池容量
   1. CPU 密集型任务：将池大小设置为接近可用 CPU 核心数，以最小化上下文切换。
   2. I/O 密集型任务：将池大小设置为高于 CPU 核心数，因为 goroutines 将花费时间等待 I/O 操作。
   3. 混合工作负载：从池大小为 CPU 核心数的 2 倍开始，并根据监控/压测结果进行调整。
2. 生命周期
   • 短时程：减少内存使用，但增加工作进程创建/销毁开销
   • 持续时间更长：增加内存使用，但减少工作进程创建/销毁开销
3. 内存管理策略
   • 启用预分配：减少操作期间的内存分配开销，适用于高吞吐量场景
   • 禁用预分配：节省初始内存使用，更适合可能无法充分利用其容量的池

Workload Type	Pool Size	PreAlloc	ExpiryDuration	Nonblocking	Other Considerations
CPU-bound	≈ GOMAXPROCS	true	Longer	false	Consider CPU affinity
I/O-bound	> GOMAXPROCS	false	Medium	Depends	Monitor blocking I/O
Bursty	Start small	false	Shorter	true	Use Tune() for peaks
Stable, high-volume	Larger	true	Long or disable	false	Consider WithMaxBlockingTasks
Memory-constrained	Smaller	false	Shorter	true	Monitor memory usage
Latency-sensitive	Medium	true	Long or disable	true	Minimize garbage collection

Example Configurations

• High-Throughput System

  ▼
  go
  复制代码
  p, _ := ants.NewPool(
      runtime.GOMAXPROCS(0) * 8,
      ants.WithPreAlloc(true),
      ants.WithDisablePurge(true),
      ants.WithMaxBlockingTasks(50000),
  )
  

• Low-Latency System

  ▼
  go
  复制代码
  p, _ := ants.NewPool(
      runtime.GOMAXPROCS(0) * 4,
      ants.WithPreAlloc(true),
      ants.WithNonblocking(true),
      ants.WithDisablePurge(true),
  )
  

• Memory-Constrained Environment

  ▼
  go
  复制代码
  p, _ := ants.NewPool(
      runtime.GOMAXPROCS(0),
      ants.WithExpiryDuration(100 * time.Millisecond),
      ants.WithMaxBlockingTasks(100),
  )
  

监控指标

1. 运行中的工作者：使用 pool.Running() 跟踪活动工作者的数量
2. 空闲工作者：使用 pool.Free() 监控可用容量
3. 等待时间：跟踪任务执行前的等待时长
4. 完成时间：衡量任务完成所需的时间
5. 错误率：监控提交错误，特别是 ErrPoolOverload

迭代策略

1. 根据工作负载类型从合理的默认值开始
2. 使用现实的工作负载进行基准测试
3. 一次调整一个参数并测量影响
4. 在生产中监控并根据需要调整