使用Golang怎么制作一個(gè)Thrift客戶端連接池

發(fā)布時(shí)間：2020-12-19 13:52:38 來源：億速云閱讀：266 作者：Leah 欄目：開發(fā)技術(shù)

今天就跟大家聊聊有關(guān)使用Golang怎么制作一個(gè)Thrift客戶端連接池，可能很多人都不太了解，為了讓大家更加了解，小編給大家總結(jié)了以下內(nèi)容，希望大家根據(jù)這篇文章可以有所收獲。

1.1 運(yùn)行環(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)
}

注：請(qǐng)通過安裝Thrift IDL編譯器，并生成客戶端、服務(wù)端代碼。

1.3 對(duì)照實(shí)驗(yàn)說明

通過腳本開啟100個(gè)協(xié)程并發(fā)調(diào)用rpc服務(wù)10分鐘，統(tǒng)計(jì)這段時(shí)間內(nèi)，未使用thrift客戶端連接池與使用客戶端連接池服務(wù)的平均吞吐量、Thrift API調(diào)用平均延遲、機(jī)器端口消耗等數(shù)據(jù)進(jìn)行性能對(duì)比。

實(shí)驗(yàn)一：未使用thrift客戶端連接池

實(shí)驗(yàn)二：使用thrift客戶端連接池

2 Thrift客戶端連接池實(shí)現(xiàn)

2.1 連接池的功能

首先，我們要明確一下連接池的職責(zé)，這里我簡(jiǎn)單的總結(jié)一下，連接池主要功能是維護(hù)連接的創(chuàng)建、釋放，通過緩存連接來復(fù)用連接，減少創(chuàng)建連接所帶來的開銷，提高系統(tǒng)的吞吐量，一般連接池還會(huì)有連接斷開的重連機(jī)制、超時(shí)機(jī)制等。這里我們可以先定義出大部分連接池都會(huì)有的功能，只是定義，可以先不管每個(gè)功能的具體實(shí)現(xiàn)。每一個(gè)空閑Thrift客戶端其實(shí)底層都維護(hù)著一條空閑TCP連接，空閑Thrift客戶端與空閑連接在這里其實(shí)是同一個(gè)概念。

......
// Thrift客戶端創(chuàng)建方法，留給業(yè)務(wù)去實(shí)現(xiàn)
type ThriftDial func(addr string, connTimeout time.Duration) (*IdleClient, error)
// 關(guān)閉Thrift客戶端，留給業(yè)務(wù)實(shí)現(xiàn)
type ThriftClientClose func(c *IdleClient) error
// Thrift客戶端連接池
type ThriftPool struct {
 // Thrift客戶端創(chuàng)建邏輯，業(yè)務(wù)自己實(shí)現(xiàn)
 Dial ThriftDial
 // Thrift客戶端關(guān)閉邏輯，業(yè)務(wù)自己實(shí)現(xiàn)
 Close ThriftClientClose
 // 空閑客戶端，用雙端隊(duì)列存儲(chǔ)
 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)建連接超時(shí)時(shí)間
 ConnTimeout time.Duration
 // 空閑客戶端超時(shí)時(shí)間，超時(shí)主動(dòng)釋放連接，關(guān)閉客戶端
 IdleTimeout time.Duration
 // 獲取Thrift客戶端超時(shí)時(shí)間
 Timeout time.Duration
 // 獲取Thrift客戶端失敗重試間隔
 interval time.Duration
}
// Thrift客戶端
type IdleClient struct {
 // Thrift傳輸層，封裝了底層連接建立、維護(hù)、關(guān)閉、數(shù)據(jù)讀寫等細(xì)節(jié)
 Transport thrift.TTransport
 // 真正的Thrift客戶端，業(yè)務(wù)創(chuàng)建傳入
 RawClient interface{}
}
// 封裝了Thrift客戶端
type idleConn struct {
 // 空閑Thrift客戶端
 c *IdleClient
 // 最近一次放入空閑隊(duì)列的時(shí)間
 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客戶端丟進(jìn)空閑連接池，更新數(shù)據(jù)，返回
 ......
}
// 超時(shí)管理，定期釋放空閑太久的連接
func (p *ThriftPool) CheckTimeout() {
 // 掃描空閑連接池，將空閑太久的連接主動(dòng)釋放掉，并更新數(shù)據(jù)
 ......
}
// 異常連接重連
func (p *ThriftPool) Reconnect(client *IdleClient) (newClient *IdleClient, err error) {
 // 1. 關(guān)閉舊客戶端
 // 2. 創(chuàng)建新的客戶端，并返回
 ......
}
// 其他方法
......

這里有兩個(gè)關(guān)鍵的數(shù)據(jù)結(jié)構(gòu)，ThriftPool和IdleClient，ThriftPool負(fù)責(zé)實(shí)現(xiàn)整個(gè)連接池的功能，IdleClient封裝了真正的Thrift客戶端。

先看一下ThriftPool的定義：

