Pool介绍#
总所周知Go 是一个自动垃圾回收的编程语言，采用三色并发标记算法标记对象并回收。如果你想使用 Go 开发一个高性能的应用程序的话，就必须考虑垃圾回收给性能带来的影响。因为Go 在垃圾回收的时候会有一个STW（stop-the-world，程序暂停）的时间，并且如果对象太多，做标记也需要时间。所以如果采用对象池来创建对象，增加对象的重复利用率，使用的时候就不必在堆上重新创建对象可以节省开销。在Go中，golang提供了对象重用的机制，也就是sync.Pool对象池。 sync.Pool是可伸缩的，并发安全的。其大小仅受限于内存的大小，可以被看作是一个存放可重用对象的值的容器。 设计的目的是存放已经分配的但是暂时不用的对象，在需要用到的时候直接从pool中取。
任何存放区其中的值可以在任何时候被删除而不通知，在高负载下可以动态的扩容，在不活跃时对象池会收缩。它对外提供了三个方法：New、Get 和 Put。下面用一个简短的例子来说明一下Pool使用：
package main import ( "fmt" "sync" ) var pool *sync.Pool type Person struct { Name string } func init() { pool = &sync.Pool{ New: func() interface{} { fmt.Println("creating a new person") return new(Person) }, } } func main() { person := pool.Get().(*Person) fmt.Println("Get Pool Object1：", person) person.Name = "first" pool.Put(person) fmt.Println("Get Pool Object2：", pool.Get().(*Person)) fmt.Println("Get Pool Object3：", pool.Get().(*Person)) }
结果：
creating a new person Get Pool Object1： &{} Get Pool Object2： &{first} creating a new person Get Pool Object3： &{}
这里我用了init方法初始化了一个pool，然后get了三次，put了一次到pool中，如果pool中没有对象，那么会调用New函数创建一个新的对象，否则会从put进去的对象中获取。
存储在池中的任何项目都可以随时自动删除，并且不会被通知。Pool可以安全地同时使用多个goroutine。池的目的是缓存已分配但未使用的对象以供以后重用，从而减轻对gc的压力。也就是说，它可以轻松构建高效，线程安全的free列表。但是，它不适用于所有free列表。池的适当使用是管理一组默认共享的临时项，并且可能由包的并发独立客户端重用。池提供了一种在许多客户端上分摊分配开销的方法。很好地使用池的一个例子是fmt包，它维护一个动态大小的临时输出缓冲区存储。底层存储队列在负载下（当许多goroutine正在积极打印时）进行缩放，并在静止时收缩。另一方面，作为短期对象的一部分维护的空闲列表不适合用于池， 因为在该场景中开销不能很好地摊销。 使这些对象实现自己的空闲列表更有效。首次使用后不得复制池。
pool 的两个特点
1、在本地私有池和本地共享池均获取 obj 失败时,则会从其他p偷一个 obj 返回给调用方。
2、obj在池中的生命周期取决于垃圾回收任务的下一次执行时间,并且从池中获取到的值可能是 put 进去的其中一个值，也可能是 newfun处 新生成的一个值，在应用时很容易入坑。
在多个goroutine之间使用同一个pool做到高效，是因为sync.pool为每个P都分配了一个子池，
当执行一个pool的get或者put操作的时候都会先把当前的goroutine固定到某个P的子池上面，
然后再对该子池进行操作。每个子池里面有一个私有对象和共享列表对象，
私有对象是只有对应的P能够访问，因为一个P同一时间只能执行一个goroutine，
【因此对私有对象存取操作是不需要加锁的】。
源码分析
type Pool struct { // 不允许复制,一个结构体,有一个Lock()方法,嵌入别的结构体中,表示不允许复制 // noCopy对象，拥有一个Lock方法，使得Cond对象在进行go vet扫描的时候，能够被检测到是否被复制 noCopy noCopy //local 和 localSize 维护一个动态 poolLocal 数组 // 每个固定大小的池， 真实类型是 [P]poolLocal // 其实就是一个[P]poolLocal 的指针地址 local unsafe.Pointer // local fixed-size per-P pool, actual type is [P]poolLocal localSize uintptr // size of the local array victim unsafe.Pointer // local from previous cycle victimSize uintptr // size of victims array // New optionally specifies a function to generate // a value when Get would otherwise return nil. // It may not be changed concurrently with calls to Get. // New 是一个回调函数指针,当Get 获取到目标对象为 nil 时，需要调用此处的回调函数用于生成 新的对象 New func() interface{} }
-
Pool结构体里面noCopy代表这个结构体是禁止拷贝的，它可以在我们使用
go vet工具的时候生效；
local是一个poolLocal数组的指针，localSize代表这个数组的大小；同样victim也是一个poolLocal数组的指针，每次垃圾回收的时候，Pool 会把 victim 中的对象移除，然后把 local 的数据给 victim；local和victim的逻辑我们下面会详细介绍到。
New函数是在创建pool的时候设置的，当pool没有缓存对象的时候，会调用New方法生成一个新的对象。
下面我们对照着pool的结构图往下讲，避免找不到北：
// Local per-P Pool appendix. /* 因为poolLocal中的对象可能会被其他P偷走， private域保证这个P不会被偷光，至少保留一个对象供自己用。 否则，如果这个P只剩一个对象，被偷走了， 那么当它本身需要对象时又要从别的P偷回来，造成了不必要的开销 */ type poolLocalInternal struct { private interface{} // Can be used only by the respective P. shared poolChain // Local P can pushHead/popHead; any P can popTail. } type poolLocal struct { poolLocalInternal // Prevents false sharing on widespread platforms with // 128 mod (cache line size) = 0 . /** cache使用中常见的一个问题是false sharing。 当不同的线程同时读写同一cache line上不同数据时就可能发生false sharing。 false sharing会导致多核处理器上严重的系统性能下降。 字节对齐，避免 false sharing （伪共享） */ pad [128 - unsafe.Sizeof(poolLocalInternal{})%128]byte }
local字段存储的是一个poolLocal数组的指针，poolLocal数组大小是goroutine中P的数量，访问时，P的id对应poolLocal数组下标索引，所以Pool的最大个数runtime.GOMAXPROCS(0)。
通过这样的设计，每个P都有了自己的本地空间，多个 goroutine 使用同一个 Pool 时，减少了竞争，提升了性能。如果对goroutine的P、G、M有疑惑的同学不妨看看这篇文章：The Go scheduler。
poolLocal里面有一个pad数组用来占位用，防止在 cache line 上分配多个 poolLocalInternal从而造成false sharing，cache使用中常见的一个问题是false sharing。当不同的线程同时读写同一cache line上不同数据时就可能发生false sharing。false sharing会导致多核处理器上严重的系统性能下降。具体的可以参考伪共享(False Sharing)。
poolLocalInternal包含两个字段private和shared。
private代表缓存的一个元素，只能由相应的一个 P 存取。因为一个 P 同时只能执行一个 goroutine，所以不会有并发的问题；所以无需加锁
shared则可以由任意的 P 访问，但是只有本地的 P 才能 pushHead/popHead，其它 P 可以 popTail。因为可能有多个goroutine同时操作，所以需要加锁。
type poolChain struct { // head is the poolDequeue to push to. This is only accessed // by the producer, so doesn't need to be synchronized. head *poolChainElt // tail is the poolDequeue to popTail from. This is accessed // by consumers, so reads and writes must be atomic. tail *poolChainElt } type poolChainElt struct { poolDequeue // next and prev link to the adjacent poolChainElts in this // poolChain. // // next is written atomically by the producer and read // atomically by the consumer. It only transitions from nil to // non-nil. // // prev is written atomically by the consumer and read // atomically by the producer. It only transitions from // non-nil to nil. next, prev *poolChainElt } type poolDequeue struct { // headTail packs together a 32-bit head index and a 32-bit // tail index. Both are indexes into vals modulo len(vals)-1. // // tail = index of oldest data in queue // head = index of next slot to fill // // Slots in the range [tail, head) are owned by consumers. // A consumer continues to own a slot outside this range until // it nils the slot, at which point ownership passes to the // producer. // // The head index is stored in the most-significant bits so // that we can atomically add to it and the overflow is // harmless. headTail uint64 // vals is a ring buffer of interface{} values stored in this // dequeue. The size of this must be a power of 2. // // vals[i].typ is nil if the slot is empty and non-nil // otherwise. A slot is still in use until *both* the tail // index has moved beyond it and typ has been set to nil. This // is set to nil atomically by the consumer and read // atomically by the producer. vals []eface } type eface struct { typ, val unsafe.Pointer }
poolChain是一个双端队列，里面的head和tail分别指向队列头尾；poolDequeue里面存放真正的数据，是一个单生产者、多消费者的固定大小的无锁的环状队列，headTail是环状队列的首位位置的指针，可以通过位运算解析出首尾的位置，生产者可以从 head 插入、head 删除，而消费者仅可从 tail 删除。
这个双端队列的模型大概是这个样子：
poolDequeue里面的环状队列大小是固定的，后面分析源码我们会看到，当环状队列满了的时候会创建一个size是原来两倍大小的环状队列。大家这张图好好体会一下，会反复用到。
Get方法#
// Get selects an arbitrary item from the Pool, removes it from the // Pool, and returns it to the caller. // Get may choose to ignore the pool and treat it as empty. // Callers should not assume any relation between values passed to Put and // the values returned by Get. // // If Get would otherwise return nil and p.New is non-nil, Get returns // the result of calling p.New. func (p *Pool) Get() interface{} { if race.Enabled { race.Disable() } l, pid := p.pin() //1.把当前goroutine绑定在当前的P上 x := l.private //2.优先从local的private中获取 l.private = nil if x == nil { // Try to pop the head of the local shard. We prefer // the head over the tail for temporal locality of // reuse. x, _ = l.shared.popHead() //3，private没有，那么从shared的头部获取 if x == nil { x = p.getSlow(pid) //4. 如果都没有，那么去别的local上去偷一个 } } runtime_procUnpin() //解除抢占 if race.Enabled { race.Enable() if x != nil { race.Acquire(poolRaceAddr(x)) } } //5. 如果没有获取到，尝试使用New函数生成一个新的 if x == nil && p.New != nil { x = p.New() } return x }
-
这一段代码首先会将当前goroutine绑定在当前的P上返回对应的local，然后尝试从local的private中获取，然后需要把private字段置空，因为已经拿到了想要的对象；
-
private中获取不到，那么就去shared的头部获取；
-
shared也没有，那么尝试遍历所有的 local，尝试从它们的 shared 弹出一个元素；
-
最后如果还是没有，那么就直接调用预先设置好的 New 函数，创建一个出来。
pin#
// pin 会将当前 goroutine 订到 P 上, 禁止抢占(preemption) 并从 poolLocal 池中返回 P 对应的 poolLocal // 调用方必须在完成取值后调用 runtime_procUnpin() 来取消禁止抢占。 // pin pins the current goroutine to P, disables preemption and // returns poolLocal pool for the P and the P's id. // Caller must call runtime_procUnpin() when done with the pool. func (p *Pool) pin() (*poolLocal, int) { pid := runtime_procPin() // In pinSlow we store to local and then to localSize, here we load in opposite order. // Since we've disabled preemption, GC cannot happen in between. // Thus here we must observe local at least as large localSize. // We can observe a newer/larger local, it is fine (we must observe its zero-initialized-ness). // 因为可能存在动态的 P（运行时调整 P 的个数）procresize/GOMAXPROCS // 如果 P.id 没有越界，则直接返回 PID /** 具体的逻辑就是首先拿到当前的pid， 然后以pid作为index找到local中的poolLocal， 但是如果pid大于了localsize， 说明当前线程的poollocal不存在,就会新创建一个poolLocal */ s := atomic.LoadUintptr(&p.localSize) // load-acquire l := p.local // load-consume if uintptr(pid) < s { return indexLocal(l, pid), pid } // 没有结果时，涉及全局加锁 // 例如重新分配数组内存，添加到全局列表 return p.pinSlow() } func indexLocal(l unsafe.Pointer, i int) *poolLocal { lp := unsafe.Pointer(uintptr(l) + uintptr(i)*unsafe.Sizeof(poolLocal{})) return (*poolLocal)(lp) }
pin方法里面首先会调用runtime_procPin方法会先获取当前goroutine，然后绑定到对应的M上，然后返回M目前绑定的P的id，因为这个pid后面会用到，防止在使用途中P被抢占，具体的细节可以看这篇：https://zhuanlan.zhihu.com/p/99710992。接下来会使用原子操作取出localSize，如果当前pid大于localSize，那么就表示Pool还没创建对应的poolLocal，那么调用pinSlow进行创建工作，否则调用indexLocal取出pid对应的poolLocal返回。
indexLocal里面是使用了地址操作，传入的i是数组的index值，所以需要获取poolLocal{}的size做一下地址的位移操作，然后再转成转成poolLocal地址返回。
pinSlow#
func (p *Pool) pinSlow() (*poolLocal, int) { // Retry under the mutex. // Can not lock the mutex while pinned. //因为需要对全局进行加锁，pinSlow() 会首先取消 P 的不可抢占，然后使用 allPoolsMu 进行加锁 runtime_procUnpin() // 解除pin allPoolsMu.Lock() // 加上全局锁 defer allPoolsMu.Unlock() pid := runtime_procPin() // pin住 // poolCleanup won't be called while we are pinned. s := p.localSize l := p.local // 重新对pid进行检查 再次检查是否符合条件，因为可能中途已被其他线程调用 // 当再次固定 P 时 poolCleanup 不会被调用 if uintptr(pid) < s { return indexLocal(l, pid), pid } // 初始化local前会将pool放入到allPools数组中 if p.local == nil { allPools = append(allPools, p) } // If GOMAXPROCS changes between GCs, we re-allocate the array and lose the old one. size := runtime.GOMAXPROCS(0) // 当前P的数量 local := make([]poolLocal, size) atomic.StorePointer(&p.local, unsafe.Pointer(&local[0])) // store-release atomic.StoreUintptr(&p.localSize, uintptr(size)) // store-release return &local[pid], pid }
因为allPoolsMu是一个全局Mutex锁，因此上锁会比较慢可能被阻塞，所以上锁前调用runtime_procUnpin方法解除pin的操作；
在解除绑定后，pinSlow 可能被其他的线程调用过了，p.local 可能会发生变化。因此这时候需要再次对 pid 进行检查。
最后初始化local，并使用原子操作对local和localSize设值，返回当前P对应的local。
到这里pin方法终于讲完了。画一个简单的图描述一下这整个流程：
下面我们再回到Get方法中，如果private中没有值，那么会调用shared的popHead方法获取值。
popHead#
func (c *poolChain) popHead() (interface{}, bool) { d := c.head // 这里头部是一个poolChainElt // 遍历poolChain链表 for d != nil { // 从poolChainElt的环状列表中获取值 if val, ok := d.popHead(); ok { return val, ok } // There may still be unconsumed elements in the // previous dequeue, so try backing up. // load poolChain下一个对象 d = loadPoolChainElt(&d.prev) } return nil, false } // popHead removes and returns the element at the head of the queue. // It returns false if the queue is empty. It must only be called by a // single producer. func (d *poolDequeue) popHead() (interface{}, bool) { var slot *eface for { ptrs := atomic.LoadUint64(&d.headTail) head, tail := d.unpack(ptrs) // headTail的高32位为head，低32位为tail if tail == head { // Queue is empty. // 首尾相等，那么这个队列就是空的 return nil, false } // Confirm tail and decrement head. We do this before // reading the value to take back ownership of this // slot. head-- // 这里需要head--之后再获取slot ptrs2 := d.pack(head, tail) if atomic.CompareAndSwapUint64(&d.headTail, ptrs, ptrs2) { // We successfully took back slot. slot = &d.vals[head&uint32(len(d.vals)-1)] break } } val := *(*interface{})(unsafe.Pointer(slot)) if val == dequeueNil(nil) { // 说明没取到缓存的对象，返回 nil val = nil } // Zero the slot. Unlike popTail, this isn't racing with // pushHead, so we don't need to be careful here. *slot = eface{} // 重置slot return val, true }
poolChain的popHead方法里面会获取到poolChain的头结点，不记得poolChain数据结构的同学建议往上面翻一下再回来。接着有个for循环会挨个从poolChain的头结点往下遍历，直到获取对象返回。
-
poolDequeue的popHead方法首先会获取到headTail的值，然后调用unpack解包，headTail是一个64位的值，高32位表示head，低32位表示tail。
-
判断head和tail是否相等，相等那么这个队列就是空的；
-
如果队列不是空的，那么将head减一之后再使用，因为head当前指的位置是空值，表示下一个新对象存放的位置；
-
CAS重新设值新的headTail，成功之后获取slot，这里因为vals大小是2的n 次幂，因此
len(d.vals)-1)之后低n位全是1，和head取与之后可以获取到head的低n位的值； -
如果slot所对应的对象是dequeueNil，那么表示是空值，直接返回，否则将slot指针对应位置的值置空，返回val。
如果shared的popHead方法也没获取到值，那么就需要调用getSlow方法获取了。
getSlow#
// 从其他P的共享缓冲区偷取 obj func (p *Pool) getSlow(pid int) interface{} { // See the comment in pin regarding ordering of the loads. size := atomic.LoadUintptr(&p.localSize) // load-acquire 获取当前 poolLocal 的大小 locals := p.local // load-consume 获取当前 poolLocal // Try to steal one element from other procs. // 遍历locals列表，从其他的local的shared列表尾部获取对象 for i := 0; i < int(size); i++ { l := indexLocal(locals, (pid+i+1)%int(size)) if x, _ := l.shared.popTail(); x != nil { return x } } // Try the victim cache. We do this after attempting to steal // from all primary caches because we want objects in the // victim cache to age out if at all possible. size = atomic.LoadUintptr(&p.victimSize) if uintptr(pid) >= size { return nil } locals = p.victim l := indexLocal(locals, pid) // victim的private不为空则返回 if x := l.private; x != nil { l.private = nil return x } // 遍历victim对应的locals列表，从其他的local的shared列表尾部获取对象 for i := 0; i < int(size); i++ { l := indexLocal(locals, (pid+i)%int(size)) if x, _ := l.shared.popTail(); x != nil { return x } } // Mark the victim cache as empty for future gets don't bother // with it. // 获取不到，将victimSize置为0 atomic.StoreUintptr(&p.victimSize, 0) return nil }
getSlow方法会遍历locals列表，这里需要注意的是，遍历是从索引为 pid+1 的 poolLocal 处开始，尝试调用shared的popTail方法获取对象；如果没有拿到，则从 victim 里找。如果都没找到，那么就将victimSize置为0，下次就不找victim了。
poolChain&popTail#
func (c *poolChain) popTail() (interface{}, bool) { d := loadPoolChainElt(&c.tail) if d == nil { return nil, false // 如果最后一个节点是空的，那么直接返回 } for { // It's important that we load the next pointer // *before* popping the tail. In general, d may be // transiently empty, but if next is non-nil before // the pop and the pop fails, then d is permanently // empty, which is the only condition under which it's // safe to drop d from the chain. // 这里获取的是next节点，与一般的双向链表是相反的 d2 := loadPoolChainElt(&d.next) if val, ok := d.popTail(); ok { return val, ok } if d2 == nil { // This is the only dequeue. It's empty right // now, but could be pushed to in the future. return nil, false } // The tail of the chain has been drained, so move on // to the next dequeue. Try to drop it from the chain // so the next pop doesn't have to look at the empty // dequeue again. // 因为d已经没有数据了，所以重置tail为d2，并删除d2的上一个节点 if atomic.CompareAndSwapPointer((*unsafe.Pointer)(unsafe.Pointer(&c.tail)), unsafe.Pointer(d), unsafe.Pointer(d2)) { // We won the race. Clear the prev pointer so // the garbage collector can collect the empty // dequeue and so popHead doesn't back up // further than necessary. storePoolChainElt(&d2.prev, nil) } d = d2 } }
- 判断poolChain，如果最后一个节点是空的，那么直接返回；
- 进入for循环，获取tail的next节点，这里需要注意的是这个双向链表与一般的链表是反向的，不清楚的可以再去看看第一张图；
- 调用popTail获取poolDequeue列表的对象，有对象直接返回；
- d2为空则表示已经遍历完整个poolChain双向列表了，都为空，那么直接返回；
- 通过CAS将tail重置为d2，因为d已经没有数据了，并将d2的prev节点置为nil，然后将d置为d2，进入下一个循环；
poolDequeue&popTail#
// popTail removes and returns the element at the tail of the queue. // It returns false if the queue is empty. It may be called by any // number of consumers. func (d *poolDequeue) popTail() (interface{}, bool) { var slot *eface for { ptrs := atomic.LoadUint64(&d.headTail) head, tail := d.unpack(ptrs) // 和pophead一样，将headTail解包 if tail == head { // Queue is empty. // 首位相等，表示列表中没有数据，返回 return nil, false } // Confirm head and tail (for our speculative check // above) and increment tail. If this succeeds, then // we own the slot at tail. ptrs2 := d.pack(head, tail+1) // CAS重置tail位置 if atomic.CompareAndSwapUint64(&d.headTail, ptrs, ptrs2) { // Success. // 获取tail位置对象 slot = &d.vals[tail&uint32(len(d.vals)-1)] break } } // We now own slot. val := *(*interface{})(unsafe.Pointer(slot)) if val == dequeueNil(nil) { val = nil } // Tell pushHead that we're done with this slot. Zeroing the // slot is also important so we don't leave behind references // that could keep this object live longer than necessary. // // We write to val first and then publish that we're done with // this slot by atomically writing to typ. slot.val = nil atomic.StorePointer(&slot.typ, nil) // At this point pushHead owns the slot. return val, true }
如果看懂了popHead，那么这个popTail方法是和它非常的相近的。
popTail简单来说也是从队列尾部移除一个元素，如果队列为空，返回 false。但是需要注意的是，这个popTail可能会被多个消费者调用，所以需要循环CAS获取对象；在poolDequeue环状列表中tail是有数据的，不必像popHead中head--。
最后，需要将slot置空。
大家可以再对照一下图回顾一下代码：
Put方法#
// Put adds x to the pool. func (p *Pool) Put(x interface{}) { if x == nil { return } if race.Enabled { if fastrand()%4 == 0 { // Randomly drop x on floor. return } race.ReleaseMerge(poolRaceAddr(x)) race.Disable() } l, _ := p.pin() // 先获得当前P绑定的 localPool if l.private == nil { l.private = x x = nil } if x != nil { l.shared.pushHead(x) } // 调用方必须在完成取值后调用 runtime_procUnpin() 来取消禁用抢占 runtime_procUnpin() if race.Enabled { race.Enable() } }
看完了Get方法，看Put方法就容易多了。同样Put方法首先会去Pin住当前goroutine和P，然后尝试将 x 赋值给 private 字段。如果private不为空，那么就调用pushHead将其放入到shared队列中。
poolChain&pushHead#
func (c *poolChain) pushHead(val interface{}) { d := c.head // 头节点没有初始化，那么设值一下 if d == nil { // Initialize the chain. const initSize = 8 // Must be a power of 2 d = new(poolChainElt) d.vals = make([]eface, initSize) c.head = d storePoolChainElt(&c.tail, d) } // 将对象加入到环状队列中 if d.pushHead(val) { return } // The current dequeue is full. Allocate a new one of twice // the size. newSize := len(d.vals) * 2 // 这里做了限制，单个环状队列不能超过2的30次方大小 if newSize >= dequeueLimit { // Can't make it any bigger. newSize = dequeueLimit } // 初始化新的环状列表，大小是d的两倍 d2 := &poolChainElt{prev: d} d2.vals = make([]eface, newSize) c.head = d2 storePoolChainElt(&d.next, d2) // push到新的队列中 d2.pushHead(val) }
如果头节点为空，那么需要创建一个新的poolChainElt对象作为头节点，大小为8；然后调用pushHead放入到环状队列中；
如果放置失败，那么创建一个 poolChainElt 节点，并且双端队列的长度翻倍，当然长度也不能超过dequeueLimit，即2的30次方；
然后将新的节点d2和d互相绑定一下，并将d2设值为头节点，将传入的对象push到d2中；
poolDequeue&pushHead#
// pushHead adds val at the head of the queue. It returns false if the // queue is full. It must only be called by a single producer. func (d *poolDequeue) pushHead(val interface{}) bool { ptrs := atomic.LoadUint64(&d.headTail) head, tail := d.unpack(ptrs) // 解包headTail // 判断队列是否已满 if (tail+uint32(len(d.vals)))&(1<<dequeueBits-1) == head { // Queue is full. return false } slot := &d.vals[head&uint32(len(d.vals)-1)] // 找到head的槽位 // Check if the head slot has been released by popTail. typ := atomic.LoadPointer(&slot.typ) // 检查slot是否和popTail有冲突 if typ != nil { // Another goroutine is still cleaning up the tail, so // the queue is actually still full. return false } // The head slot is free, so we own it. if val == nil { val = dequeueNil(nil) } // 将 val 赋值到 slot，并将 head 指针值加 1 *(*interface{})(unsafe.Pointer(slot)) = val // Increment head. This passes ownership of slot to popTail // and acts as a store barrier for writing the slot. atomic.AddUint64(&d.headTail, 1<<dequeueBits) return true }
首先通过位运算判断队列是否已满，也就是将尾部指针加上 len(d.vals) ，因为head指向的是将要被填充的位置，所以head和tail位置是相隔len(d.vals)，然后再取低 31 位，看它是否和 head 相等。如果队列满了，直接返回 false；
然后找到找到head的槽位slot，并判断typ是否为空，因为popTail 是先设置 val，再将 typ 设置为 nil，所以如果有冲突，那么直接返回；
最后设值slot，并将head加1返回；
GC#
在pool.go文件的 init 函数里，注册了 GC 发生时，如何清理 Pool 的函数:
func init() { runtime_registerPoolCleanup(poolCleanup) } // 当 stop the world (STW) 来临，在 GC 之前会调用该函数 func poolCleanup() { // This function is called with the world stopped, at the beginning of a garbage collection. // It must not allocate and probably should not call any runtime functions. // Because the world is stopped, no pool user can be in a // pinned section (in effect, this has all Ps pinned). // Drop victim caches from all pools. for _, p := range oldPools { p.victim = nil p.victimSize = 0 } // Move primary cache to victim cache. for _, p := range allPools { p.victim = p.local p.victimSize = p.localSize p.local = nil p.localSize = 0 } // The pools with non-empty primary caches now have non-empty // victim caches and no pools have primary caches. oldPools, allPools = allPools, nil }
poolCleanup 会在 STW 阶段被调用。主要是将 local 和 victim 作交换，那么不至于GC 把所有的 Pool 都清空了，而是需要两个 GC 周期才会被释放。如果 sync.Pool 的获取、释放速度稳定，那么就不会有新的池对象进行分配。存在Pool中的对象会在没有任何通知的情况下被自动移除掉。实际上，这个清理过程是在每次垃圾回收之前做的。垃圾回收是固定两分钟触发一次。而且每次清理会将Pool中的所有对象都清理掉！
总结#
整个设计充分利用了go.runtime的调度器优势：一个P下goroutine竞争的无锁化；
一个goroutine固定在一个局部调度器P上，从当前 P 对应的 poolLocal 取值， 若取不到，则从对应的 shared 数组上取，若还是取不到；则尝试从其他 P 的 shared 中偷。 若偷不到，则调用 New 创建一个新的对象。池中所有临时对象在一次 GC 后会被全部清空。
通过以上的解读，我们可以看到，Get方法并不会对获取到的对象值做任何的保证，因为放入本地池中的值有可能会在任何时候被删除，但是不通知调用者。放入共享池中的值有可能被其他的goroutine偷走。 所以对象池比较适合用来存储一些临时且状态无关的数据，但是不适合用来存储数据库连接的实例，因为存入对象池重的值有可能会在垃圾回收时被删除掉，这违反了数据库连接池建立的初衷。
根据上面的说法，Golang的对象池严格意义上来说是一个临时的对象池，适用于储存一些会在goroutine间分享的临时对象。主要作用是减少GC，提高性能。在Golang中最常见的使用场景是fmt包中的输出缓冲区。
在Golang中如果要实现连接池的效果，可以用container/list来实现，开源界也有一些现成的实现，比如go-commons-pool，具体的读者可以去自行了解。
Go语言的goroutine虽然可以创建很多，但是真正能物理上并发运行的goroutine数量是有限的，是由runtime.GOMAXPROCS(0)设置的。所以这个Pool高效的设计的地方就在于将数据分散在了各个真正并发的线程中，每个线程优先从自己的poolLocal中获取数据，很大程度上降低了锁竞争。
Reference#
https://www.cnblogs.com/luozhiyun/p/14194872.html
https://blog.csdn.net/u010853261/article/details/90647884