Golang連線池的幾種實現案例小結
因為TCP的三隻握手等等原因，建立一個連線是一件成本比較高的行為。所以在一個需要多次與特定實體互動的程式中，就需要維持一個連線池，裡面有可以複用的連線可供重複使用。
而維持一個連線池，最基本的要求就是要做到：thread safe（執行緒安全），尤其是在Golang這種特性是goroutine的語言中。
實現簡單的連線池
type Pool struct {
m sync.Mutex // 保證多個goroutine訪問時候，closed的執行緒安全
res chan io.Closer //連線儲存的chan
factory func() (io.Closer,error) //新建連線的工廠方法
closed bool //連線池關閉標誌
}
這個簡單的連線池，我們利用chan來儲存池裡的連線。而新建結構體的方法也比較簡單：
func New(fn func() (io.Closer,error),size uint) (*Pool,error) {
if size <= 0 {
return nil,errors.New("size的值太小了。")
}
return &Pool{
factory: fn,res: make(chan io.Closer,size),},nil
}
只需要提供對應的工廠函式和連線池的大小就可以了。
獲取連線
那麼我們要怎麼從中獲取資源呢？因為我們內部儲存連線的結構是chan，所以只需要簡單的select就可以保證執行緒安全：
//從資源池裡獲取一個資源
func (p *Pool) Acquire() (io.Closer,error) {
select {
case r,ok := <-p.res:
log.Println("Acquire:共享資源")
if !ok {
return nil,ErrPoolClosed
}
return r,nil
default:
log.Println("Acquire:新生成資源")
return p.factory()
}
}
我們先從連線池的res這個chan裡面獲取，如果沒有的話我們就利用我們早已經準備好的工廠函式進行構造連線。同時我們在從res獲取連線的時候利用ok先確定了這個連線池是否已經關閉。如果已經關閉的話我們就返回早已經準備好的連線已關閉錯誤。
關閉連線池
那麼既然提到關閉連線池，我們是怎麼樣關閉連線池的呢？
//關閉資源池，釋放資源
func (p *Pool) Close() {
p.m.Lock()
defer p.m.Unlock()
if p.closed {
return
}
p.closed = true
//關閉通道，不讓寫入了
close(p.res)
//關閉通道里的資源
for r:=range p.res {
r.Close()
}
}
這邊我們需要先進行p.m.Lock()上鎖操作，這麼做是因為我們需要對結構體裡面的closed進行讀寫。需要先把這個標誌位設定後，關閉res這個chan，使得Acquire方法無法再獲取新的連線。我們再對res這個chan裡面的連線進行Close操作。
釋放連線
釋放連線首先得有個前提，就是連線池還沒有關閉。如果連線池已經關閉再往res裡面送連線的話就好觸發panic。
func (p *Pool) Release(r io.Closer){
//保證該操作和Close方法的操作是安全的
p.m.Lock()
defer p.m.Unlock()
//資源池都關閉了，就省這一個沒有釋放的資源了，釋放即可
if p.closed {
r.Close()
return
}
select {
case p.res <- r:
log.Println("資源釋放到池子裡了")
default:
log.Println("資源池滿了，釋放這個資源吧")
r.Close()
}
}
以上就是一個簡單且執行緒安全的連線池實現方式了。我們可以看到的是，現在連線池雖然已經實現了，但是還有幾個小缺點:
- 我們對連線最大的數量沒有限制，如果執行緒池空的話都我們預設就直接新建一個連線返回了。一旦併發量高的話將會不斷新建連線，很容易（尤其是MySQL）造成too many connections的報錯發生。
- 既然我們需要保證最大可獲取連線數量，那麼我們就不希望數量定的太死。希望空閒的時候可以維護一定的空閒連線數量idleNum，但是又希望我們能限制最大可獲取連線數量maxNum。
- 第一種情況是併發過多的情況，那麼如果併發量過少呢？現在我們在新建一個連線並且歸還後，我們很長一段時間不再使用這個連線。那麼這個連線很有可能在幾個小時甚至更長時間之前就已經建立的了。長時間閒置的連線我們並沒有辦法保證它的可用性。便有可能我們下次獲取的連線是已經失效的連線。
那麼我們可以從已經成熟使用的MySQL連線池庫和Redis連線池庫中看看，它們是怎麼解決這些問題的。
Golang標準庫的Sql連線池
Golang的連線池實現在標準庫database/sql/sql.go下。當我們執行：
db,err := sql.Open("mysql","xxxx")
的時候，就會開啟一個連線池。我們可以看看返回的db的結構體：
type DB struct {
waitDuration int64 // Total time waited for new connections.
mu sync.Mutex // protects following fields
freeConn []*driverConn
connRequests map[uint64]chan connRequest
nextRequest uint64 // Next key to use in connRequests.
numOpen int // number of opened and pending open connections
// Used to signal the need for new connections
// a goroutine running connectionOpener() reads on this chan and
// maybeOpenNewConnections sends on the chan (one send per needed connection)
// It is closed during db.Close(). The close tells the connectionOpener
// goroutine to exit.
openerCh chan struct{}
closed bool
maxIdle int // zero means defaultMaxIdleConns; negative means 0
maxOpen int // <= 0 means unlimited
maxLifetime time.Duration // maximum amount of time a connection may be reused
cleanerCh chan struct{}
waitCount int64 // Total number of connections waited for.
maxIdleClosed int64 // Total number of connections closed due to idle.
maxLifetimeClosed int64 // Total number of connections closed due to max free limit.
}
上面省去了一些暫時不需要關注的field。我們可以看的，DB這個連線池內部儲存連線的結構freeConn，並不是我們之前使用的chan，而是**[]driverConn**，一個連線切片。同時我們還可以看到，裡面有maxIdle等相關變數來控制空閒連線數量。值得注意的是，DB的初始化函式Open函式並沒有新建資料庫連線。而新建連線在哪個函式呢？我們可以在Query方法一路往回找，我們可以看到這個函式：func (db *DB) conn(ctx context.Context,strategy connReuseStrategy) (*driverConn,error)。而我們從連線池獲取連線的方法，就從這裡開始：
獲取連線
// conn returns a newly-opened or cached *driverConn.
func (db *DB) conn(ctx context.Context,error) {
// 先判斷db是否已經關閉。
db.mu.Lock()
if db.closed {
db.mu.Unlock()
return nil,errDBClosed
}
// 注意檢測context是否已經被超時等原因被取消。
select {
default:
case <-ctx.Done():
db.mu.Unlock()
return nil,ctx.Err()
}
lifetime := db.maxLifetime
// 這邊如果在freeConn這個切片有空閒連線的話，就left pop一個出列。注意的是，這邊因為是切片操作，所以需要前面需要加鎖且獲取後進行解鎖操作。同時判斷返回的連線是否已經過期。
numFree := len(db.freeConn)
if strategy == cachedOrNewConn && numFree > 0 {
conn := db.freeConn[0]
copy(db.freeConn,db.freeConn[1:])
db.freeConn = db.freeConn[:numFree-1]
conn.inUse = true
db.mu.Unlock()
if conn.expired(lifetime) {
conn.Close()
return nil,driver.ErrBadConn
}
// Lock around reading lastErr to ensure the session resetter finished.
conn.Lock()
err := conn.lastErr
conn.Unlock()
if err == driver.ErrBadConn {
conn.Close()
return nil,driver.ErrBadConn
}
return conn,nil
}
// 這邊就是等候獲取連線的重點了。當空閒的連線為空的時候，這邊將會新建一個request（的等待連線 的請求）並且開始等待
if db.maxOpen > 0 && db.numOpen >= db.maxOpen {
// 下面的動作相當於往connRequests這個map插入自己的號碼牌。
// 插入號碼牌之後這邊就不需要阻塞等待繼續往下走邏輯。
req := make(chan connRequest,1)
reqKey := db.nextRequestKeyLocked()
db.connRequests[reqKey] = req
db.waitCount++
db.mu.Unlock()
waitStart := time.Now()
// Timeout the connection request with the context.
select {
case <-ctx.Done():
// context取消操作的時候，記得從connRequests這個map取走自己的號碼牌。
db.mu.Lock()
delete(db.connRequests,reqKey)
db.mu.Unlock()
atomic.AddInt64(&db.waitDuration,int64(time.Since(waitStart)))
select {
default:
case ret,ok := <-req:
// 這邊值得注意了，因為現在已經被context取消了。但是剛剛放了自己的號碼牌進去排隊裡面。意思是說不定已經發了連線了，所以得注意歸還！
if ok && ret.conn != nil {
db.putConn(ret.conn,ret.err,false)
}
}
return nil,ctx.Err()
case ret,ok := <-req:
// 下面是已經獲得連線後的操作了。檢測一下獲得連線的狀況。因為有可能已經過期了等等。
atomic.AddInt64(&db.waitDuration,int64(time.Since(waitStart)))
if !ok {
return nil,errDBClosed
}
if ret.err == nil && ret.conn.expired(lifetime) {
ret.conn.Close()
return nil,driver.ErrBadConn
}
if ret.conn == nil {
return nil,ret.err
}
ret.conn.Lock()
err := ret.conn.lastErr
ret.conn.Unlock()
if err == driver.ErrBadConn {
ret.conn.Close()
return nil,driver.ErrBadConn
}
return ret.conn,ret.err
}
}
// 下面就是如果上面說的限制情況不存在，可以建立先連線時候，要做的建立連線操作了。
db.numOpen++ // optimistically
db.mu.Unlock()
ci,err := db.connector.Connect(ctx)
if err != nil {
db.mu.Lock()
db.numOpen-- // correct for earlier optimism
db.maybeOpenNewConnections()
db.mu.Unlock()
return nil,err
}
db.mu.Lock()
dc := &driverConn{
db: db,createdAt: nowFunc(),ci: ci,inUse: true,}
db.addDepLocked(dc,dc)
db.mu.Unlock()
return dc,nil
}
簡單來說，DB結構體除了用的是slice來儲存連線，還加了一個類似排隊機制的connRequests來解決獲取等待連線的過程。同時在判斷連線健康性都有很好的兼顧。那麼既然有了排隊機制，歸還連線的時候是怎麼做的呢？
釋放連線
我們可以直接找到func (db *DB) putConnDBLocked(dc *driverConn,err error) bool這個方法。就像註釋說的，這個方法主要的目的是：
Satisfy a connRequest or put the driverConn in the idle pool and return true or return false.
我們主要來看看裡面重點那幾行：
...
// 如果已經超過最大開啟數量了，就不需要在迴歸pool了
if db.maxOpen > 0 && db.numOpen > db.maxOpen {
return false
}
// 這邊是重點了，基本來說就是從connRequest這個map裡面隨機抽一個在排隊等著的請求。取出來後發給他。就不用歸還池子了。
if c := len(db.connRequests); c > 0 {
var req chan connRequest
var reqKey uint64
for reqKey,req = range db.connRequests {
break
}
delete(db.connRequests,reqKey) // 刪除這個在排隊的請求。
if err == nil {
dc.inUse = true
}
// 把連線給這個正在排隊的連線。
req <- connRequest{
conn: dc,err: err,}
return true
} else if err == nil && !db.closed {
// 既然沒人排隊，就看看到了最大連線數目沒有。沒到就歸還給freeConn。
if db.maxIdleConnsLocked() > len(db.freeConn) {
db.freeConn = append(db.freeConn,dc)
db.startCleanerLocked()
return true
}
db.maxIdleClosed++
}
...
我們可以看到，當歸還連線時候，如果有在排隊輪候的請求就不歸還給池子直接發給在輪候的人了。
現在基本就解決前面說的小問題了。不會出現連線太多導致無法控制too many connections的情況。也很好了維持了連線池的最小數量。同時也做了相關對於連線健康性的檢查操作。
值得注意的是，作為標準庫的程式碼，相關注釋和程式碼都非常完美，真的可以看的神清氣爽。
redis Golang實現的Redis客戶端
這個Golang實現的Redis客戶端，是怎麼實現連線池的。這邊的思路非常奇妙，還是能學習到不少好思路。當然了，由於程式碼註釋比較少，啃起來第一下還是有點迷糊的。相關程式碼地址在https://github.com/go-redis/redis/blob/master/internal/pool/pool.go 可以看到。
而它的連線池結構如下
type ConnPool struct {
...
queue chan struct{}
connsMu sync.Mutex
conns []*Conn
idleConns []*Conn
poolSize int
idleConnsLen int
stats Stats
_closed uint32 // atomic
closedCh chan struct{}
}
我們可以看到裡面儲存連線的結構還是slice。但是我們可以重點看看queue，conns，idleConns這幾個變數，後面會提及到。但是值得注意的是！我們可以看到，這裡有兩個**[]Conn**結構：conns、idleConns，那麼問題來了：
到底連線存在哪裡？
新建連線池連線
我們先從新建連線池連線開始看：
func NewConnPool(opt *Options) *ConnPool {
....
p.checkMinIdleConns()
if opt.IdleTimeout > 0 && opt.IdleCheckFrequency > 0 {
go p.reaper(opt.IdleCheckFrequency)
}
....
}
初始化連線池的函式有個和前面兩個不同的地方。
- checkMinIdleConns方法，在連線池初始化的時候就會往連線池填滿空閒的連線。
- go p.reaper(opt.IdleCheckFrequency)則會在初始化連線池的時候就會起一個go程，週期性的淘汰連線池裡面要被淘汰的連線。
獲取連線
func (p *ConnPool) Get(ctx context.Context) (*Conn,error) {
if p.closed() {
return nil,ErrClosed
}
//這邊和前面sql獲取連線函式的流程先不同。sql是先看看連線池有沒有空閒連線，有的話先獲取不到再排隊。這邊是直接先排隊獲取令牌，排隊函式後面會分析。
err := p.waitTurn(ctx)
if err != nil {
return nil,err
}
//前面沒出error的話，就已經排隊輪候到了。接下來就是獲取的流程。
for {
p.connsMu.Lock()
//從空閒連線裡面先獲取一個空閒連線。
cn := p.popIdle()
p.connsMu.Unlock()
if cn == nil {
// 沒有空閒連線時候直接跳出迴圈。
break
}
// 判斷是否已經過時，是的話close掉了然後繼續取出。
if p.isStaleConn(cn) {
_ = p.CloseConn(cn)
continue
}
atomic.AddUint32(&p.stats.Hits,1)
return cn,nil
}
atomic.AddUint32(&p.stats.Misses,1)
// 如果沒有空閒連線的話，這邊就直接新建連線了。
newcn,err := p.newConn(ctx,true)
if err != nil {
// 歸還令牌。
p.freeTurn()
return nil,err
}
return newcn,nil
}
我們可以試著回答開頭那個問題：連線到底存在哪裡？答案是從cn := p.popIdle()這句話可以看出，獲取連線這個動作，是從idleConns裡面獲取的，而裡面的函式也證明了這一點。但是，真的是這樣的嘛？我們後面再看看。
同時我的理解是：
- sql的排隊意味著我對連線池申請連線後，把自己的編號告訴連線池。連線那邊一看到有空閒了，就叫我的號。我答應了一聲，然後連線池就直接給個連線給我。我如果不歸還，連線池就一直不叫下一個號。
- redis這邊的意思是，我去和連線池申請的不是連線而是令牌。我就一直排隊等著，連線池給我令牌了，我才去倉庫裡面找空閒連線或者自己新建一個連線。用完了連線除了歸還連線外，還得歸還令牌。當然了，如果我自己新建連接出錯了，我哪怕拿不到連接回家，我也得把令牌給回連線池，不然連線池的令牌數少了，最大連線數也會變小。
而：
func (p *ConnPool) freeTurn() {
<-p.queue
}
func (p *ConnPool) waitTurn(ctx context.Context) error {
...
case p.queue <- struct{}{}:
return nil
...
}
就是在靠queue這個chan來維持令牌數量。
那麼conns的作用是什麼呢？我們可以來看看新建連線這個函式：
新建連線
func (p *ConnPool) newConn(ctx context.Context,pooled bool) (*Conn,error) {
cn,err := p.dialConn(ctx,pooled)
if err != nil {
return nil,err
}
p.connsMu.Lock()
p.conns = append(p.conns,cn)
if pooled {
// 如果連線池滿了，會在後面移除。
if p.poolSize >= p.opt.PoolSize {
cn.pooled = false
} else {
p.poolSize++
}
}
p.connsMu.Unlock()
return cn,nil
}
基本邏輯出來了。就是如果新建連線的話，我並不會直接放在idleConns裡面，而是先放conns裡面。同時先看池子滿了沒有。滿的話後面歸還的時候會標記，後面會刪除。那麼這個後面會刪除，指的是什麼時候呢？那就是下面說的歸還連線的時候了。
歸還連線
func (p *ConnPool) Put(cn *Conn) {
if cn.rd.Buffered() > 0 {
internal.Logger.Printf("Conn has unread data")
p.Remove(cn,BadConnError{})
return
}
//這就是我們剛剛說的後面了，前面標記過不要入池的，這邊就刪除了。當然了，裡面也會進行freeTurn操作。
if !cn.pooled {
// 這個方法就是前面的標誌位，判斷裡面可以知道，前面標誌不要池化的，這裡會將它刪除。
p.Remove(cn,nil)
return
}
p.connsMu.Lock()
p.idleConns = append(p.idleConns,cn)
p.idleConnsLen++
p.connsMu.Unlock()
//我們可以看到很明顯的這個歸還號碼牌的動作。
p.freeTurn()
}
答案就是，所有的連線其實是存放在conns這個切片裡面。如果這個連線是空閒等待的狀態的話，那就在idleConns裡面加一個自己的指標！
其實歸還的過程，就是檢查一下我打算還的這個連線，是不是超售的產物，如果是就沒必要池化了，直接刪除就可以了。不是的話，就是把連線自身（一個指標）在idleConns也append一下。
等等，上面的邏輯似乎有點不對？我們來理一下獲取連線流程：
- 先waitTurn，拿到令牌。而令牌數量是根據pool裡面的queue決定的。
- 拿到令牌了，去庫房idleConns裡面拿空閒的連線。沒有的話就自己newConn一個，並且把他記錄到conns裡面。
- 用完了，就呼叫put歸還：也就是從conns新增這個連線的指標到idleConns。歸還的時候就檢查在newConn時候是不是已經做了超賣標記了。是的話就不轉移到idleConns。
我當時疑惑了好久，既然始終都需要獲得令牌才能得到連線，令牌數量是定的。為什麼還會超賣呢？翻了一下原始碼，我的答案是：
雖然Get方法獲取連線是newConn這個私用方法，受到令牌管制導致不會出現超賣。但是這個方法接受傳參：pooled bool。所以我猜是擔心其他人呼叫這個方法時候，不管三七二十一就傳了true，導致poolSize越來越大。
總的來說，redis這個連線池的連線數控制，還是在queue這個我稱為令牌的chan進行操作。
總結
上面可以看到，連線池的最基本的保證，就是獲取連線時候的執行緒安全。但是在實現諸多額外特性時候卻又從不同角度來實現。還是非常有意思的。但是不管儲存結構是用chan還是還是slice，都可以很好的實現這一點。如果像sql或者redis那樣用slice來儲存連線，就得維護一個結構來表示排隊等候的效果。
到此這篇關於Golang連線池的幾種實現案例小結的文章就介紹到這了,更多相關Golang 連線池內容請搜尋我們以前的文章或繼續瀏覽下面的相關文章希望大家以後多多支援我們！