go get -u github.com/panjf2000/ants/v2

import "github.com/panjf2000/ants/v2"

var pool *ants.Pool

func init() {
    var err error
    pool, err = ants.NewPool(10) // 創(chuàng)建容量為10的協(xié)程池
    if err != nil {
        log.Fatalf("Failed to create goroutine pool: %v", err)
    }
}

// Submit submits a task to the pool.
//
// Note that you are allowed to call Pool.Submit() from the current Pool.Submit(),
// but what calls for special attention is that you will get blocked with the last
// Pool.Submit() call once the current Pool runs out of its capacity, and to avoid this,
// you should instantiate a Pool with ants.WithNonblocking(true).
func (p *Pool) Submit(task func()) error 

err := pool.Submit(func() {
    fmt.Println("Task executed by goroutine:", runtime.NumGoroutine())
})
if err != nil {
    log.Println("Failed to submit task:", err)
}

pool.Release()

fmt.Printf("Running goroutines: %d\n", pool.Running())

fmt.Printf("Pool capacity: %d\n", pool.Cap())

pool.Tune(20) // 將容量調(diào)整為20

type goWorker struct {
    pool *Pool
    task chan func()
    recycleTime time.Time
}

type Pool struct {
    capacity int32
    running int32
    lock sync.Locker
    workers workerArray
    state int32
    cond *sync.Cond
    workerCache sync.Pool
    waiting int32
    heartbeatDone int32
    stopHeartbeat context.CancelFunc
    options *Options
}

type workerArray interface {
    len() int
    isEmpty() bool
    insert(worker *goWorker) error
    detach() *goWorker
    retrieveExpiry(duration time.Duration) []*goWorker
    reset()
}

func NewPool(size int, options ...Option) (*Pool, error) {
    // 讀取用戶配置，做一些前置校驗(yàn)，默認(rèn)值賦值等前處理動(dòng)作...
    opts := loadOptions(options...)

	// 構(gòu)造好 Pool 數(shù)據(jù)結(jié)構(gòu)；
    p := &Pool{
        capacity: int32(size),
        lock:     internal.NewSpinLock(),
        options:  opts,
    }
    // 構(gòu)造對(duì)象池
    p.workerCache.New = func() interface{} {
        return &goWorker{
            pool: p,
            task: make(chan func(), workerChanCap),
        }
    }
    
   // 構(gòu)造好 goWorker 對(duì)象池 workerCache，聲明好工廠函數(shù)；
    p.workers = newWorkerArray(stackType, 0)
	
  //  golang 標(biāo)準(zhǔn)庫(kù)提供的并發(fā)協(xié)調(diào)器，用于實(shí)現(xiàn)指定條件下阻塞和喚醒協(xié)程的操作.
    p.cond = sync.NewCond(p.lock)

 // 異步啟動(dòng) goWorker 過(guò)期銷毀協(xié)程.
    var ctx context.Context
    ctx, p.stopHeartbeat = context.WithCancel(context.Background())
    go p.purgePeriodically(ctx)
    return p, nil
}

func (p *Pool) Submit(task func()) error {

	// 從 Pool 中取出一個(gè)可用的 goWorker；
    var w *goWorker
    if w = p.retrieveWorker(); w == nil {
        return ErrPoolOverload
    }
	
	// 將用戶提交的任務(wù)包添加到 goWorker 的 channel 中.
    w.task <- task
    return nil
}

func (p *Pool) retrieveWorker() (w *goWorker) {
    // 聲明了一個(gè)構(gòu)造 goWorker 的函數(shù) spawnWorker 用于兜底，從對(duì)象池 workerCache 中獲取 goWorker；
    spawnWorker := func() {
        w = p.workerCache.Get().(*goWorker)
        w.run()
    }

    p.lock.Lock()

	// 嘗試從池中取出一個(gè)空閑的 goWorker；
    w = p.workers.detach()
    if w != nil { 
        p.lock.Unlock()
    
    // 倘若池子容量未超過(guò)上限, 從對(duì)象池中取出一個(gè) goWorker 
    } else if capacity := p.Cap(); capacity == -1 || capacity > p.Running() {
        p.lock.Unlock()
        spawnWorker()
    } else { 

	  // 倘若池子容量超限，且池子為非阻塞模式，直接拋回錯(cuò)誤；
        if p.options.Nonblocking {
            p.lock.Unlock()
            return
        }
	
	// 倘若池子容量超限，且池子為阻塞模式，則基于并發(fā)協(xié)調(diào)器 cond 掛起等待有空閑 worker；
    retry:
      // 若阻塞任務(wù)已達(dá)最大限制，也直接返回；
        if p.options.MaxBlockingTasks != 0 && p.Waiting() >= p.options.MaxBlockingTasks {
            p.lock.Unlock()
            return
        }

		// 增加等待數(shù)并使用 cond 條件變量掛起當(dāng)前協(xié)程；
        p.addWaiting(1)
        p.cond.Wait() // block and wait for an available worker
        p.addWaiting(-1)

	 // 被喚醒后（可能是因?yàn)?scavenger 清理協(xié)程），判斷是否還有運(yùn)行中的 worker；
        var nw int
        if nw = p.Running(); nw == 0 { // awakened by the scavenger
            p.lock.Unlock()
            spawnWorker()
            return
        }
  
      // 再次嘗試重新獲取一個(gè)空閑 worker；
        if w = p.workers.detach(); w == nil {
            if nw < p.Cap() {
                p.lock.Unlock()
                spawnWorker()
                return
            }
            goto retry
        }
       // 獲取到了可用 worker，解鎖并返回；
        p.lock.Unlock()
    }
    return
}

func (w *goWorker) run() {
    w.pool.addRunning(1)
    go func() {
        defer func() {
            w.pool.addRunning(-1)
            w.pool.workerCache.Put(w)
            if p := recover(); p != nil {
                // panic 后處理
            }
            w.pool.cond.Signal()
        }()

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