Golang連接池的幾種實(shí)現(xiàn)案例小結(jié)

type Pool struct {
 m sync.Mutex // 保證多個(gè)goroutine訪問(wèn)時(shí)候，closed的線程安全
 res chan io.Closer //連接存儲(chǔ)的chan
 factory func() (io.Closer,error) //新建連接的工廠方法
 closed bool //連接池關(guān)閉標(biāo)志
}

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
}

//從資源池里獲取一個(gè)資源
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()
 }
}

//關(guān)閉資源池，釋放資源
func (p *Pool) Close() {
 p.m.Lock()
 defer p.m.Unlock()

 if p.closed {
 return
 }

 p.closed = true

 //關(guān)閉通道，不讓寫(xiě)入了
 close(p.res)

 //關(guān)閉通道里的資源
 for r:=range p.res {
 r.Close()
 }
}

func (p *Pool) Release(r io.Closer){
 //保證該操作和Close方法的操作是安全的
 p.m.Lock()
 defer p.m.Unlock()

 //資源池都關(guān)閉了，就省這一個(gè)沒(méi)有釋放的資源了，釋放即可
 if p.closed {
 r.Close()
 return
 }

 select {
 case p.res - r:
 log.Println("資源釋放到池子里了")
 default:
 log.Println("資源池滿了，釋放這個(gè)資源吧")
 r.Close()
 }
}

db, err := sql.Open("mysql", "xxxx")

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.
}

// conn returns a newly-opened or cached *driverConn.
func (db *DB) conn(ctx context.Context, strategy connReuseStrategy) (*driverConn, error) {
 // 先判斷db是否已經(jīng)關(guān)閉。
 db.mu.Lock()
 if db.closed {
 db.mu.Unlock()
 return nil, errDBClosed
 }
 // 注意檢測(cè)context是否已經(jīng)被超時(shí)等原因被取消。
 select {
 default:
 case -ctx.Done():
 db.mu.Unlock()
 return nil, ctx.Err()
 }
 lifetime := db.maxLifetime

 // 這邊如果在freeConn這個(gè)切片有空閑連接的話，就left pop一個(gè)出列。注意的是，這邊因?yàn)槭乔衅僮?，所以需要前面需要加鎖且獲取后進(jìn)行解鎖操作。同時(shí)判斷返回的連接是否已經(jīng)過(guò)期。
 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
 }

 // 這邊就是等候獲取連接的重點(diǎn)了。當(dāng)空閑的連接為空的時(shí)候，這邊將會(huì)新建一個(gè)request（的等待連接 的請(qǐng)求）并且開(kāi)始等待
 if db.maxOpen > 0  db.numOpen >= db.maxOpen {
 // 下面的動(dòng)作相當(dāng)于往connRequests這個(gè)map插入自己的號(hào)碼牌。
 // 插入號(hào)碼牌之后這邊就不需要阻塞等待繼續(xù)往下走邏輯。
 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取消操作的時(shí)候，記得從connRequests這個(gè)map取走自己的號(hào)碼牌。
 db.mu.Lock()
 delete(db.connRequests, reqKey)
 db.mu.Unlock()

 atomic.AddInt64(db.waitDuration, int64(time.Since(waitStart)))

 select {
 default:
 case ret, ok := -req:
    // 這邊值得注意了，因?yàn)楝F(xiàn)在已經(jīng)被context取消了。但是剛剛放了自己的號(hào)碼牌進(jìn)去排隊(duì)里面。意思是說(shuō)不定已經(jīng)發(fā)了連接了，所以得注意歸還！
 if ok  ret.conn != nil {
  db.putConn(ret.conn, ret.err, false)
 }
 }
 return nil, ctx.Err()
 case ret, ok := -req:
   // 下面是已經(jīng)獲得連接后的操作了。檢測(cè)一下獲得連接的狀況。因?yàn)橛锌赡芤呀?jīng)過(guò)期了等等。
 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
 }
 }
 // 下面就是如果上面說(shuō)的限制情況不存在，可以創(chuàng)建先連接時(shí)候，要做的創(chuàng)建連接操作了。
 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
}

