Go 并发就是如此简单——带你走读 Ants 🐜🐜🐜
Go#Go
ants 是一个高性能的 goroutine 池，实现了对大规模 goroutine 的调度管理、goroutine 复用，允许使用者在开发并发程序的时候限制 goroutine 数量，复用资源，达到更高效执行任务的效果。
功能
- 自动调度海量的 goroutines，复用 goroutines
- 定期清理过期的 goroutines，进一步节省资源
- 提供了大量实用的接口：任务提交、获取运行中的 goroutine 数量、动态调整 Pool 大小、释放 Pool、重启 Pool 等
- 优雅处理 panic，防止程序崩溃
- 资源复用，极大节省内存使用量；在大规模批量并发任务场景下甚至可能比 Go 语言的无限制 goroutine 并发具有更高的性能
- 非阻塞机制
- 预分配内存 (环形队列，可选)
工作流程
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
快速开始
安装
go get -u github.com/panjf2000/ants/v2使用
- Default Pool
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
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
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
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 减少内存分配开销
- 提供受控的并发机制
- 提升高吞吐量场景下的性能
classDiagram
direction TB
class poolCommon {
+int32 capacity
+int32 running
+sync.Locker lock
+workerQueue workers
+int32 state
+sync.Cond cond
+channel 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() boolPool 是否关闭Release()关闭并释放资源ReleaseTimeout(timeout time.Duration) error等待后释放Reboot()重启 Pool
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
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
状态流转
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
源码解读
创建流程
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
// 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}-
创建
poolCommonfunc 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}在当前部分的代码中主要完成了对于公共参数的初始化、清理流程与定时流程创建
-
创建
任务队列pool.workerCache.New = func() any {return &goWorker{pool: pool,task: make(chan func(), workerChanCap),}}
任务提交
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)func (w *goWorker) inputFunc(fn func()) {w.task <- fn}
任务执行
// 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() 启动一个新的协程来处理任务。
w.pool.addRunning(1)go func() {defer func() { ... }()for fn := range w.task { ... }}() -
清理工作协程的状态与错误捕获
-
清理工作状态
减少运行中的工作协程计数，并检查如果池已关闭且没有运行的协程，则触发 allDone 通道的关闭信号。接着，将当前的 goWorker 对象放回 workerCache 缓存池中以供复用。
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，或者记录日志以便调试。
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同步当前状态w.pool.cond.Signal()
-
-
执行任务并归还当前协程
for fn := range w.task {if fn == nil {return}fn()if ok := w.pool.revertWorker(w); !ok {return}}- 任务执行
fn() - 归还协程
revertWorker(w *goWorker) bool
- 任务执行
负载均衡
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()，选择任务数最少的池的索引
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
工作队列
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()重置队列
构造函数
提供了一个工厂函数来创建适当的队列类型：
func newWorkerQueue(qType queueType, size int) workerQueue队列类型
ants 定义了两种队列类型作为常量：
const ( queueTypeStack queueType = 1 << iota queueTypeLoopQueue)-
工作栈实现
工作栈是基于动态切片实现的先进后出（LIFO）队列。
type workerStackstruct {items []worker // Slice to store workersexpiry []worker // Temporary storage for expired workers}-
插入操作
向栈末尾添加一个工作线程，具有 $O(1)$ 平摊复杂度：
func (ws *workerStack) insert(w worker)error {ws.items = append(ws.items, w)return nil}由于工作栈没有大小限制，此操作永远不会失败。
-
弹出操作
移除并返回最近添加的工作者，复杂度为 $O(1)$：
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 leaksws.items = ws.items[:l-1]return w} -
刷新操作
使用二分搜索来识别和移除超过指定持续时间的空闲工蚁：
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 midif expiryTime.Before(ws.items[mid].lastUsedTime()) {r = mid - 1} else {l = mid + 1}}return r}二分搜索有效地找到陈旧和新鲜工作者的边界，具有 $O(\log n)$ 的复杂度。
-
-
工作循环队列实现
工作循环队列是基于环形固定大小数组的先进先出（FIFO）队列。
type loopQueuestruct {items []worker // Fixed-size array for workersexpiry []worker // Temporary storage for expired workersheadint // Index for the first elementtailint // Index for the next insertionsizeint // Fixed size of the queueisFullbool // Flag indicating if the queue is full}-
插入操作
添加一个工作节点到尾部，具有 $O(1)$ 复杂度：
func (wq *loopQueue)insert(w worker)error {if wq.isFull {return errQueueIsFull}wq.items[wq.tail] = wwq.tail = (wq.tail + 1) % wq.sizeif wq.tail == wq.head {wq.isFull = true}return nil}由于循环队列具有固定大小，如果队列已满，此操作可能会失败并显示
errQueueIsFullor 陷入等待。 -
删除操作
移除并返回头位置的最早工作进程，具有 $O(1)$ 复杂度：
func (wq *loopQueue) detach() worker {if wq.isEmpty() {return nil}w := wq.items[wq.head]wq.items[wq.head] = nilwq.head = (wq.head + 1) % wq.sizewq.isFull = falsereturn w}实现了先进先出（FIFO）行为，为工作者提供公平的调度。
-
刷新操作
使用二分搜索算法来识别和删除循环数组中的过时工作节点：
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 arrayif 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.sizewq.head = headif len(wq.expiry) > 0 {wq.isFull = false}return wq.expiry}
-
Focus Case
过期机制
两个队列实现都提供了一个 refresh 方法，用于在一段时间后移除空闲的工作者。这对于内存管理至关重要，尤其是在长时间运行的应用程序中。
刷新机制特别有趣，因为它使用二分搜索高效地找到过期边界，工作进程按最后使用时间排序。我们可以看一下其在
worker loop queue中的实现
首先进行边界检查
将循环队列的逻辑索引映射为线性索引
将循环队列的逻辑索引映射为线性索引，以便进行二分查找。通过计算尾部和头部的相对位置，确定有效的查找范围，并将队列头部作为基准索引。这一点的设计真的非常有意思🤔
计算中间值
在二分查找的循环中，方法计算中间位置的逻辑索引，并将其映射回实际的队列索引。然后比较中间位置任务的最后使用时间与过期时间，调整查找范围。如果中间任务的时间早于过期时间，则移动左边界，否则移动右边界。
在这一步中，我们常遇到的问题也就是左开右闭和左闭右开带来的常规问题了。
往往在迭代问题中，我们将问题拆解，将复杂问题转换为：
- 确定边界
- 确定迭代策略
返回索引位置
同步机制
在展开这部分之前，我们首先要阐明一下什么是同步机制、同步机制在这里具体是指什么。其实所谓的同步机制，就是指确保安全并发访问和池组件之间的信号同步。
-
锁机制 Spinlock
ants 库使用自定义的自旋锁实现，而不是标准
sync.Mutex，以在高冲突场景中提高性能。在 Pool 中主要使用锁来保护对工作队列的并发安全。classDiagram direction TB class Locker { <<interface>> +Lock() +Unlock() } class spinLock { -uint32 value } Locker <|.. spinLock : implements-
核心流程
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 -
源码走读
package syncimport ("runtime""sync""sync/atomic")type spinLock uint32const maxBackoff = 16func (sl *spinLock) Lock() {backoff := 1for !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()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()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
-
-
原子操作
Field Type Purpose Operations capacityint32Pool maximum size LoadInt32,StoreInt32runningint32Active worker count AddInt32,LoadInt32stateint32Pool state (OPENED/CLOSED) LoadInt32,CompareAndSwapInt32waitingint32Blocked submitters count AddInt32,LoadInt32purgeDoneint32Purge goroutine state StoreInt32ticktockDoneint32Ticktock goroutine state StoreInt32nowatomic.ValueCached current time Store,Load -
信号同步
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// 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 workerp.addWaiting(-1)if p.IsClosed() {p.lock.Unlock()return nil, ErrPoolClosed}goto retry}-
通过加锁机制保护共享资源，防止并发访问问题。
-
进入 retry 标签后，尝试从
workers队列中分离出一个可用的worker。如果成功获取，则解锁并返回该workerif w = p.workers.detach(); w != nil {p.lock.Unlock()return} -
如果队列为空且当前运行的
worker数量未达到池的容量限制，则从workerCache缓存中获取一个新的worker并启动它。if capacity := p.Cap(); capacity == -1 || capacity > p.Running() {p.lock.Unlock()w = p.workerCache.Get().(worker)w.run()return} -
在非阻塞模式下，或者当等待的任务数已达到最大限制时，方法会直接返回错误
ErrPoolOverload，表示无法获取workerif p.options.Nonblocking || (p.options.MaxBlockingTasks != 0 && p.Waiting() >= p.options.MaxBlockingTasks) {p.lock.Unlock()return nil, ErrPoolOverload} -
如果以上条件均不满足，方法会将当前调用者标记为等待状态，并通过
p.cond.Wait()阻塞，直到有新的worker被放回池中。解锁前会检查池是否已关闭，若关闭则返回ErrPoolClosedp.addWaiting(1)p.cond.Wait()p.addWaiting(-1)if p.IsClosed() {p.lock.Unlock()return nil, ErrPoolClosed} -
通过
goto retry标签重新尝试获取worker，确保在资源可用时能够成功分配。
-
-
释放
worker// 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/113if 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放回池中if capacity := p.Cap(); (capacity > 0 && p.Running() > capacity) || p.IsClosed() {p.cond.Broadcast()return false} -
更新该
worker的最后使用时间为当前时间，以便后续清理过期worker时使用worker.setLastUsedTime(p.nowTime()) -
加锁以保护共享资源，避免并发问题。
-
在锁的作用域内，再次检查池是否已关闭。如果池已关闭，则解锁并返回
false，防止内存泄漏if p.IsClosed() {p.lock.Unlock()return false} -
通知卡在
retrieveWorker()中的调用者，在worker队列中有一个可用的worker。p.cond.Signal()
-
-
| 操作 | 同步机制 |
|---|---|
| 创建 | 锁和条件变量的常规初始化 |
| 任务提交 | 锁、原子计数器、条件变量 |
| 工作者复用 | 队列访问锁、状态原子操作 |
| 池大小调整 | 容量原子更新、条件广播 |
| 池关闭 | 状态原子变更、上下文取消、条件广播 |
| 池重启 | 状态原子变更、上下文重建 |
调优策略
初始参数
- 池容量
- CPU 密集型任务：将池大小设置为接近可用 CPU 核心数，以最小化上下文切换。
- I/O 密集型任务：将池大小设置为高于 CPU 核心数，因为 goroutines 将花费时间等待 I/O 操作。
- 混合工作负载：从池大小为 CPU 核心数的 2 倍开始，并根据监控/压测结果进行调整。
- 生命周期
- 短时程：减少内存使用，但增加工作进程创建/销毁开销
- 持续时间更长：增加内存使用，但减少工作进程创建/销毁开销
- 内存管理策略
- 启用预分配：减少操作期间的内存分配开销，适用于高吞吐量场景
- 禁用预分配：节省初始内存使用，更适合可能无法充分利用其容量的池
| 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
Low-Latency System
Memory-Constrained Environment
监控指标
- 运行中的工作者：使用
pool.Running()跟踪活动工作者的数量 - 空闲工作者：使用
pool.Free()监控可用容量 - 等待时间：跟踪任务执行前的等待时长
- 完成时间：衡量任务完成所需的时间
- 错误率：监控提交错误，特别是
ErrPoolOverload
迭代策略
- 根据工作负载类型从合理的默认值开始
- 使用现实的工作负载进行基准测试
- 一次调整一个参数并测量影响
- 在生产中监控并根据需要调整