Go創(chuàng)建Grpc鏈接池實(shí)現(xiàn)過(guò)程詳解
常規(guī)用法
gRPC
四種基本使用
- 請(qǐng)求響應(yīng)模式
- 客戶端數(shù)據(jù)流模式
- 服務(wù)端數(shù)據(jù)流模式
- 雙向流模式
常見(jiàn)的gRPC
調(diào)用寫(xiě)法
func main(){ //... some code // 鏈接grpc服務(wù) conn , err := grpc.Dial(":8000",grpc.WithInsecure) if err != nil { //...log } defer conn.Close() //...some code
存在的問(wèn)題:面臨高并發(fā)的情況,性能問(wèn)題很容易就會(huì)出現(xiàn),例如我們?cè)谧鲂阅軠y(cè)試的時(shí)候,就會(huì)發(fā)現(xiàn),打一會(huì)性能測(cè)試,客戶端請(qǐng)求服務(wù)端的時(shí)候就會(huì)報(bào)錯(cuò):
rpc error: code = Unavailable desc = all SubConns are in TransientFailure, latest connection error: connection error: desc = "transport: Error while dialing dial tcp xxx:xxx: connect: connection refused
實(shí)際去查看問(wèn)題的時(shí)候,很明顯,這是 gRPC 的連接數(shù)被打滿了,很多連接都還未完全釋放。[#本文來(lái)源:janrs.com#]
gRPC
的通信本質(zhì)上也是 TCP
的連接,那么一次連接就需要三次握手,和四次揮手,每一次建立連接和釋放連接的時(shí)候,都需要走這么一個(gè)過(guò)程,如果我們頻繁的建立和釋放連接,這對(duì)于資源和性能其實(shí)都是一個(gè)大大的浪費(fèi)。
在服務(wù)端,gRPC
服務(wù)端的鏈接管理不用我們操心,但是 gRPC
客戶端的鏈接管理非常有必要關(guān)心,要實(shí)現(xiàn)復(fù)用客戶端的連接。
創(chuàng)建鏈接池
創(chuàng)建鏈接池需要考慮的問(wèn)題:
- 連接池是否支持?jǐn)U縮容
- 空閑的連接是否支持超時(shí)自行關(guān)閉,是否支持保活
- 池子滿的時(shí)候,處理的策略是什么樣的
創(chuàng)建鏈接池接口
type Pool interface { // 獲取一個(gè)新的連接 , 當(dāng)關(guān)閉連接的時(shí)候,會(huì)將該連接放入到池子中 Get() (Conn, error) // 關(guān)閉連接池,自然連接池子中的連接也不再可用 Close() error //[#本文來(lái)源:janrs.com#] Status() string }
實(shí)現(xiàn)鏈接池接口
創(chuàng)建鏈接池代碼
func New(address string, option Options) (Pool, error) { if address == "" { return nil, errors.New("invalid address settings") } if option.Dial == nil { return nil, errors.New("invalid dial settings") } if option.MaxIdle <= 0 || option.MaxActive <= 0 || option.MaxIdle > option.MaxActive { return nil, errors.New("invalid maximum settings") } if option.MaxConcurrentStreams <= 0 { return nil, errors.New("invalid maximun settings") } p := &pool{ index: 0, current: int32(option.MaxIdle), ref: 0, opt: option, conns: make([]*conn, option.MaxActive), address: address, closed: 0, } for i := 0; i < p.opt.MaxIdle; i++ { c, err := p.opt.Dial(address) if err != nil { p.Close() return nil, fmt.Errorf("dial is not able to fill the pool: %s", err) } p.conns[i] = p.wrapConn(c, false) } log.Printf("new pool success: %v\n", p.Status()) return p, nil }
關(guān)于以上的代碼,需要特別注意每一個(gè)連接的建立也是在 New
里面完成的,[#本文來(lái)源:janrs.com#]只要有 1
個(gè)連接未建立成功,那么咱們的連接池就算是建立失敗,咱們會(huì)調(diào)用 p.Close()
將之前建立好的連接全部釋放掉。
關(guān)閉鏈接池代碼
// 關(guān)閉連接池 func (p *pool) Close() error { atomic.StoreInt32(&p.closed, 1) atomic.StoreUint32(&p.index, 0) atomic.StoreInt32(&p.current, 0) atomic.StoreInt32(&p.ref, 0) p.deleteFrom(0) log.Printf("[janrs.com]close pool success: %v\n", p.Status()) return nil }
從具體位置刪除鏈接池代碼
// 清除從 指定位置開(kāi)始到 MaxActive 之間的連接 func (p *pool) deleteFrom(begin int) { for i := begin; i < p.opt.MaxActive; i++ { p.reset(i) } }
銷毀具體的鏈接代碼
// 清除具體的連接 func (p *pool) reset(index int) { conn := p.conns[index] if conn == nil { return } conn.reset() p.conns[index] = nil }
關(guān)閉鏈接
代碼
func (c *conn) reset() error { cc := c.cc c.cc = nil c.once = false // 本文博客來(lái)源:janrs.com if cc != nil { return cc.Close() } return nil } func (c *conn) Close() error { c.pool.decrRef() if c.once { return c.reset() } return nil }
在使用連接池通過(guò) pool.Get()
拿到具體的連接句柄 conn
之后,會(huì)使用 conn.Close()
關(guān)閉連接,實(shí)際上也是會(huì)走到上述的 Close()
實(shí)現(xiàn)的位置,但是并未指定當(dāng)然也沒(méi)有權(quán)限顯示的指定將 once
置位為 false
,也就是對(duì)于調(diào)用者來(lái)說(shuō),是關(guān)閉了連接,對(duì)于連接池來(lái)說(shuō),實(shí)際上是將連接歸還到連接池中。
擴(kuò)縮容
關(guān)鍵代碼
func (p *pool) Get() (Conn, error) { // the first selected from the created connections nextRef := p.incrRef() p.RLock() current := atomic.LoadInt32(&p.current) p.RUnlock() if current == 0 { return nil, ErrClosed } if nextRef <= current*int32(p.opt.MaxConcurrentStreams) { next := atomic.AddUint32(&p.index, 1) % uint32(current) return p.conns[next], nil } // 本文博客來(lái)源:janrs.com // the number connection of pool is reach to max active if current == int32(p.opt.MaxActive) { // the second if reuse is true, select from pool's connections if p.opt.Reuse { next := atomic.AddUint32(&p.index, 1) % uint32(current) return p.conns[next], nil } // the third create one-time connection c, err := p.opt.Dial(p.address) return p.wrapConn(c, true), err } // the fourth create new connections given back to pool p.Lock() current = atomic.LoadInt32(&p.current) if current < int32(p.opt.MaxActive) && nextRef > current*int32(p.opt.MaxConcurrentStreams) { // 2 times the incremental or the remain incremental ##janrs.com increment := current if current+increment > int32(p.opt.MaxActive) { increment = int32(p.opt.MaxActive) - current } var i int32 var err error for i = 0; i < increment; i++ { c, er := p.opt.Dial(p.address) if er != nil { err = er break } p.reset(int(current + i)) p.conns[current+i] = p.wrapConn(c, false) } // 本文博客來(lái)源:janrs.com current += i log.Printf("#janrs.com#grow pool: %d ---> %d, increment: %d, maxActive: %d\n", p.current, current, increment, p.opt.MaxActive) atomic.StoreInt32(&p.current, current) if err != nil { p.Unlock() return nil, err } } p.Unlock() next := atomic.AddUint32(&p.index, 1) % uint32(current) return p.conns[next], nil }
Get
代碼邏輯
- 先增加連接的引用計(jì)數(shù),如果在設(shè)定
current*int32(p.opt.MaxConcurrentStreams)
范圍內(nèi),那么直接取連接進(jìn)行使用即可。 - 若當(dāng)前的連接數(shù)達(dá)到了最大活躍的連接數(shù),那么就看我們新建池子的時(shí)候傳遞的
option
中的reuse
參數(shù)是否是true
,若是復(fù)用,則隨機(jī)取出連接池中的任意連接提供使用,如果不復(fù)用,則新建一個(gè)連接。 - 其余的情況,就需要我們進(jìn)行
2
倍或者1
倍的數(shù)量對(duì)連接池進(jìn)行擴(kuò)容了。
也可以在 Get
的實(shí)現(xiàn)上進(jìn)行縮容,具體的縮容策略可以根據(jù)實(shí)際情況來(lái)定,例如當(dāng)引用計(jì)數(shù) nextRef
只有當(dāng)前活躍連接數(shù)的 10%
的時(shí)候(這只是一個(gè)例子),就可以考慮縮容了。
性能測(cè)試
有關(guān)鏈接池的創(chuàng)建以及性能測(cè)試
以上就是Go創(chuàng)建Grpc鏈接池實(shí)現(xiàn)過(guò)程詳解的詳細(xì)內(nèi)容,更多關(guān)于Go創(chuàng)建Grpc鏈接池的資料請(qǐng)關(guān)注腳本之家其它相關(guān)文章!
相關(guān)文章
這些關(guān)于Go中interface{}的注意事項(xiàng)你都了解嗎
這篇文章主要為大家詳細(xì)介紹了學(xué)習(xí)Go語(yǔ)言時(shí)需要了解的interface{}注意事項(xiàng),文中的示例代碼講解詳細(xì),感興趣的小伙伴可以跟隨小編一起了解一下2023-03-03Go Gin框架中的binding驗(yàn)證器使用小結(jié)
Gin框架中的binding驗(yàn)證器為我們提供了簡(jiǎn)便的數(shù)據(jù)綁定和驗(yàn)證功能,通過(guò)合理使用binding和validate標(biāo)簽,我們可以確保API接口的數(shù)據(jù)合法性和完整性,這篇文章主要介紹了Go Gin框架中的binding驗(yàn)證器使用指南,需要的朋友可以參考下2024-07-07Go語(yǔ)言服務(wù)器開(kāi)發(fā)實(shí)現(xiàn)最簡(jiǎn)單HTTP的GET與POST接口
這篇文章主要介紹了Go語(yǔ)言服務(wù)器開(kāi)發(fā)實(shí)現(xiàn)最簡(jiǎn)單HTTP的GET與POST接口,實(shí)例分析了Go語(yǔ)言http包的使用技巧,需要的朋友可以參考下2015-02-02用go gin server來(lái)做文件上傳服務(wù)
今天小編就為大家分享一篇關(guān)于用go gin server來(lái)做文件上傳服務(wù),小編覺(jué)得內(nèi)容挺不錯(cuò)的,現(xiàn)在分享給大家,具有很好的參考價(jià)值,需要的朋友一起跟隨小編來(lái)看看吧2019-04-04Golang map實(shí)踐及實(shí)現(xiàn)原理解析
這篇文章主要介紹了Golang map實(shí)踐以及實(shí)現(xiàn)原理,Go 語(yǔ)言中,通過(guò)哈希查找表實(shí)現(xiàn) map,用鏈表法解決哈希沖突,本文結(jié)合實(shí)例代碼給大家介紹的非常詳細(xì),需要的朋友參考下吧2022-06-06golang?pprof?監(jiān)控系列?go?trace統(tǒng)計(jì)原理與使用解析
這篇文章主要為大家介紹了golang?pprof?監(jiān)控系列?go?trace統(tǒng)計(jì)原理與使用解析,有需要的朋友可以借鑒參考下,希望能夠有所幫助,祝大家多多進(jìn)步,早日升職加薪2023-04-04