使用golang怎么實(shí)現(xiàn)一個(gè)sql連接池

使用golang怎么實(shí)現(xiàn)一個(gè)sql連接池？針對(duì)這個(gè)問(wèn)題，這篇文章詳細(xì)介紹了相對(duì)應(yīng)的分析和解答，希望可以幫助更多想解決這個(gè)問(wèn)題的小伙伴找到更簡(jiǎ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=utf8&allowOldPasswords=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()
}

用法很簡(jiǎn)單，首先Open打開(kāi)一個(gè)數(shù)據(jù)庫(kù)，然后調(diào)用Query、Exec執(zhí)行數(shù)據(jù)庫(kù)操作，github.com/go-sql-driver/mysql具體實(shí)現(xiàn)了database/sql/driver的接口，所以最終具體的數(shù)據(jù)庫(kù)操作都是調(diào)用github.com/go-sql-driver/mysql實(shí)現(xiàn)的方法，同一個(gè)數(shù)據(jù)庫(kù)只需要調(diào)用一次Open即可，下面根據(jù)具體的操作分析下”database/sql”都干了哪些事。

1.驅(qū)動(dòng)注冊(cè)

import _ "github.com/go-sql-driver/mysql"前面的”_”作用時(shí)不需要把該包都導(dǎo)進(jìn)來(lái)，只執(zhí)行包的init()方法，mysql驅(qū)動(dòng)正是通過(guò)這種方式注冊(cè)到”database/sql”中的：

//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) {
 ...
}

init()通過(guò)Register()方法將mysql驅(qū)動(dòng)添加到sql.drivers(類型：make(map[string]driver.Driver))中,MySQLDriver實(shí)現(xiàn)了driver.Driver接口：

//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)
}

假如我們同時(shí)用到多種數(shù)據(jù)庫(kù)，就可以通過(guò)調(diào)用sql.Register將不同數(shù)據(jù)庫(kù)的實(shí)現(xiàn)注冊(cè)到sql.drivers中去，用的時(shí)候再根據(jù)注冊(cè)的name將對(duì)應(yīng)的driver取出。

2.連接池實(shí)現(xiàn)

先看下連接池整體處理流程：

2.1 初始化DB

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

sql.Open()是取出對(duì)應(yīng)的db，這時(shí)mysql還沒(méi)有建立連接，只是初始化了一個(gè)sql.DB結(jié)構(gòu)，這是非常重要的一個(gè)結(jié)構(gòu)，所有相關(guān)的數(shù)據(jù)都保存在此結(jié)構(gòu)中；Open同時(shí)啟動(dòng)了一個(gè)connectionOpener協(xié)程，后面再具體分析其作用。

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

 mu   sync.Mutex   //鎖，操作DB各成員時(shí)用到
 freeConn  []*driverConn  //空閑連接
 connRequests []chan connRequest //阻塞請(qǐng)求隊(duì)列，等連接數(shù)達(dá)到最大限制時(shí)，后續(xù)請(qǐng)求將插入此隊(duì)列等待可用連接
 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ù)庫(kù)最大連接數(shù)
 maxLifetime time.Duration   //連接最長(zhǎng)存活期，超過(guò)這個(gè)時(shí)間連接將不再被復(fù)用
 cleanerCh chan struct{}
}

maxIdle(默認(rèn)值2)、maxOpen(默認(rèn)值0，無(wú)限制)、maxLifetime(默認(rèn)值0，永不過(guò)期)可以分別通過(guò)SetMaxIdleConns、SetMaxOpenConns、SetConnMaxLifetime設(shè)定。

2.2 獲取連接

上面說(shuō)了Open時(shí)是沒(méi)有建立數(shù)據(jù)庫(kù)連接的，只有等用的時(shí)候才會(huì)實(shí)際建立連接，獲取可用連接的操作有兩種策略：cachedOrNewConn(有可用空閑連接則優(yōu)先使用，沒(méi)有則創(chuàng)建)、alwaysNewConn(不管有沒(méi)有空閑連接都重新創(chuàng)建)，下面以一個(gè)query的例子看下具體的操作：

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

