前言
golang的”database/sql”是操作數(shù)據(jù)庫時常用的包,這個包定義了一些sql操作的接口,具體的實現(xiàn)還需要不同數(shù)據(jù)庫的實現(xiàn),mysql比較優(yōu)秀的一個驅(qū)動是:github.com/go-sql-driver/mysql,在接口、驅(qū)動的設(shè)計上”database/sql”的實現(xiàn)非常優(yōu)秀,對于類似設(shè)計有很多值得我們借鑒的地方,比如beego框架cache的實現(xiàn)模式就是借鑒了這個包的實現(xiàn);”database/sql”除了定義接口外還有一個重要的功能:連接池,我們在實現(xiàn)其他網(wǎng)絡(luò)通信時也可以借鑒其實現(xiàn)。
連接池的作用這里就不再多說了,我們先從一個簡單的示例看下”database/sql”怎么用:
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()
}
用法很簡單,首先Open打開一個數(shù)據(jù)庫,然后調(diào)用Query、Exec執(zhí)行數(shù)據(jù)庫操作,github.com/go-sql-driver/mysql具體實現(xiàn)了database/sql/driver的接口,所以最終具體的數(shù)據(jù)庫操作都是調(diào)用github.com/go-sql-driver/mysql實現(xiàn)的方法,同一個數(shù)據(jù)庫只需要調(diào)用一次Open即可,下面根據(jù)具體的操作分析下”database/sql”都干了哪些事。
1.驅(qū)動注冊
import _ "github.com/go-sql-driver/mysql"前面的”_”作用時不需要把該包都導(dǎo)進來,只執(zhí)行包的init()方法,mysql驅(qū)動正是通過這種方式注冊到”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()通過Register()方法將mysql驅(qū)動添加到sql.drivers(類型:make(map[string]driver.Driver))中,MySQLDriver實現(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ù)據(jù)庫,就可以通過調(diào)用sql.Register將不同數(shù)據(jù)庫的實現(xiàn)注冊到sql.drivers中去,用的時候再根據(jù)注冊的name將對應(yīng)的driver取出。
2.連接池實現(xiàn)
先看下連接池整體處理流程:
2.1 初始化DB
db, err := sql.Open("mysql", "username:password@tcp(host)/db_name?charset=utf8allowOldPasswords=1")
sql.Open()是取出對應(yīng)的db,這時mysql還沒有建立連接,只是初始化了一個sql.DB結(jié)構(gòu),這是非常重要的一個結(jié)構(gòu),所有相關(guān)的數(shù)據(jù)都保存在此結(jié)構(gòu)中;Open同時啟動了一個connectionOpener協(xié)程,后面再具體分析其作用。
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ù)達到最大限制時,后續(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{}
}
maxIdle(默認(rèn)值2)、maxOpen(默認(rèn)值0,無限制)、maxLifetime(默認(rèn)值0,永不過期)可以分別通過SetMaxIdleConns、SetMaxOpenConns、SetConnMaxLifetime設(shè)定。
2.2 獲取連接
上面說了Open時是沒有建立數(shù)據(jù)庫連接的,只有等用的時候才會實際建立連接,獲取可用連接的操作有兩種策略:cachedOrNewConn(有可用空閑連接則優(yōu)先使用,沒有則創(chuàng)建)、alwaysNewConn(不管有沒有空閑連接都重新創(chuàng)建),下面以一個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)獲取到了可用連接,下面進行具體的數(shù)據(jù)庫操作
return db.queryConn(ci, ci.releaseConn, query, args)
}
數(shù)據(jù)庫連接由db.query()獲取:
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)達到最大限制則將請求加入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
}
總結(jié)一下上面獲取連接的過程:
* step1:首先檢查下freeConn里是否有空閑連接,如果有且未超時則直接復(fù)用,返回連接,如果沒有或連接已經(jīng)過期則進入下一步;
* step2:檢查當(dāng)前已經(jīng)建立及準(zhǔn)備建立的連接數(shù)是否已經(jīng)達到最大值,如果達到最大值也就意味著無法再創(chuàng)建新的連接了,當(dāng)前請求需要在這等著連接釋放,這時當(dāng)前協(xié)程將創(chuàng)建一個channel:chan connRequest,并將其插入db.connRequests隊列,然后阻塞在接收chan connRequest上,等到有連接可用時這里將拿到釋放的連接,檢查可用后返回;如果還未達到最大值則進入下一步;
* step3:創(chuàng)建一個連接,首先將numOpen加1,然后再創(chuàng)建連接,如果等到創(chuàng)建完連接再把numOpen加1會導(dǎo)致多個協(xié)程同時創(chuàng)建連接時一部分會浪費,所以提前將numOpen占住,創(chuàng)建失敗再將其減掉;如果創(chuàng)建連接成功則返回連接,失敗則進入下一步
* step4:創(chuàng)建連接失敗時有一個善后操作,當(dāng)然并不僅僅是將最初占用的numOpen數(shù)減掉,更重要的一個操作是通知connectionOpener協(xié)程根據(jù)db.connRequests等待的長度創(chuàng)建連接,這個操作的原因是:
numOpen在連接成功創(chuàng)建前就加了1,這時候如果numOpen已經(jīng)達到最大值再有獲取conn的請求將阻塞在step2,這些請求會等著先前進來的請求釋放連接,假設(shè)先前進來的這些請求創(chuàng)建連接全部失敗,那么如果它們直接返回了那些等待的請求將一直阻塞在那,因為不可能有連接釋放(極限值,如果部分創(chuàng)建成功則會有部分釋放),直到新請求進來重新成功創(chuàng)建連接,顯然這樣是有問題的,所以maybeOpenNewConnections將通知connectionOpener根據(jù)db.connRequests長度及可創(chuàng)建的最大連接數(shù)重新創(chuàng)建連接,然后將新創(chuàng)建的連接發(fā)給阻塞的請求。
注意:如果maxOpen=0將不會有請求阻塞等待連接,所有請求只要從freeConn中取不到連接就會新創(chuàng)建。
另外Query、Exec有個重試機制,首先優(yōu)先使用空閑連接,如果2次取到的連接都無效則嘗試新創(chuàng)建連接。
獲取到可用連接后將調(diào)用具體數(shù)據(jù)庫的driver處理sql。
2.3 釋放連接
數(shù)據(jù)庫連接在被使用完成后需要歸還給連接池以供其它請求復(fù)用,釋放連接的操作是:putConn():
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
}
釋放的過程:
* step1:首先檢查下當(dāng)前歸還的連接在使用過程中是否發(fā)現(xiàn)已經(jīng)無效,如果無效則不再放入連接池,然后檢查下等待連接的請求數(shù)新建連接,類似獲取連接時的異常處理,如果連接有效則進入下一步;
* step2:檢查下當(dāng)前是否有等待連接阻塞的請求,有的話將當(dāng)前連接發(fā)給最早的那個請求,沒有的話則再判斷空閑連接數(shù)是否達到上限,沒有則放入freeConn空閑連接池,達到上限則將連接關(guān)閉釋放。
* step3:(只執(zhí)行一次)啟動connectionCleaner協(xié)程定時檢查feeConn中是否有過期連接,有則剔除。
有個地方需要注意的是,Query、Exec操作用法有些差異:
a.Exec(update、insert、delete等無結(jié)果集返回的操作)調(diào)用完后會自動釋放連接;
b.Query(返回sql.Rows)則不會釋放連接,調(diào)用完后仍然占有連接,它將連接的所屬權(quán)轉(zhuǎn)移給了sql.Rows,所以需要手動調(diào)用close歸還連接,即使不用Rows也得調(diào)用rows.Close(),否則可能導(dǎo)致后續(xù)使用出錯,如下的用法是錯誤的:
//錯誤
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()
附:請求一個連接的函數(shù)有好幾種,執(zhí)行完畢處理連接的方式稍有差別,大致如下:
- db.Ping() 調(diào)用完畢后會馬上把連接返回給連接池。
- db.Exec() 調(diào)用完畢后會馬上把連接返回給連接池,但是它返回的Result對象還保留這連接的引用,當(dāng)后面的代碼需要處理結(jié)果集的時候連接將會被重用。
- db.Query() 調(diào)用完畢后會將連接傳遞給sql.Rows類型,當(dāng)然后者迭代完畢或者顯示的調(diào)用.Clonse()方法后,連接將會被釋放回到連接池。
- db.QueryRow()調(diào)用完畢后會將連接傳遞給sql.Row類型,當(dāng).Scan()方法調(diào)用之后把連接釋放回到連接池。
- db.Begin() 調(diào)用完畢后將連接傳遞給sql.Tx類型對象,當(dāng).Commit()或.Rollback()方法調(diào)用后釋放連接。
總結(jié)
以上就是這篇文章的全部內(nèi)容了,希望本文的內(nèi)容對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價值,如果有疑問大家可以留言交流,謝謝大家對腳本之家的支持。
您可能感興趣的文章:- Golang連接池的幾種實現(xiàn)案例小結(jié)
- golang實現(xiàn)基于channel的通用連接池詳解
- Golang你一定要懂的連接池實現(xiàn)