Go語言 | 基于channel實現(xiàn)的并發(fā)安全的字節(jié)池

字節(jié)切片[]byte是我們在編碼中經(jīng)常使用到的，比如要讀取文件的內(nèi)容，或者從io.Reader獲取數(shù)據(jù)等，都需要[]byte做緩沖。

func?ReadFull(r?Reader,?buf?[]byte)?(n?int,?err?error)
func?(f?*File)?Read(b?[]byte)?(n?int,?err?error)

以上是兩個使用到[]byte作為緩沖區(qū)的方法。那么現(xiàn)在問題來了，如果對于以上方法我們有大量的調(diào)用，那么就要聲明很多個[]byte，這需要太多的內(nèi)存的申請和釋放，也就會有太多的GC。

MinIO 的字節(jié)池

這個時候，我們需要重用已經(jīng)創(chuàng)建好的[]byte來提高對象的使用率，降低內(nèi)存的申請和GC。這時候我們可以使用sync.Pool來實現(xiàn)，不過最近我在研究開源項目MinIO的時候，發(fā)現(xiàn)他們使用channel的方式實現(xiàn)字節(jié)池。

type?BytePoolCap?struct?{
????c????chan?[]byte
????w????int
????wcap?int
}

BytePoolCap結(jié)構(gòu)體的定義比較簡單，共有三個字段：

1. c是一個chan，用于充當(dāng)字節(jié)緩存池

2. w是指使用make函數(shù)創(chuàng)建[]byte時候的len參數(shù)

3. wcap指使用make函數(shù)創(chuàng)建[]byte時候的cap參數(shù)

有了BytePoolCap結(jié)構(gòu)體，就可以為其定義Get方法，用于獲取一個緩存的[]byte了。

func?(bp?*BytePoolCap)?Get()?(b?[]byte)?{
????select?{
????case?b?=?<-bp.c:
????//?reuse?existing?buffer
????default:
????????//?create?new?buffer
????????if?bp.wcap?>?0?{
????????????b?=?make([]byte,?bp.w,?bp.wcap)
????????}?else?{
????????????b?=?make([]byte,?bp.w)
????????}
????}
????return
}

以上是采用經(jīng)典的select+chan的方式，能獲取到[]byte緩存則獲取，獲取不到就執(zhí)行default分支，使用make函數(shù)生成一個[]byte。

從這里也可以看到，結(jié)構(gòu)體中定義的w和wcap字段，用于make函數(shù)的len和cap參數(shù)。

有了Get方法，還要有Put方法，這樣就可以把使用過的[]byte放回字節(jié)池，便于重用。

func?(bp?*BytePoolCap)?Put(b?[]byte)?{
????select?{
????case?bp.c?<-?b:
????????//?buffer?went?back?into?pool
????default:
????????//?buffer?didn't?go?back?into?pool,?just?discard
????}
}

Put方法也是采用select+chan，能放則放，不能放就丟棄這個[]byte。

使用BytePoolCap

已經(jīng)定義好了Get和Put就可以使用了，在使用前，BytePoolCap還定義了一個工廠函數(shù)，用于生成*BytePoolCap，比較方便。

func?NewBytePoolCap(maxSize?int,?width?int,?capwidth?int)?(bp?*BytePoolCap)?{
????return?&BytePoolCap{
????????c:????make(chan?[]byte,?maxSize),
????????w:????width,
????????wcap:?capwidth,
????}
}

把相關(guān)的參數(shù)暴露出去，可以讓調(diào)用者自己定制。這里的maxSize表示要創(chuàng)建的chan有多大，也就是字節(jié)池的大小，最大存放數(shù)量。

bp?:=?bpool.NewBytePoolCap(500,?1024,?1024)
buf:=bp.Get()
defer?bp.Put(buf)

//使用buf,不再舉例

以上就是使用字節(jié)池的一般套路，使用后記得放回以便復(fù)用。

和sync.Pool對比

兩者原理基本上差不多，都多協(xié)程安全。sync.Pool可以存放任何對象，BytePoolCap只能存放[]byte，不過也正因為其自定義，存放的對象類型明確，不用經(jīng)過一層類型斷言轉(zhuǎn)換，同時也可以自己定制對象池的大小等。

關(guān)于二者的性能，我做了下Benchmark測試，整體看MinIO的BytePoolCap更好一些。

var?bp?=?bpool.NewBytePoolCap(500,?1024,?1024)
var?sp?=?&sync.Pool{
????New:?func()?interface{}?{
????????return?make([]byte,?1024,?1024)
????},
}

模擬的兩個字節(jié)池，[]byte的長度和容量都是1024。然后是兩個模擬使用字節(jié)池，這里我啟動500協(xié)程，模擬并發(fā)，使用不模擬并發(fā)的話，BytePoolCap完全是一個[]byte的分配，完全秒殺sync.Pool，對sync.Pool不公平。

func?opBytePool(bp?*bpool.BytePoolCap)?{
????var?wg?sync.WaitGroup
????wg.Add(500)
????for?i?:=?0;?i?500;?i++?{
????????go?func(bp?*bpool.BytePoolCap)?{
????????????buffer?:=?bp.Get()
????????????defer?bp.Put(buffer)
????????????mockReadFile(buffer)
????????????wg.Done()
????????}(bp)
????}
????wg.Wait()
}

func?opSyncPool(sp?*sync.Pool)?{
????var?wg?sync.WaitGroup
????wg.Add(500)
????for?i?:=?0;?i?500;?i++?{
????????go?func(sp?*sync.Pool)?{
????????????buffer?:=?sp.Get().([]byte)
????????????defer?sp.Put(buffer)
????????????mockReadFile(buffer)
????????????wg.Done()
????????}(sp)
????}
????wg.Wait()
}

接下來就是我模擬的讀取我本機文件的一個函數(shù)mockReadFile(buffer):

func?mockReadFile(b?[]byte)?{
????f,?_?:=?os.Open("water")
????for?{
????????n,?err?:=?io.ReadFull(f,?b)
????????if?n?==?0?||?err?==?io.EOF?{
????????????break
????????}
????}
}

然后運行go test -bench=. -benchmem -run=none?查看測試結(jié)果：

pkg:?flysnow.org/hello
BenchmarkBytePool-8?????????1489????????????979113?ns/op???????????36504?B/op???????1152?allocs/op
BenchmarkSyncPool-8?????????1008???????????1172429?ns/op???????????57788?B/op???????1744?allocs/op

從測試結(jié)果看BytePoolCap在內(nèi)存分配，每次操作分配字節(jié)，每次操作耗時來看，都比sync.Pool更有優(yōu)勢。

小結(jié)

很多優(yōu)秀的開源項目，可以看到很多優(yōu)秀的源代碼實現(xiàn)，并且會根據(jù)自己的業(yè)務(wù)場景，做出更好的優(yōu)化。

本文為原創(chuàng)文章，轉(zhuǎn)載注明出處,歡迎掃碼關(guān)注公眾號flysnow_org或者網(wǎng)站asf https://www.flysnow.org/?，第一時間看后續(xù)精彩文章。覺得好的話，請猛擊文章右下角「在看」，感謝支持。