// Thrift客戶端創(chuàng)建方法，留給業(yè)務(wù)去實(shí)現(xiàn)
type ThriftDial func(addr string, connTimeout time.Duration) (*IdleClient, error)
// 關(guān)閉Thrift客戶端，留給業(yè)務(wù)實(shí)現(xiàn)
type ThriftClientClose func(c *IdleClient) error
// Thrift客戶端連接池
type ThriftPool struct {
 // Thrift客戶端創(chuàng)建邏輯，業(yè)務(wù)自己實(shí)現(xiàn)
 Dial ThriftDial
 // Thrift客戶端關(guān)閉邏輯，業(yè)務(wù)自己實(shí)現(xiàn)
 Close ThriftClientClose
 // 空閑客戶端，用雙端隊(duì)列存儲(chǔ)
 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)建連接超時(shí)時(shí)間
 ConnTimeout time.Duration
 // 空閑客戶端超時(shí)時(shí)間，超時(shí)主動(dòng)釋放連接，關(guān)閉客戶端
 IdleTimeout time.Duration
 // 獲取Thrift客戶端超時(shí)時(shí)間
 Timeout time.Duration
 // 獲取Thrift客戶端失敗重試間隔
 interval time.Duration
}

Thrift客戶端創(chuàng)建與關(guān)閉，涉及到業(yè)務(wù)細(xì)節(jié)，這里抽離成Dial方法和Close方法。

連接池需要維護(hù)空閑客戶端，這里用雙端隊(duì)列來存儲(chǔ)。

一般的連接池，都應(yīng)該支持最大連接數(shù)配置，MaxConn可以配置連接池最大連接數(shù)，同時(shí)我們用count來記錄連接池當(dāng)前已經(jīng)創(chuàng)建的連接。

為了實(shí)現(xiàn)連接池的超時(shí)管理，當(dāng)然也得有相關(guān)超時(shí)配置。

連接池的狀態(tài)、當(dāng)前連接數(shù)等這些屬性，是多協(xié)程并發(fā)操作的，這里用同步鎖lock來確保并發(fā)操作安全。

在看一下IdleClient實(shí)現(xiàn)：

// Thrift客戶端
type IdleClient struct {
 // Thrift傳輸層，封裝了底層連接建立、維護(hù)、關(guān)閉、數(shù)據(jù)讀寫等細(xì)節(jié)
 Transport thrift.TTransport
 // 真正的Thrift客戶端，業(yè)務(wù)創(chuàng)建傳入
 RawClient interface{}
}
// 封裝了Thrift客戶端
type idleConn struct {
 // 空閑Thrift客戶端
 c *IdleClient
 // 最近一次放入空閑隊(duì)列的時(shí)間
 t time.Time
}

RawClient是真正的Thrift客戶端，與實(shí)際邏輯相關(guān)。

Transport Thrift傳輸層，Thrift傳輸層，封裝了底層連接建立、維護(hù)、關(guān)閉、數(shù)據(jù)讀寫等細(xì)節(jié)。

idleConn封裝了IdleClient，用來實(shí)現(xiàn)空閑連接超時(shí)管理，idleConn記錄一個(gè)時(shí)間，這個(gè)時(shí)間是Thrift客戶端最近一次被放入空閑隊(duì)列的時(shí)間。

2.2 獲取連接

......
var nowFunc = time.Now
......
// 獲取Thrift空閑客戶端
func (p *ThriftPool) Get() (*IdleClient, error) {
 return p.get(nowFunc().Add(p.config.Timeout))
}
// 獲取連接的邏輯實(shí)現(xiàn)
// expire設(shè)定了一個(gè)超時(shí)時(shí)間點(diǎn)，當(dāng)沒有可用連接時(shí)，程序會(huì)休眠一小段時(shí)間后重試
// 如果一直獲取不到連接，一旦到達(dá)超時(shí)時(shí)間點(diǎn)，則報(bào)ErrOverMax錯(cuò)誤
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()
 // 不采用遞歸的方式來實(shí)現(xiàn)重試機(jī)制，防止棧溢出，這里改用循環(huán)方式來實(shí)現(xiàn)重試
 for {
 // 休眠一段時(shí)間再重試
 time.Sleep(p.config.interval)
 // 超時(shí)退出
 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)建連接時(shí)，TCP握手太久，導(dǎo)致p.count未能及時(shí)+1，而新的請(qǐng)求已經(jīng)到來
 // 從而導(dǎo)致短暫性實(shí)際連接數(shù)大于p.count（大部分鏈接由于無法進(jìn)入空閑鏈接隊(duì)列，而被關(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
 }
 // 從隊(duì)頭中獲取空閑連接
 ele := p.idle.Front()
 idlec := ele.Value.(*idleConn)
 p.idle.Remove(ele)
 p.lock.Unlock()
 // 連接從空閑隊(duì)列獲取，可能已經(jīng)關(guān)閉了，這里再重新檢查一遍
 if !idlec.c.Check() {
 atomic.AddInt32(&p.count, -1)
 return nil, ErrSocketDisconnect
 }
 return idlec.c, nil
}

