github.com/go-sql-driver/mysql连接池的作用这里就不再多说了，我们先从一个简单的示例看下”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=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()
}github.com/go-sql-driver/mysqldatabase/sql/drivergithub.com/go-sql-driver/mysql1.驱动注册
import _ "github.com/go-sql-driver/mysql"init()//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()sql.driversdriver.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)
}sql.Registersql.drivers2.连接池实现
先看下连接池整体处理流程：
2.1 初始化DB
db, err := sql.Open("mysql", "username:password@tcp(host)/db_name?charset=utf8&allowOldPasswords=1")sql.Open()sql.DBconnectionOpenertype DB struct {
driver driver.Driver //数据库实现驱动
dsn string //数据库连接、配置参数信息，比如username、host、password等
numClosed uint64
mu sync.Mutex //锁，操作DB各成员时用到
freeConn []*driverConn //空闲连接
connRequests []chan connRequest //阻塞请求队列，等连接数达到最大限制时，后续请求将插入此队列等待可用连接
numOpen int //已建立连接或等待建立连接数
openerCh chan struct{} //用于connectionOpener
closed bool
dep map[finalCloser]depSet
lastPut map[*driverConn]string // stacktrace of last conn's put; debug only
maxIdle int //最大空闲连接数
maxOpen int //数据库最大连接数
maxLifetime time.Duration //连接最长存活期，超过这个时间连接将不再被复用
cleanerCh chan struct{}
}maxIdlemaxOpenmaxLifetime(默认值0，永不过期)SetMaxIdleConnsSetMaxOpenConnsSetConnMaxLifetime2.2 获取连接
Openrows, err := db.Query("select * from test")database/sql/sql.gofunc (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
}
//到这已经获取到了可用连接，下面进行具体的数据库操作
return db.queryConn(ci, ci.releaseConn, query, args)
}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
}
//如果没有空闲连接，而且当前建立的连接数已经达到最大限制则将请求加入connRequests队列，
//并阻塞在这里，直到其它协程将占用的连接释放或connectionOpenner创建
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是已经建立或即将建立连接数，这里还没有建立连接，只是乐观的认为后面会成功，失败的时候再将此值减1
db.mu.Unlock()
ci, err := db.driver.Open(db.dsn) //调用driver的Open方法建立连接
if err != nil { //创建连接失败
db.mu.Lock()
db.numOpen-- // correct for earlier optimism
db.maybeOpenNewConnections() //通知connectionOpener协程尝试重新建立连接，否则在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
}chan connRequestdb.connRequestschan connRequestdb.connRequestsmaybeOpenNewConnectionsdb.connRequestsmaxOpen=0QueryExec获取到可用连接后将调用具体数据库的driver处理sql。
2.3 释放连接
putConn()func (db *DB) putConn(dc *driverConn, err error) {
...
//如果连接已经无效，则不再放入连接池
if err == driver.ErrBadConn {
db.maybeOpenNewConnections()
dc.Close() //这里最终将numOpen数减掉
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
}
//有等待连接的请求则将连接发给它们，否则放入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：首先检查下当前归还的连接在使用过程中是否发现已经无效，如果无效则不再放入连接池，然后检查下等待连接的请求数新建连接，类似获取连接时的异常处理，如果连接有效则进入下一步；
* step2：检查下当前是否有等待连接阻塞的请求，有的话将当前连接发给最早的那个请求，没有的话则再判断空闲连接数是否达到上限，没有则放入freeConn空闲连接池，达到上限则将连接关闭释放。
* step3：(只执行一次)启动connectionCleaner协程定时检查feeConn中是否有过期连接，有则剔除。
QueryExecExecQuerysql.Rows//错误
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() //将连接的所属权归还，释放连接
row,err := db.Query("select * from test")
//other op
row.Close()