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負(fù)責(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ù)更高的場景，端口消耗的情況會更加嚴(yán)重，實際上，這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)驗，希望能給大家一個參考，也希望大家多多支持腳本之家。如有錯誤或未考慮完全的地方，望不吝賜教。
您可能感興趣的文章:- golang 通過ssh代理連接mysql的操作
- 淺談golang結(jié)構(gòu)體偷懶初始化
- golang連接kafka消費進ES操作
- golang實現(xiàn)各種情況的get請求操作
- Golang 實現(xiàn)分片讀取http超大文件流和并發(fā)控制
- 在Golang中使用http.FileServer返回靜態(tài)文件的操作
- golang 生成定單號的操作
