快捷導航

go 對象池化組件 bytebufferpool使用詳解

更新時間：2022年10月07日 10:35:15 作者：FfFJ

這篇文章主要為大家介紹了go 對象池化組件 bytebufferpool使用詳解，有需要的朋友可以借鑒參考下，希望能夠有所幫助，祝大家多多進步，早日升職加薪

1. 針對問題

在編程開發(fā)的過程中，我們經(jīng)常會有創(chuàng)建同類對象的場景，這樣的操作可能會對性能產(chǎn)生影響，一個比較常見的做法是使用對象池，需要創(chuàng)建對象的時候，我們先從對象池中查找，如果有空閑對象，則從對象池中移除這個對象并將其返回給調(diào)用者使用，只有在池中無空閑對象的時候，才會真正創(chuàng)建一個新對象

另一方面，對于使用完的對象，我們并不會對它進行銷毀，而是將它放回到對象池以供后續(xù)使用，使用對象池在頻繁創(chuàng)建和銷毀對象的情況下，能大幅的提升性能，同時為了避免對象池中的對象占用過多的內(nèi)存，對象池一般還配有特定的清理策略，Go的標準庫sync.Pool就是這樣一個例子，sync.Pool 中的對象會被垃圾回收清理掉

這類對象中，有一種比較特殊的是字節(jié)切片，在做字符串拼接的時候，為了拼接高效，我們通常將中間結(jié)果存放在一個字節(jié)緩沖中，拼接完之后，再從字節(jié)緩沖區(qū)生成字符串

Go標準庫bytes.Buffer封裝字節(jié)切片，提供一些使用接口，我們知道切片的容量是有限的，容量不足時需要進行擴容，而頻繁的擴容容易造成性能抖動

bytebufferpool實現(xiàn)了自己的Buffer類型，并引入一個簡單的算法降低擴容帶來的性能損失

2. 使用方法

bytebufferpool的接入很輕量

func main() {
   bf := bytebufferpool.Get()
   bf.WriteString("Hello")
   bf.WriteString(" World!!")
   fmt.Println(bf.String())
}

上面的這種用法使用的是defaultPool，bytebufferpool的Pool對象是公開的，也可以自行新建

3. 源碼剖析

bytebufferpool是如何做到最大程度減小內(nèi)存分配和浪費的呢，先宏觀的看整個Pool的定義，然后細化到相關(guān)的方法，就可以找到答案

bytebufferpool中Pool結(jié)構(gòu)體的定義為

type Pool struct {
   calls       [steps]uint64
   calibrating uint64
   defaultSize uint64
   maxSize     uint64
   pool sync.Pool
}

其中calls存儲了某一個區(qū)間內(nèi)不同大小對象的個數(shù)，calibrating是一個標志位，標志當前Pool是否在重新規(guī)劃中，defaultSize是元素新建時的默認大小，它的選取邏輯是當前calls中出現(xiàn)次數(shù)最多的對象對應(yīng)的區(qū)間最大值，這樣可以防止從對象池中撈取之后的頻繁擴容，maxSize限制了放入Pool中的最大元素的大小，防止因為一些很大的對象占用過多的內(nèi)存

bytebufferpool中定義了一些和defaultSize及maxSize計算相關(guān)的常量

const (
   minBitSize = 6 // 2**6=64 is a CPU cache line size
   steps      = 20
   minSize = 1 << minBitSize
   maxSize = 1 << (minBitSize + steps - 1)
   calibrateCallsThreshold = 42000
   maxPercentile           = 0.95
)

其中minBitSize表示的是第一個區(qū)間對象大小的最大值(2的xx次方-1)，在bytebufferpool中，將對象大小分為20個區(qū)間，也就是steps，第一個區(qū)間為[0, 2^6-1]，第二個為[2^6, 2^7-1]...，依此類推

calibrateCallsThreshold表示如果某個區(qū)間內(nèi)對象的數(shù)量超過這個閾值，則對Pool中的變量進行重新的計算，maxPercentile用于計算Pool中的maxSize，表示前95%的元素大小

bytebufferpool中的方法也比較少，核心的是Get和Put方法

Get

func (p *Pool) Get() *ByteBuffer {
   v := p.pool.Get()
   if v != nil {
      return v.(*ByteBuffer)
   }
   return &ByteBuffer{
      B: make([]byte, 0, atomic.LoadUint64(&p.defaultSize)),
   }
}

可以看到，如果對象池中沒有對象的話，會申請defaultSize大小的切片返回

Put

func (p *Pool) Put(b *ByteBuffer) {
   idx := index(len(b.B))
   if atomic.AddUint64(&p.calls[idx], 1) > calibrateCallsThreshold {
      p.calibrate()
   }
   maxSize := int(atomic.LoadUint64(&p.maxSize))
   if maxSize == 0 || cap(b.B) <= maxSize {
      b.Reset()
      p.pool.Put(b)
   }
}

Put方法會比較麻煩，我們分步來看

計算放入元素在calls數(shù)組中的位置

func index(n int) int {
   n--
   n >>= minBitSize
   idx := 0
   for n > 0 {
      n >>= 1
      idx++
   }
   if idx >= steps {
      idx = steps - 1
   }
   return idx
}

這里的邏輯就是先將長度右移minBitSize，如果依然大于0，則每次右移一位，idx加1，最后如果idx超出了總的steps(20)，則位置就在最后一個區(qū)間

判斷當前區(qū)間放入元素的個數(shù)是否超過了calibrateCallsThreshold指定的閾值，超過則重新計算Pool中元素的值

func (p *Pool) calibrate() {
   // 如果正在重新計算，則返回，控制多并發(fā)
   if !atomic.CompareAndSwapUint64(&p.calibrating, 0, 1) {
      return
   }
   // 計算每一段區(qū)間中的元素個數(shù) & 元素總個數(shù)
   a := make(callSizes, 0, steps)
   var callsSum uint64
   for i := uint64(0); i < steps; i++ {
      calls := atomic.SwapUint64(&p.calls[i], 0)
      callsSum += calls
      a = append(a, callSize{
         calls: calls,
         size:  minSize << i,
      })
   }
   // 按照對象元素的個數(shù)從大到小排序
   sort.Sort(a)
   // defaultSize 為內(nèi)部切片的默認大小，減少擴容次數(shù)
   // maxSize 限制放入pool中的最大元素大小
   defaultSize := a[0].size
   maxSize := defaultSize
   // 將前95%元素中的最大size給maxSize
   maxSum := uint64(float64(callsSum) * maxPercentile)
   callsSum = 0
   for i := 0; i < steps; i++ {
      if callsSum > maxSum {
         break
      }
      callsSum += a[i].calls
      size := a[i].size
      if size > maxSize {
         maxSize = size
      }
   }
   // 對defaultSize和maxSize進行賦值
   atomic.StoreUint64(&p.defaultSize, defaultSize)
   atomic.StoreUint64(&p.maxSize, maxSize)
   atomic.StoreUint64(&p.calibrating, 0)
}