...
 // 如果已經(jīng)超過(guò)最大打開(kāi)數(shù)量了，就不需要在回歸pool了
 if db.maxOpen > 0  db.numOpen > db.maxOpen {
 return false
 }
 // 這邊是重點(diǎn)了，基本來(lái)說(shuō)就是從connRequest這個(gè)map里面隨機(jī)抽一個(gè)在排隊(duì)等著的請(qǐng)求。取出來(lái)后發(fā)給他。就不用歸還池子了。
 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) // 刪除這個(gè)在排隊(duì)的請(qǐng)求。
 if err == nil {
 dc.inUse = true
 }
  // 把連接給這個(gè)正在排隊(duì)的連接。
 req - connRequest{
 conn: dc,
 err: err,
 }
 return true
 } else if err == nil  !db.closed {
  // 既然沒(méi)人排隊(duì)，就看看到了最大連接數(shù)目沒(méi)有。沒(méi)到就歸還給freeConn。
 if db.maxIdleConnsLocked() > len(db.freeConn) {
 db.freeConn = append(db.freeConn, dc)
 db.startCleanerLocked()
 return true
 }
 db.maxIdleClosed++
 }
...

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{}
}

func NewConnPool(opt *Options) *ConnPool {
 ....
 p.checkMinIdleConns()

 if opt.IdleTimeout > 0  opt.IdleCheckFrequency > 0 {
 go p.reaper(opt.IdleCheckFrequency)
 }
 ....
}

func (p *ConnPool) Get(ctx context.Context) (*Conn, error) {
 if p.closed() {
 return nil, ErrClosed
 }
 
 //這邊和前面sql獲取連接函數(shù)的流程先不同。sql是先看看連接池有沒(méi)有空閑連接，有的話先獲取不到再排隊(duì)。這邊是直接先排隊(duì)獲取令牌，排隊(duì)函數(shù)后面會(huì)分析。
 err := p.waitTurn(ctx)
 if err != nil {
 return nil, err
 }
 //前面沒(méi)出error的話，就已經(jīng)排隊(duì)輪候到了。接下來(lái)就是獲取的流程。
 for {
 p.connsMu.Lock()
  //從空閑連接里面先獲取一個(gè)空閑連接。
 cn := p.popIdle()
 p.connsMu.Unlock()

 if cn == nil {
   // 沒(méi)有空閑連接時(shí)候直接跳出循環(huán)。
 break
 }
 // 判斷是否已經(jīng)過(guò)時(shí)，是的話close掉了然后繼續(xù)取出。
 if p.isStaleConn(cn) {
 _ = p.CloseConn(cn)
 continue
 }

 atomic.AddUint32(p.stats.Hits, 1)
 return cn, nil
 }

 atomic.AddUint32(p.stats.Misses, 1)
 
 // 如果沒(méi)有空閑連接的話，這邊就直接新建連接了。
 newcn, err := p.newConn(ctx, true)
 if err != nil {
  // 歸還令牌。
 p.freeTurn()
 return nil, err
 }

 return newcn, nil
}

func (p *ConnPool) freeTurn() {
 -p.queue
}
func (p *ConnPool) waitTurn(ctx context.Context) error {
...
 case p.queue - struct{}{}:
 return nil
...
}

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 {
 // 如果連接池滿了，會(huì)在后面移除。
 if p.poolSize >= p.opt.PoolSize {
 cn.pooled = false
 } else {
 p.poolSize++
 }
 }
 p.connsMu.Unlock()
 return cn, nil
}

func (p *ConnPool) Put(cn *Conn) {
 if cn.rd.Buffered() > 0 {
 internal.Logger.Printf("Conn has unread data")
 p.Remove(cn, BadConnError{})
 return
 }
 //這就是我們剛剛說(shuō)的后面了，前面標(biāo)記過(guò)不要入池的，這邊就刪除了。當(dāng)然了，里面也會(huì)進(jìn)行freeTurn操作。
 if !cn.pooled {
  // 這個(gè)方法就是前面的標(biāo)志位，判斷里面可以知道，前面標(biāo)志不要池化的，這里會(huì)將它刪除。
 p.Remove(cn, nil)
 return
 }

 p.connsMu.Lock()
 p.idleConns = append(p.idleConns, cn)
 p.idleConnsLen++
 p.connsMu.Unlock()
 //我們可以看到很明顯的這個(gè)歸還號(hào)碼牌的動(dòng)作。
 p.freeTurn()
}