database/sql/sql.go：

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ù)庫(kù)操作
 return db.queryConn(ci, ci.releaseConn, query, args)
}

數(shù)據(jù)庫(kù)連接由db.query()獲?。?/p>

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

 //從freeConn取一個(gè)空閑連接
 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
 }

 //如果沒(méi)有空閑連接，而且當(dāng)前建立的連接數(shù)已經(jīng)達(dá)到最大限制則將請(qǐng)求加入connRequests隊(duì)列，
 //并阻塞在這里，直到其它協(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) { //連接過(guò)期了
   ret.conn.Close()
   return nil, driver.ErrBadConn
  }
  return ret.conn, ret.err
 }

 db.numOpen++ //上面說(shuō)了numOpen是已經(jīng)建立或即將建立連接數(shù)，這里還沒(méi)有建立連接，只是樂(lè)觀的認(rèn)為后面會(huì)成功，失敗的時(shí)候再將此值減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中等待的請(qǐng)求將一直阻塞，知道下次有連接建立
  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
}

總結(jié)一下上面獲取連接的過(guò)程：

* step1：首先檢查下freeConn里是否有空閑連接，如果有且未超時(shí)則直接復(fù)用，返回連接，如果沒(méi)有或連接已經(jīng)過(guò)期則進(jìn)入下一步；

* step2：檢查當(dāng)前已經(jīng)建立及準(zhǔn)備建立的連接數(shù)是否已經(jīng)達(dá)到最大值，如果達(dá)到最大值也就意味著無(wú)法再創(chuàng)建新的連接了，當(dāng)前請(qǐng)求需要在這等著連接釋放，這時(shí)當(dāng)前協(xié)程將創(chuàng)建一個(gè)channel：chan connRequest，并將其插入db.connRequests隊(duì)列，然后阻塞在接收chan connRequest上，等到有連接可用時(shí)這里將拿到釋放的連接，檢查可用后返回；如果還未達(dá)到最大值則進(jìn)入下一步；

* step3：創(chuàng)建一個(gè)連接，首先將numOpen加1，然后再創(chuàng)建連接，如果等到創(chuàng)建完連接再把numOpen加1會(huì)導(dǎo)致多個(gè)協(xié)程同時(shí)創(chuàng)建連接時(shí)一部分會(huì)浪費(fèi)，所以提前將numOpen占住，創(chuàng)建失敗再將其減掉；如果創(chuàng)建連接成功則返回連接，失敗則進(jìn)入下一步

* step4：創(chuàng)建連接失敗時(shí)有一個(gè)善后操作，當(dāng)然并不僅僅是將最初占用的numOpen數(shù)減掉，更重要的一個(gè)操作是通知connectionOpener協(xié)程根據(jù)db.connRequests等待的長(zhǎng)度創(chuàng)建連接，這個(gè)操作的原因是：

numOpen在連接成功創(chuàng)建前就加了1，這時(shí)候如果numOpen已經(jīng)達(dá)到最大值再有獲取conn的請(qǐng)求將阻塞在step2，這些請(qǐng)求會(huì)等著先前進(jìn)來(lái)的請(qǐng)求釋放連接，假設(shè)先前進(jìn)來(lái)的這些請(qǐng)求創(chuàng)建連接全部失敗，那么如果它們直接返回了那些等待的請(qǐng)求將一直阻塞在那，因?yàn)椴豢赡苡羞B接釋放(極限值，如果部分創(chuàng)建成功則會(huì)有部分釋放)，直到新請(qǐng)求進(jìn)來(lái)重新成功創(chuàng)建連接，顯然這樣是有問(wèn)題的，所以maybeOpenNewConnections將通知connectionOpener根據(jù)db.connRequests長(zhǎng)度及可創(chuàng)建的最大連接數(shù)重新創(chuàng)建連接，然后將新創(chuàng)建的連接發(fā)給阻塞的請(qǐng)求。

注意：如果maxOpen=0將不會(huì)有請(qǐng)求阻塞等待連接，所有請(qǐng)求只要從freeConn中取不到連接就會(huì)新創(chuàng)建。

