Ants协程池库

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

// 实现 Locker 接口
type spinLock uint32
// 最大回退次数
const maxBackoff = 16
// 加锁
func (sl *spinLock) Lock() {
    backoff := 1
    // 基于 CAS 机制，尝试获取锁
    for !atomic.CompareAndSwapUint32((*uint32)(sl), 0, 1) {
        // 执行 backoff 次 cpu 让出时间片次数
        for i := 0; i < backoff; i++ {
            // 使当前goroutine让出CPU时间片
            runtime.Gosched()
        }
        if backoff < maxBackoff {
            // 左移后赋值 等于 backoff = backoff << 1
            // 左移一位就是乘以 2 的 1 次方
            backoff <<= 1
        }
    }
}

// 释放锁
func (sl *spinLock) Unlock() {
    atomic.StoreUint32((*uint32)(sl), 0)
}

type Pool struct {

    // 协程池容量
    capacity int32

    // 当前协程池中正在运行的协程数
    running int32

    // ants 实现的自旋锁，用于同步并发操作
    lock sync.Locker

    // 存放一组 Worker
    workers workerArray

    // 协程池状态 （1-关闭、0-开启）
    state int32

    // 并发协调器，用于阻塞模式下，挂起和唤醒等待资源的协程
    cond *sync.Cond

    // worker 对象池
    workerCache sync.Pool

    // 等待的协程数量
    waiting int32

    // 回收协程是否关闭
    heartbeatDone int32

    // 关闭回收协程的控制器函数
    stopHeartbeat context.CancelFunc

    // 协程池的配置
    options *Options
}

type goWorker struct {
    // goWorker 所属的协程池
    pool *Pool
    // 接收实际执行任务的管道
    task chan func()
    // goWorker 回收到协程池的时间
    recycleTime time.Time
}

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

func NewPool(size int, options ...Option) (*Pool, error) {
    // 读取一些自定义的配置
    opts := loadOptions(options...)

    ...
    // 创建 Pool 对象
    p := &Pool{
        capacity: int32(size),
        lock:     internal.NewSpinLock(),
        options:  opts,
    }
     // 指定 sync.Pool 创建 worker 的方法
    p.workerCache.New = func() interface{} {
        return &goWorker{
            pool: p,
            task: make(chan func(), workerChanCap),
        }
    }
    // 初始化 Pool 时是否进行内存预分配
    // 区分 workerArray 的实现方式
    if p.options.PreAlloc {
        if size == -1 {
            return nil, ErrInvalidPreAllocSize
        }
        // 预先分配固定 Size 的池子
        p.workers = newWorkerArray(loopQueueType, size)
    } else {
        // 初始化不创建，运行时再创建
        p.workers = newWorkerArray(stackType, 0)
    }

    p.cond = sync.NewCond(p.lock)

    // 开启一个 goroutine 清理过期的 worker
    go p.purgePeriodically()

    return p, nil
}