快捷導(dǎo)航

go 對(duì)象池化組件 bytebufferpool使用詳解

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

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

1. 針對(duì)問題

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

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

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

Go標(biāo)準(zhǔn)庫bytes.Buffer封裝字節(jié)切片，提供一些使用接口，我們知道切片的容量是有限的，容量不足時(shí)需要進(jìn)行擴(kuò)容，而頻繁的擴(kuò)容容易造成性能抖動(dòng)

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

2. 使用方法

bytebufferpool的接入很輕量

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

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

3. 源碼剖析

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

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

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

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

bytebufferpool中定義了一些和defaultSize及maxSize計(jì)算相關(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表示的是第一個(gè)區(qū)間對(duì)象大小的最大值(2的xx次方-1)，在bytebufferpool中，將對(duì)象大小分為20個(gè)區(qū)間，也就是steps，第一個(gè)區(qū)間為[0, 2^6-1]，第二個(gè)為[2^6, 2^7-1]...，依此類推

calibrateCallsThreshold表示如果某個(gè)區(qū)間內(nèi)對(duì)象的數(shù)量超過這個(gè)閾值，則對(duì)Pool中的變量進(jìn)行重新的計(jì)算，maxPercentile用于計(jì)算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)),
   }
}

可以看到，如果對(duì)象池中沒有對(duì)象的話，會(huì)申請(qǐng)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方法會(huì)比較麻煩，我們分步來看

計(jì)算放入元素在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)，則位置就在最后一個(gè)區(qū)間

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

func (p *Pool) calibrate() {
   // 如果正在重新計(jì)算，則返回，控制多并發(fā)
   if !atomic.CompareAndSwapUint64(&p.calibrating, 0, 1) {
      return
   }
   // 計(jì)算每一段區(qū)間中的元素個(gè)數(shù) & 元素總個(gè)數(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,
      })
   }
   // 按照對(duì)象元素的個(gè)數(shù)從大到小排序
   sort.Sort(a)
   // defaultSize 為內(nèi)部切片的默認(rèn)大小，減少擴(kuò)容次數(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
      }
   }
   // 對(duì)defaultSize和maxSize進(jìn)行賦值
   atomic.StoreUint64(&p.defaultSize, defaultSize)
   atomic.StoreUint64(&p.maxSize, maxSize)
   atomic.StoreUint64(&p.calibrating, 0)
}