另外Query、Exec有個(gè)重試機(jī)制，首先優(yōu)先使用空閑連接，如果2次取到的連接都無(wú)效則嘗試新創(chuàng)建連接。

獲取到可用連接后將調(diào)用具體數(shù)據(jù)庫(kù)的driver處理sql。

2.3 釋放連接

數(shù)據(jù)庫(kù)連接在被使用完成后需要?dú)w還給連接池以供其它請(qǐng)求復(fù)用，釋放連接的操作是：putConn()：

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

 //如果連接已經(jīng)無(wú)效，則不再放入連接池
 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
 }
 //有等待連接的請(qǐng)求則將連接發(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
}

釋放的過(guò)程：

* step1：首先檢查下當(dāng)前歸還的連接在使用過(guò)程中是否發(fā)現(xiàn)已經(jīng)無(wú)效，如果無(wú)效則不再放入連接池，然后檢查下等待連接的請(qǐng)求數(shù)新建連接，類似獲取連接時(shí)的異常處理，如果連接有效則進(jìn)入下一步；

* step2：檢查下當(dāng)前是否有等待連接阻塞的請(qǐng)求，有的話將當(dāng)前連接發(fā)給最早的那個(gè)請(qǐng)求，沒(méi)有的話則再判斷空閑連接數(shù)是否達(dá)到上限，沒(méi)有則放入freeConn空閑連接池，達(dá)到上限則將連接關(guān)閉釋放。

* step3：(只執(zhí)行一次)啟動(dòng)connectionCleaner協(xié)程定時(shí)檢查feeConn中是否有過(guò)期連接，有則剔除。

有個(gè)地方需要注意的是，Query、Exec操作用法有些差異：

a.Exec(update、insert、delete等無(wú)結(jié)果集返回的操作)調(diào)用完后會(huì)自動(dòng)釋放連接；

b.Query(返回sql.Rows)則不會(huì)釋放連接，調(diào)用完后仍然占有連接，它將連接的所屬權(quán)轉(zhuǎn)移給了sql.Rows，所以需要手動(dòng)調(diào)用close歸還連接，即使不用Rows也得調(diào)用rows.Close()，否則可能導(dǎo)致后續(xù)使用出錯(cuò)，如下的用法是錯(cuò)誤的：

//錯(cuò)誤
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()

附：請(qǐng)求一個(gè)連接的函數(shù)有好幾種，執(zhí)行完畢處理連接的方式稍有差別，大致如下：

• db.Ping() 調(diào)用完畢后會(huì)馬上把連接返回給連接池。

• db.Exec() 調(diào)用完畢后會(huì)馬上把連接返回給連接池，但是它返回的Result對(duì)象還保留這連接的引用，當(dāng)后面的代碼需要處理結(jié)果集的時(shí)候連接將會(huì)被重用。

• db.Query() 調(diào)用完畢后會(huì)將連接傳遞給sql.Rows類型，當(dāng)然后者迭代完畢或者顯示的調(diào)用.Clonse()方法后，連接將會(huì)被釋放回到連接池。

• db.QueryRow()調(diào)用完畢后會(huì)將連接傳遞給sql.Row類型，當(dāng).Scan()方法調(diào)用之后把連接釋放回到連接池。

• db.Begin() 調(diào)用完畢后將連接傳遞給sql.Tx類型對(duì)象，當(dāng).Commit()或.Rollback()方法調(diào)用后釋放連接。

golang適合做什么

golang可以做服務(wù)器端開(kāi)發(fā)，但golang很適合做日志處理、數(shù)據(jù)打包、虛擬機(jī)處理、數(shù)據(jù)庫(kù)代理等工作。在網(wǎng)絡(luò)編程方面，它還廣泛應(yīng)用于web應(yīng)用、API應(yīng)用等領(lǐng)域。

關(guān)于使用golang怎么實(shí)現(xiàn)一個(gè)sql連接池問(wèn)題的解答就分享到這里了，希望以上內(nèi)容可以對(duì)大家有一定的幫助，如果你還有很多疑惑沒(méi)有解開(kāi)，可以關(guān)注億速云行業(yè)資訊頻道了解更多相關(guān)知識(shí)。