Golang 實現(xiàn)Thrift客戶端連接池方式
1 前言
閱讀文章之前,請先了解一下thrift相關(guān)知識。thrift官方并沒有提供客戶端連接池的實現(xiàn)方案,而我們在實際使用時,thrift客戶端必須復(fù)用,來保證較為可觀的吞吐量,并避免在高QPS調(diào)用情況下,不斷的創(chuàng)建、釋放客戶端所帶來的機器端口耗盡問題。
本文會詳細講解如何實現(xiàn)一個簡單可靠的thrift客戶端連接池,并通過對照實驗來說明thrift客戶端連接池所帶來的好處。
由于篇幅的原因,本文只粘出關(guān)鍵代碼,源代碼請查看Thrift Client Pool Demo
1.1 運行環(huán)境
Golang版本: go1.14.3 darwin/amd64
Thrift Golang庫版本: 0.13.0
Thrift IDL編輯器版本: 0.13.0
1.2 .thrift文件
namespace java com.czl.api.thrift.model namespace cpp com.czl.api namespace php com.czl.api namespace py com.czl.api namespace js com.czl.apixianz namespace go com.czl.api struct ApiRequest { 1: required i16 id; } struct ApiResponse{ 1:required string name; } // service1 service ApiService1{ ApiResponse query(1:ApiRequest request) } // service2 service ApiService2{ ApiResponse query(1:ApiRequest request) }
注:請通過安裝Thrift IDL編譯器,并生成客戶端、服務(wù)端代碼。
1.3 對照實驗說明
通過腳本開啟100個協(xié)程并發(fā)調(diào)用rpc服務(wù)10分鐘,統(tǒng)計這段時間內(nèi),未使用thrift客戶端連接池與使用客戶端連接池服務(wù)的平均吞吐量、Thrift API調(diào)用平均延遲、機器端口消耗等數(shù)據(jù)進行性能對比。
實驗一: 未使用thrift客戶端連接池
實驗二: 使用thrift客戶端連接池
2 Thrift客戶端連接池實現(xiàn)
2.1 連接池的功能
首先,我們要明確一下連接池的職責(zé),這里我簡單的總結(jié)一下,連接池主要功能是維護連接的創(chuàng)建、釋放,通過緩存連接來復(fù)用連接,減少創(chuàng)建連接所帶來的開銷,提高系統(tǒng)的吞吐量,一般連接池還會有連接斷開的重連機制、超時機制等。這里我們可以先定義出大部分連接池都會有的功能,只是定義,可以先不管每個功能的具體實現(xiàn)。每一個空閑Thrift客戶端其實底層都維護著一條空閑TCP連接,空閑Thrift客戶端與空閑連接在這里其實是同一個概念。
...... // Thrift客戶端創(chuàng)建方法,留給業(yè)務(wù)去實現(xiàn) type ThriftDial func(addr string, connTimeout time.Duration) (*IdleClient, error) // 關(guān)閉Thrift客戶端,留給業(yè)務(wù)實現(xiàn) type ThriftClientClose func(c *IdleClient) error // Thrift客戶端連接池 type ThriftPool struct { // Thrift客戶端創(chuàng)建邏輯,業(yè)務(wù)自己實現(xiàn) Dial ThriftDial // Thrift客戶端關(guān)閉邏輯,業(yè)務(wù)自己實現(xiàn) Close ThriftClientClose // 空閑客戶端,用雙端隊列存儲 idle list.List // 同步鎖,確保count、status、idle等公共數(shù)據(jù)并發(fā)操作安全 lock *sync.Mutex // 記錄當(dāng)前已經(jīng)創(chuàng)建的Thrift客戶端,確保MaxConn配置 count int32 // Thrift客戶端連接池狀態(tài),目前就open和stop兩種 status uint32 // Thrift客戶端連接池相關(guān)配置 config *ThriftPoolConfig } // 連接池配置 type ThriftPoolConfig struct { // Thrfit Server端地址 Addr string // 最大連接數(shù) MaxConn int32 // 創(chuàng)建連接超時時間 ConnTimeout time.Duration // 空閑客戶端超時時間,超時主動釋放連接,關(guān)閉客戶端 IdleTimeout time.Duration // 獲取Thrift客戶端超時時間 Timeout time.Duration // 獲取Thrift客戶端失敗重試間隔 interval time.Duration } // Thrift客戶端 type IdleClient struct { // Thrift傳輸層,封裝了底層連接建立、維護、關(guān)閉、數(shù)據(jù)讀寫等細節(jié) Transport thrift.TTransport // 真正的Thrift客戶端,業(yè)務(wù)創(chuàng)建傳入 RawClient interface{} } // 封裝了Thrift客戶端 type idleConn struct { // 空閑Thrift客戶端 c *IdleClient // 最近一次放入空閑隊列的時間 t time.Time } // 獲取Thrift空閑客戶端 func (p *ThriftPool) Get() (*IdleClient, error) { // 1. 從空閑池中獲取空閑客戶端,獲取到更新數(shù)據(jù),返回,否則執(zhí)行第2步 // 2. 創(chuàng)建新到Thrift客戶端,更新數(shù)據(jù),返回Thrift客戶端 ...... } // 歸還Thrift客戶端 func (p *ThriftPool) Put(client *IdleCLient) error { // 1. 如果客戶端已經(jīng)斷開,更新數(shù)據(jù),返回,否則執(zhí)行第2步 // 2. 將Thrift客戶端丟進空閑連接池,更新數(shù)據(jù),返回 ...... } // 超時管理,定期釋放空閑太久的連接 func (p *ThriftPool) CheckTimeout() { // 掃描空閑連接池,將空閑太久的連接主動釋放掉,并更新數(shù)據(jù) ...... } // 異常連接重連 func (p *ThriftPool) Reconnect(client *IdleClient) (newClient *IdleClient, err error) { // 1. 關(guān)閉舊客戶端 // 2. 創(chuàng)建新的客戶端,并返回 ...... } // 其他方法 ......
這里有兩個關(guān)鍵的數(shù)據(jù)結(jié)構(gòu),ThriftPool和IdleClient,ThriftPool負責(zé)實現(xiàn)整個連接池的功能,IdleClient封裝了真正的Thrift客戶端。
先看一下ThriftPool的定義:
// Thrift客戶端創(chuàng)建方法,留給業(yè)務(wù)去實現(xiàn) type ThriftDial func(addr string, connTimeout time.Duration) (*IdleClient, error) // 關(guān)閉Thrift客戶端,留給業(yè)務(wù)實現(xiàn) type ThriftClientClose func(c *IdleClient) error // Thrift客戶端連接池 type ThriftPool struct { // Thrift客戶端創(chuàng)建邏輯,業(yè)務(wù)自己實現(xiàn) Dial ThriftDial // Thrift客戶端關(guān)閉邏輯,業(yè)務(wù)自己實現(xiàn) Close ThriftClientClose // 空閑客戶端,用雙端隊列存儲 idle list.List // 同步鎖,確保count、status、idle等公共數(shù)據(jù)并發(fā)操作安全 lock *sync.Mutex // 記錄當(dāng)前已經(jīng)創(chuàng)建的Thrift客戶端,確保MaxConn配置 count int32 // Thrift客戶端連接池狀態(tài),目前就open和stop兩種 status uint32 // Thrift客戶端連接池相關(guān)配置 config *ThriftPoolConfig } // 連接池配置 type ThriftPoolConfig struct { // Thrfit Server端地址 Addr string // 最大連接數(shù) MaxConn int32 // 創(chuàng)建連接超時時間 ConnTimeout time.Duration // 空閑客戶端超時時間,超時主動釋放連接,關(guān)閉客戶端 IdleTimeout time.Duration // 獲取Thrift客戶端超時時間 Timeout time.Duration // 獲取Thrift客戶端失敗重試間隔 interval time.Duration }
Thrift客戶端創(chuàng)建與關(guān)閉,涉及到業(yè)務(wù)細節(jié),這里抽離成Dial方法和Close方法。
連接池需要維護空閑客戶端,這里用雙端隊列來存儲。
一般的連接池,都應(yīng)該支持最大連接數(shù)配置,MaxConn可以配置連接池最大連接數(shù),同時我們用count來記錄連接池當(dāng)前已經(jīng)創(chuàng)建的連接。
為了實現(xiàn)連接池的超時管理,當(dāng)然也得有相關(guān)超時配置。
連接池的狀態(tài)、當(dāng)前連接數(shù)等這些屬性,是多協(xié)程并發(fā)操作的,這里用同步鎖lock來確保并發(fā)操作安全。
在看一下IdleClient實現(xiàn):
// Thrift客戶端 type IdleClient struct { // Thrift傳輸層,封裝了底層連接建立、維護、關(guān)閉、數(shù)據(jù)讀寫等細節(jié) Transport thrift.TTransport // 真正的Thrift客戶端,業(yè)務(wù)創(chuàng)建傳入 RawClient interface{} } // 封裝了Thrift客戶端 type idleConn struct { // 空閑Thrift客戶端 c *IdleClient // 最近一次放入空閑隊列的時間 t time.Time }
RawClient是真正的Thrift客戶端,與實際邏輯相關(guān)。
Transport Thrift傳輸層,Thrift傳輸層,封裝了底層連接建立、維護、關(guān)閉、數(shù)據(jù)讀寫等細節(jié)。
idleConn封裝了IdleClient,用來實現(xiàn)空閑連接超時管理,idleConn記錄一個時間,這個時間是Thrift客戶端最近一次被放入空閑隊列的時間。
2.2 獲取連接
...... var nowFunc = time.Now ...... // 獲取Thrift空閑客戶端 func (p *ThriftPool) Get() (*IdleClient, error) { return p.get(nowFunc().Add(p.config.Timeout)) } // 獲取連接的邏輯實現(xiàn) // expire設(shè)定了一個超時時間點,當(dāng)沒有可用連接時,程序會休眠一小段時間后重試 // 如果一直獲取不到連接,一旦到達超時時間點,則報ErrOverMax錯誤 func (p *ThriftPool) get(expire time.Time) (*IdleClient, error) { if atomic.LoadUint32(&p.status) == poolStop { return nil, ErrPoolClosed } // 判斷是否超額 p.lock.Lock() if p.idle.Len() == 0 && atomic.LoadInt32(&p.count) >= p.config.MaxConn { p.lock.Unlock() // 不采用遞歸的方式來實現(xiàn)重試機制,防止棧溢出,這里改用循環(huán)方式來實現(xiàn)重試 for { // 休眠一段時間再重試 time.Sleep(p.config.interval) // 超時退出 if nowFunc().After(expire) { return nil, ErrOverMax } p.lock.Lock() if p.idle.Len() == 0 && atomic.LoadInt32(&p.count) >= p.config.MaxConn { p.lock.Unlock() } else { // 有可用鏈接,退出for循環(huán) break } } } if p.idle.Len() == 0 { // 先加1,防止首次創(chuàng)建連接時,TCP握手太久,導(dǎo)致p.count未能及時+1,而新的請求已經(jīng)到來 // 從而導(dǎo)致短暫性實際連接數(shù)大于p.count(大部分鏈接由于無法進入空閑鏈接隊列,而被關(guān)閉,處于TIME_WATI狀態(tài)) atomic.AddInt32(&p.count, 1) p.lock.Unlock() client, err := p.Dial(p.config.Addr, p.config.ConnTimeout) if err != nil { atomic.AddInt32(&p.count, -1) return nil, err } // 檢查連接是否有效 if !client.Check() { atomic.AddInt32(&p.count, -1) return nil, ErrSocketDisconnect } return client, nil } // 從隊頭中獲取空閑連接 ele := p.idle.Front() idlec := ele.Value.(*idleConn) p.idle.Remove(ele) p.lock.Unlock() // 連接從空閑隊列獲取,可能已經(jīng)關(guān)閉了,這里再重新檢查一遍 if !idlec.c.Check() { atomic.AddInt32(&p.count, -1) return nil, ErrSocketDisconnect } return idlec.c, nil }
p.Get()的邏輯比較清晰:如果空閑隊列沒有連接,且當(dāng)前連接已經(jīng)到達p.config.MaxConn,就休眠等待重試;當(dāng)滿足獲取連接條件時p.idle.Len() != 0 || atomic.LoadInt32(&p.count) < p.config.MaxConn,有空閑連接,則返回空閑連接,減少創(chuàng)建連接的開銷,沒有的話,再重新創(chuàng)建一條新的連接。
這里有兩個關(guān)鍵的地方需要注意:
等待重試的邏輯,不要用遞歸的方式來實現(xiàn),防止運行棧溢出。
// 遞歸的方法實現(xiàn)等待重試邏輯 func (p *ThriftPool) get(expire time.Time) (*IdleClient, error) { // 超時退出 if nowFunc().After(expire) { return nil, ErrOverMax } if atomic.LoadUint32(&p.status) == poolStop { return nil, ErrPoolClosed } // 判斷是否超額 p.lock.Lock() if p.idle.Len() == 0 && atomic.LoadInt32(&p.count) >= p.config.MaxConn { p.lock.Unlock() // 休眠遞歸重試 time.Sleep(p.config.interval) p.get(expire) } ....... }
注意p.lock.Lock()的和p.lock.UnLock()調(diào)用時機,確保公共數(shù)據(jù)并發(fā)操作安全。
2.3 釋放連接
// 歸還Thrift客戶端 func (p *ThriftPool) Put(client *IdleClient) error { if client == nil { return nil } if atomic.LoadUint32(&p.status) == poolStop { err := p.Close(client) client = nil return err } if atomic.LoadInt32(&p.count) > p.config.MaxConn || !client.Check() { atomic.AddInt32(&p.count, -1) err := p.Close(client) client = nil return err } p.lock.Lock() p.idle.PushFront(&idleConn{ c: client, t: nowFunc(), }) p.lock.Unlock() return nil }
p.Put()邏輯也比較簡單,如果連接已經(jīng)失效,p.count需要-1,并進行連接關(guān)閉操作。否則丟到空閑隊列里,這里還是丟到隊頭,沒錯,還是丟到隊頭,p.Get()和p.Put()都是從隊頭操作,有點像堆操作,為啥這么處理,等下面說到空閑連接超時管理就清楚了,這里先記住丟回空閑隊列的時候,會更新空閑連接的時間。
2.4 超時管理
獲取連接超時管理p.Get()方法已經(jīng)講過了,創(chuàng)建連接超時管理由p.Dial()去實現(xiàn),下面說的是空閑連接的超時管理,空閑隊列的連接,如果一直沒有使用,超過一定時間,需要主動關(guān)閉掉,服務(wù)端的資源有限,不需要用的連接就主動關(guān)掉,而且連接放太久,服務(wù)端也會主動關(guān)掉。
// 超時管理,定期釋放空閑太久的連接 func (p *ThriftPool) CheckTimeout() { p.lock.Lock() for p.idle.Len() != 0 { ele := p.idle.Back() if ele == nil { break } v := ele.Value.(*idleConn) if v.t.Add(p.config.IdleTimeout).After(nowFunc()) { break } //timeout && clear p.idle.Remove(ele) p.lock.Unlock() p.Close(v.c) //close client connection atomic.AddInt32(&p.count, -1) p.lock.Lock() } p.lock.Unlock() return }
清理超時空閑連接的時候,是從隊尾開始清理掉超時或者無效的連接,直到找到第一個可用的連接或者隊列為空。p.Get()和p.Put()都從隊頭操作隊列,保證了活躍的連接都在隊頭,如果一開始創(chuàng)建的連接太多,后面業(yè)務(wù)操作變少,不需要那么多連接的時候,那多余的連接就會沉到隊尾,被超時管理所清理掉。另外,這樣設(shè)計也可以優(yōu)化操作的時間復(fù)雜度<O(n)。
2.5 重連機制
事實上,thrift的transport層并沒有提供一個檢查連接是否有效的方法,一開始實現(xiàn)連接池的時候,檢測方法是調(diào)用thrift.TTransport.IsOpen()來判斷
// 檢測連接是否有效 func (c *IdleClient) Check() bool { if c.Transport == nil || c.RawClient == nil { return false } return c.Transport.IsOpen() }
可在測試階段發(fā)現(xiàn)當(dāng)?shù)讓赢?dāng)TCP連接被異常斷開的時候(服務(wù)端重啟、服務(wù)端宕機等),c.Transport.IsOpen()并不能如期的返回false,如果我們查看thrift的源碼,可以發(fā)現(xiàn),其實c.Transport.IsOpen()只和我們是否調(diào)用了c.Transport.Open()方法有關(guān)。為了能實現(xiàn)斷開重連機制,我們只能在使用階段發(fā)現(xiàn)異常連接時,重連連接。
這里我在ThriftPool上封裝了一層代理ThriftPoolAgent,來實現(xiàn)斷開重連邏輯,具體請參考代碼實現(xiàn)。
package pool import ( "fmt" "github.com/apache/thrift/lib/go/thrift" "log" "net" ) type ThriftPoolAgent struct { pool *ThriftPool } func NewThriftPoolAgent() *ThriftPoolAgent { return &ThriftPoolAgent{} } func (a *ThriftPoolAgent) Init(pool *ThriftPool) { a.pool = pool } // 真正的業(yè)務(wù)邏輯放到do方法做,ThriftPoolAgent只要保證獲取到可用的Thrift客戶端,然后傳給do方法就行了 func (a *ThriftPoolAgent) Do(do func(rawClient interface{}) error) error { var ( client *IdleClient err error ) defer func() { if client != nil { if err == nil { if rErr := a.releaseClient(client); rErr != nil { log.Println(fmt.Sprintf("releaseClient error: %v", rErr)) } } else if _, ok := err.(net.Error); ok { a.closeClient(client) } else if _, ok = err.(thrift.TTransportException); ok { a.closeClient(client) } else { if rErr := a.releaseClient(client); rErr != nil { log.Println(fmt.Sprintf("releaseClient error: %v", rErr)) } } } }() // 從連接池里獲取鏈接 client, err = a.getClient() if err != nil { return err } if err = do(client.RawClient); err != nil { if _, ok := err.(net.Error); ok { log.Println(fmt.Sprintf("err: retry tcp, %T, %s", err, err.Error())) // 網(wǎng)絡(luò)錯誤,重建連接 client, err = a.reconnect(client) if err != nil { return err } return do(client.RawClient) } if _, ok := err.(thrift.TTransportException); ok { log.Println(fmt.Sprintf("err: retry tcp, %T, %s", err, err.Error())) // thrift傳輸層錯誤,也重建連接 client, err = a.reconnect(client) if err != nil { return err } return do(client.RawClient) } return err } return nil } // 獲取連接 func (a *ThriftPoolAgent) getClient() (*IdleClient, error) { return a.pool.Get() } // 釋放連接 func (a *ThriftPoolAgent) releaseClient(client *IdleClient) error { return a.pool.Put(client) } // 關(guān)閉有問題的連接,并重新創(chuàng)建一個新的連接 func (a *ThriftPoolAgent) reconnect(client *IdleClient) (newClient *IdleClient, err error) { return a.pool.Reconnect(client) } // 關(guān)閉連接 func (a *ThriftPoolAgent) closeClient(client *IdleClient) { a.pool.CloseConn(client) } // 釋放連接池 func (a *ThriftPoolAgent) Release() { a.pool.Release() } func (a *ThriftPoolAgent) GetIdleCount() uint32 { return a.pool.GetIdleCount() } func (a *ThriftPoolAgent) GetConnCount() int32 { return a.pool.GetConnCount() }
3 對照實驗
啟用100個協(xié)程,不斷調(diào)用Thrift服務(wù)端API 10分鐘,對比服務(wù)平均吞吐量、Thrift API調(diào)用平均延遲、機器端口消耗。
平均吞吐量(r/s) = 總成功數(shù) / 600
API調(diào)用平均延遲(ms/r) = 總成功數(shù) / API成功請求總耗時(微秒) / 1000
機器端口消耗計算:netstat -nt | grep 9444 -c
3.1 實驗一:未使用連接池
機器端口消耗
平均吞吐量、平均延遲
從結(jié)果看,API的平均延遲在77ms左右,但是服務(wù)的平均吞吐量才到360,比理論值1000 / 77 * 1000 = 1299少了很多,而且有96409次錯誤,報錯的主要原因是:connect can't assign request address,100個協(xié)程并發(fā)調(diào)用就已經(jīng)消耗了1.6w個端口,如果并發(fā)數(shù)更高的場景,端口消耗的情況會更加嚴重,實際上,這1.6w條TCP連接,幾乎都是TIME_WAIT狀態(tài),Thrfit客戶端用完就close掉,根據(jù)TCP三次握手可知主動斷開連接的一方最終將會處于TIME_WAIT狀態(tài),并等待2MSL時間。
3.2 實驗二:使用連接池
機器端口消耗
平均吞吐量、平均延遲
可以看出,用了連接池后,平均吞吐量可達到1.8w,API調(diào)用平均延遲才0.5ms,你可能會問,理論吞吐量不是可以達到1000 / 0.5 * 100 = 20w?理論歸理論,如果按照1.8w吞吐量算,一次處理過程總時間消耗是1000 / (18000 / 100) = 5.6ms,所以這里影響吞吐量的因素已經(jīng)不是API調(diào)用的耗時了,1.8w的吞吐量其實已經(jīng)挺不錯了。
另外,消耗的端口數(shù)也才194/2 = 97(除余2是因為server端也在本地跑),而且都是ESTABLISH狀態(tài),連接一直保持著,不斷的在被復(fù)用。連接被復(fù)用,少了創(chuàng)建TCP連接的三次握手環(huán)節(jié),這里也可以解釋為啥API調(diào)用的平均延遲可以從77ms降到0.5ms,不過0.5ms確實有點低,線上環(huán)境Server一般不會和Client在同一臺機器,而且業(yè)務(wù)邏輯也會比這里復(fù)雜,API調(diào)用的平均延遲會相對高一點。
4 總結(jié)
調(diào)用Thrift API必須使用Thrift客戶端連接池,否則在高并發(fā)的情況下,會有大量的TCP連接處于TIME_WAIT狀態(tài),機器端口被大量消耗,可能會導(dǎo)致部分請求失敗甚至服務(wù)不可用。每次請求都重新創(chuàng)建TCP連接,進行TCP三次握手環(huán)節(jié),API調(diào)用的延遲會比較高,服務(wù)的吞吐量也不會很高。
使用Thrift客戶端連接池,可以提高系統(tǒng)的吞吐量,同時可以避免機器端口被耗盡的危險,提高服務(wù)的可靠性。
以上為個人經(jīng)驗,希望能給大家一個參考,也希望大家多多支持腳本之家。如有錯誤或未考慮完全的地方,望不吝賜教。
相關(guān)文章
深入理解Go語言設(shè)計模式之函數(shù)式選項模式
在 Go 語言中,函數(shù)選項模式(Function Options Pattern)是一種常見且強大的設(shè)計模式,用于構(gòu)建可擴展、易于使用和靈活的 API,本文就來看看它的具體用法吧2023-05-05