golang sql連接池的實現(xiàn)方法詳解

package main

import(
 "fmt"
 "database/sql"
 _ "github.com/go-sql-driver/mysql"
)

func main(){

 db, err := sql.Open("mysql", "username:password@tcp(host)/db_name?charset=utf8allowOldPasswords=1")
 if err != nil {
  fmt.Println(err)
  return
 }
 defer db.Close()

 rows,err := db.Query("select * from test")

 for rows.Next(){
  //row.Scan(...)
 }
 rows.Close()
}

//github.com/go-sql-driver/mysql/driver.go
func init() {
 sql.Register("mysql", MySQLDriver{})
}

type MySQLDriver struct{}

func (d MySQLDriver) Open(dsn string) (driver.Conn, error) {
 ...
}

//database/sql/sql.go
func Register(name string, driver driver.Driver) {
 driversMu.Lock()
 defer driversMu.Unlock()
 if driver == nil {
  panic("sql: Register driver is nil")
 }
 if _, dup := drivers[name]; dup {
  panic("sql: Register called twice for driver " + name)
 }
 drivers[name] = driver
}

//database/sql/driver/driver.go
type Driver interface {
 // Open returns a new connection to the database.
 // The name is a string in a driver-specific format.
 //
 // Open may return a cached connection (one previously
 // closed), but doing so is unnecessary; the sql package
 // maintains a pool of idle connections for efficient re-use.
 //
 // The returned connection is only used by one goroutine at a
 // time.
 Open(name string) (Conn, error)
}

db, err := sql.Open("mysql", "username:password@tcp(host)/db_name?charset=utf8allowOldPasswords=1")

type DB struct {
 driver driver.Driver //數(shù)據(jù)庫實現(xiàn)驅(qū)動
 dsn string //數(shù)據(jù)庫連接、配置參數(shù)信息，比如username、host、password等
 numClosed uint64

 mu   sync.Mutex   //鎖，操作DB各成員時用到
 freeConn  []*driverConn  //空閑連接
 connRequests []chan connRequest //阻塞請求隊列，等連接數(shù)達(dá)到最大限制時，后續(xù)請求將插入此隊列等待可用連接
 numOpen  int     //已建立連接或等待建立連接數(shù)
 openerCh chan struct{}  //用于connectionOpener
 closed  bool
 dep   map[finalCloser]depSet
 lastPut  map[*driverConn]string // stacktrace of last conn's put; debug only
 maxIdle  int     //最大空閑連接數(shù)
 maxOpen  int     //數(shù)據(jù)庫最大連接數(shù)
 maxLifetime time.Duration   //連接最長存活期，超過這個時間連接將不再被復(fù)用
 cleanerCh chan struct{}
}

rows, err := db.Query("select * from test")

func (db *DB) Query(query string, args ...interface{}) (*Rows, error) {
 var rows *Rows
 var err error
 //maxBadConnRetries = 2
 for i := 0; i  maxBadConnRetries; i++ {
  rows, err = db.query(query, args, cachedOrNewConn)
  if err != driver.ErrBadConn {
   break
  }
 }
 if err == driver.ErrBadConn {
  return db.query(query, args, alwaysNewConn)
 }
 return rows, err
}

func (db *DB) query(query string, args []interface{}, strategy connReuseStrategy) (*Rows, error) {
 ci, err := db.conn(strategy)
 if err != nil {
  return nil, err
 }

 //到這已經(jīng)獲取到了可用連接，下面進(jìn)行具體的數(shù)據(jù)庫操作
 return db.queryConn(ci, ci.releaseConn, query, args)
}

func (db *DB) conn(strategy connReuseStrategy) (*driverConn, error) {
 db.mu.Lock()
 if db.closed {
  db.mu.Unlock()
  return nil, errDBClosed
 }
 lifetime := db.maxLifetime

 //從freeConn取一個空閑連接
 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
  }
  return conn, nil
 }

 //如果沒有空閑連接，而且當(dāng)前建立的連接數(shù)已經(jīng)達(dá)到最大限制則將請求加入connRequests隊列，
 //并阻塞在這里，直到其它協(xié)程將占用的連接釋放或connectionOpenner創(chuàng)建
 if db.maxOpen > 0  db.numOpen >= db.maxOpen {
  // Make the connRequest channel. It's buffered so that the
  // connectionOpener doesn't block while waiting for the req to be read.
  req := make(chan connRequest, 1)
  db.connRequests = append(db.connRequests, req)
  db.mu.Unlock()
  ret, ok := -req //阻塞
  if !ok {
   return nil, errDBClosed
  }
  if ret.err == nil  ret.conn.expired(lifetime) { //連接過期了
   ret.conn.Close()
   return nil, driver.ErrBadConn
  }
  return ret.conn, ret.err
 }

 db.numOpen++ //上面說了numOpen是已經(jīng)建立或即將建立連接數(shù)，這里還沒有建立連接，只是樂觀的認(rèn)為后面會成功，失敗的時候再將此值減1
 db.mu.Unlock()
 ci, err := db.driver.Open(db.dsn) //調(diào)用driver的Open方法建立連接
 if err != nil { //創(chuàng)建連接失敗
  db.mu.Lock()
  db.numOpen-- // correct for earlier optimism
  db.maybeOpenNewConnections() //通知connectionOpener協(xié)程嘗試重新建立連接，否則在db.connRequests中等待的請求將一直阻塞，知道下次有連接建立
  db.mu.Unlock()
  return nil, err
 }
 db.mu.Lock()
 dc := driverConn{
  db:  db,
  createdAt: nowFunc(),
  ci:  ci,
 }
 db.addDepLocked(dc, dc)
 dc.inUse = true
 db.mu.Unlock()
 return dc, nil
}

func (db *DB) putConn(dc *driverConn, err error) {
 ...

 //如果連接已經(jīng)無效，則不再放入連接池
 if err == driver.ErrBadConn {
  db.maybeOpenNewConnections()
  dc.Close() //這里最終將numOpen數(shù)減掉
  return
 }
 ...

 //正常歸還
 added := db.putConnDBLocked(dc, nil)
 ...
}

func (db *DB) putConnDBLocked(dc *driverConn, err error) bool {
 if db.maxOpen > 0  db.numOpen > db.maxOpen {
  return false
 }
 //有等待連接的請求則將連接發(fā)給它們，否則放入freeConn
 if c := len(db.connRequests); c > 0 {
  req := db.connRequests[0]
  // This copy is O(n) but in practice faster than a linked list.
  // TODO: consider compacting it down less often and
  // moving the base instead?
  copy(db.connRequests, db.connRequests[1:])
  db.connRequests = db.connRequests[:c-1]
  if err == nil {
   dc.inUse = true
  }
  req - connRequest{
   conn: dc,
   err: err,
  }
  return true
 } else if err == nil  !db.closed  db.maxIdleConnsLocked() > len(db.freeConn) {
  db.freeConn = append(db.freeConn, dc)
  db.startCleanerLocked()
  return true
 }
 return false
}

//錯誤
db.SetMaxOpenConns(1)
db.Query("select * from test")

row,err := db.Query("select * from test") //此操作將一直阻塞

//正確
db.SetMaxOpenConns(1)
r,_ := db.Query("select * from test")
r.Close() //將連接的所屬權(quán)歸還，釋放連接
row,err := db.Query("select * from test")
//other op
row.Close()