p.Get()的邏輯比較清晰：如果空閑隊(duì)列沒有連接，且當(dāng)前連接已經(jīng)到達(dá)p.config.MaxConn，就休眠等待重試；當(dāng)滿足獲取連接條件時(shí)p.idle.Len() != 0 || atomic.LoadInt32(&p.count) < p.config.MaxConn，有空閑連接，則返回空閑連接，減少創(chuàng)建連接的開銷，沒有的話，再重新創(chuàng)建一條新的連接。

這里有兩個(gè)關(guān)鍵的地方需要注意：

等待重試的邏輯，不要用遞歸的方式來實(shí)現(xiàn)，防止運(yùn)行棧溢出。

// 遞歸的方法實(shí)現(xiàn)等待重試邏輯
func (p *ThriftPool) get(expire time.Time) (*IdleClient, error) {
 // 超時(shí)退出
 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ī)，確保公共數(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()邏輯也比較簡(jiǎn)單，如果連接已經(jīng)失效，p.count需要-1，并進(jìn)行連接關(guān)閉操作。否則丟到空閑隊(duì)列里，這里還是丟到隊(duì)頭，沒錯(cuò)，還是丟到隊(duì)頭，p.Get()和p.Put()都是從隊(duì)頭操作，有點(diǎn)像堆操作，為啥這么處理，等下面說到空閑連接超時(shí)管理就清楚了，這里先記住丟回空閑隊(duì)列的時(shí)候，會(huì)更新空閑連接的時(shí)間。

2.4 超時(shí)管理

獲取連接超時(shí)管理p.Get()方法已經(jīng)講過了，創(chuàng)建連接超時(shí)管理由p.Dial()去實(shí)現(xiàn)，下面說的是空閑連接的超時(shí)管理，空閑隊(duì)列的連接，如果一直沒有使用，超過一定時(shí)間，需要主動(dòng)關(guān)閉掉，服務(wù)端的資源有限，不需要用的連接就主動(dòng)關(guān)掉，而且連接放太久，服務(wù)端也會(huì)主動(dòng)關(guān)掉。

// 超時(shí)管理，定期釋放空閑太久的連接
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
}

清理超時(shí)空閑連接的時(shí)候，是從隊(duì)尾開始清理掉超時(shí)或者無效的連接，直到找到第一個(gè)可用的連接或者隊(duì)列為空。p.Get()和p.Put()都從隊(duì)頭操作隊(duì)列，保證了活躍的連接都在隊(duì)頭，如果一開始創(chuàng)建的連接太多，后面業(yè)務(wù)操作變少，不需要那么多連接的時(shí)候，那多余的連接就會(huì)沉到隊(duì)尾，被超時(shí)管理所清理掉。另外，這樣設(shè)計(jì)也可以優(yōu)化操作的時(shí)間復(fù)雜度<O(n)。

2.5 重連機(jī)制

事實(shí)上，thrift的transport層并沒有提供一個(gè)檢查連接是否有效的方法，一開始實(shí)現(xiàn)連接池的時(shí)候，檢測(cè)方法是調(diào)用thrift.TTransport.IsOpen()來判斷

// 檢測(cè)連接是否有效
func (c *IdleClient) Check() bool {
 if c.Transport == nil || c.RawClient == nil {
 return false
 }
 return c.Transport.IsOpen()
}

可在測(cè)試階段發(fā)現(xiàn)當(dāng)?shù)讓赢?dāng)TCP連接被異常斷開的時(shí)候（服務(wù)端重啟、服務(wù)端宕機(jī)等），c.Transport.IsOpen()并不能如期的返回false，如果我們查看thrift的源碼，可以發(fā)現(xiàn)，其實(shí)c.Transport.IsOpen()只和我們是否調(diào)用了c.Transport.Open()方法有關(guān)。為了能實(shí)現(xiàn)斷開重連機(jī)制，我們只能在使用階段發(fā)現(xiàn)異常連接時(shí)，重連連接。

這里我在ThriftPool上封裝了一層代理ThriftPoolAgent，來實(shí)現(xiàn)斷開重連邏輯，具體請(qǐng)參考代碼實(shí)現(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ò)錯(cuò)誤，重建連接
 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傳輸層錯(cuò)誤，也重建連接
 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)建一個(gè)新的連接
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 對(duì)照實(shí)驗(yàn)

啟用100個(gè)協(xié)程，不斷調(diào)用Thrift服務(wù)端API 10分鐘，對(duì)比服務(wù)平均吞吐量、Thrift API調(diào)用平均延遲、機(jī)器端口消耗。

平均吞吐量（r/s） = 總成功數(shù) / 600

API調(diào)用平均延遲（ms/r） = 總成功數(shù) / API成功請(qǐng)求總耗時(shí)（微秒） / 1000

機(jī)器端口消耗計(jì)算：netstat -nt | grep 9444 -c

3.1 實(shí)驗(yàn)一：未使用連接池

機(jī)器端口消耗