一、如何理解数据库连接

数据库连接池是由客户端维护的存放数据库连接的池子,连接被维护在池子里面,谁用谁来取,目的是降低频繁的创建和关闭连接的开销。

关于如何理解数据库连接,大家可以借助这个TCP编程的Demo来理解。

为了便于理解,可以MySQL-Server的连接池想象成就是这个简单的Tcp-Server

func main() {
// 1. 监听端口 2.accept连接 3.开goroutine处理连接
listen, err := net.Listen("tcp", "0.0.0.0:9090")
if err != nil {
fmt.Printf("error : %v", err)
return
}
for{
conn, err := listen.Accept()
if err != nil {
fmt.Printf("Fail listen.Accept : %v", err)
continue
}
go ProcessConn(conn)
}
} // 处理网络请求
func ProcessConn(conn net.Conn) {
// defer conn.Close()
for {
bt,err:= coder.Decode(conn)
if err != nil {
fmt.Printf("Fail to decode error [%v]", err)
return
}
s := string(bt)
fmt.Printf("Read from conn:[%v]\n",s)
}
}

对于我们现在看的sql包下的连接池,可以简化认为它就是如下的tcp-client

conn, err := net.Dial("tcp", ":9090")
defer conn.Close()
if err != nil {
fmt.Printf("error : %v", err)
return
} // 将数据编码并发送出去
coder.Encode(conn,"hi server i am here"); time.Sleep(time.Second*10

总体的思路可以认为,程序启动的时候,根据我们的配置,sql包中的DB会为我们提前创建几条这样的conn,然后维护起来,不close()掉,我们想使用的时候问他拿即可。

至于为什么是这个tcp的demo呢?因为数据库连接的建立底层依赖的是tcp连接。基于tcp连接的基础上实现客户端和服务端数据的传输,再往上封装一层mysql的握手、鉴权、交互协议对数据包进行解析、反解析,进而跑通整个流程。

二、连接池的工作原理

  • 连接池的建立

    • 后台系统初始化时,连接池会根据系统的配置建立。
    • 但是在接受客户端请求之前,并没有真正的创建连接。
    • 在go语言中,先注册驱动_ "github.com/go-sql-driver/mysql"
    • 初始化DB,调用Open函数,这时其实没有真正的去获取连接,而是去获取DB操作的数据结构。
  • 连接池中连接的使用和管理
  • 连接池的关闭
    • 释放连接
    • 关闭连接的请求队列
    • connectionOpener(负责打开连接的协程)
    • connectionResetter(重制连接状态的协程)
    • connectionCleaner(定期清理过期连接的协程)

三、database/sql包结构

driver/driver.go :定义了实现数据库驱动所需要的接口,这些接口由sql包和具体的驱动包来实现

driver/types.go:定义了数据类型别名和转换

convert:rows的scan

sql.go: 关于SQL数据库的一些通用的接口、类型。包括:连接池、数据类型、连接、事物、statement

import "github.com/go-sql-driver/mysql” // 具体的驱动包
import "database/sql" // 初始化连接
func initDB() (err error) {
db, err = sql.Open("mysql", "root:root@tcp(127.0.0.1:3306)/test")
if err != nil {
panic(err)
}
// todo 不要在这里关闭它, 函数一结束,defer就执行了
// defer db.Close()
err = db.Ping()
if err != nil {
return err
}
return nil
}

四、三个重要的结构体

4.1、DB

/**
DB是代表零个或多个基础连接池的数据库句柄。 对于多个goroutine并发使用是安全的。
sql包会自动创建并释放连接。 它还维护空闲连接的空闲池。
如果数据库具有每个连接状态的概念,则可以在事务(Tx)或连接(Conn)中可靠地观察到这种状态。
调用DB.Begin之后,返回的Tx将绑定到单个连接。
在事务上调用Commit或Rollback后,该事务的连接将返回到DB的空闲连接池。
池大小可以通过SetMaxIdleConns控制。
*/
type DB struct {
// Atomic access only. At top of struct to prevent mis-alignment
// on 32-bit platforms. Of type time.Duration.
// 统计使用:等待新的连接所需要的总时间
waitDuration int64 // Total time waited for new connections. // 由具体的数据库驱动实现的 connector
connector driver.Connector // numClosed is an atomic counter which represents a total number of
// closed connections. Stmt.openStmt checks it before cleaning closed
// connections in Stmt.css.
// 关闭的连接数
numClosed uint64 mu sync.Mutex // protects following fields // 连接池,在go中,连接的封装结构体是:driverConn
freeConn []*driverConn // 连接请求的map, key是自增的int64类型的数,用于唯一标示这个请求分配的
connRequests map[uint64]chan connRequest // 类似于binlog中的next trx_ix ,下一个事物的id
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.
// 他是个chan,用于通知connectionOpener()协程应该打开新的连接了。
openerCh chan struct{} // 他是个chan,用于通知connectionResetter协程:重制连接的状态。
resetterCh chan *driverConn closed bool // 依赖,key是连接、statement
dep map[finalCloser]depSet
lastPut map[*driverConn]string // stacktrace of last conn's put; debug only // 连接池的大小,0意味着使用默认的大小2, 小于0表示不使用连接池
maxIdle int // zero means defaultMaxIdleConns; negative means 0
// 最大打开的连接数,包含连接池中的连接和连接池之外的空闲连接, 0表示不做限制
maxOpen int // <= 0 means unlimited // 连接被重用的时间,设置为0表示一直可以被重用。
maxLifetime time.Duration // maximum amount of time a connection may be reused // 他是个chan,用于通知connectionCleaner协程去请求过期的连接
// 当有设置最大存活时间时才会生效
cleanerCh chan struct{} // 等待的连接总数,当maxIdle为0时,waitCount也会一直为
// 因为maxIdle为0,每一个请求过来都会打开一条新的连接。
waitCount int64 // Total number of connections waited for. // 释放连接时,因为连接池已满而关闭的连接总数
// 如果maxLifeTime没有被设置,maxIdleClosed为0
maxIdleClosed int64 // Total number of connections closed due to idle. // 因为超过了最大连接时间,而被关闭的连接总数
maxLifetimeClosed int64 // Total number of connections closed due to max free limit. // 当DB被关闭时,关闭connection opener和session resetter这两个协程
stop func() // stop cancels the connection opener and the session resetter.
}

4.2、driverConn

连接的封装结构体:driverConn

// driverConn wraps a driver.Conn with a mutex, to
// be held during all calls into the Conn. (including any calls onto
// interfaces returned via that Conn, such as calls on Tx, Stmt,
// Result, Rows)
/**
driverConn使用互斥锁包装Conn包装
*/
type driverConn struct {
// 持有对整个数据库的抽象结构体
db *DB
createdAt time.Time sync.Mutex // guards following // 对应于具体的连接,eg.mysqlConn
ci driver.Conn // 标记当前连接的状态:当前连接是否已经关闭
closed bool
// 标记当前连接的状态:当前连接是否最终关闭,包装 ci.Close has been called
finalClosed bool // ci.Close has been called // 在这些连接上打开的statement
openStmt map[*driverStmt]bool // connectionResetter返回的结果
lastErr error // lastError captures the result of the session resetter. // guarded by db.mu
// 连接是否被占用了
inUse bool // 在归还连接时需要运行的代码。在noteUnusedDriverStatement中添加
onPut []func() // code (with db.mu held) run when conn is next returned dbmuClosed bool // same as closed, but guarded by db.mu, for removeClosedStmtLocked
}

4.3、Conn

具体的连接: driver包下的Conn如下,是个接口,需要被具体的实现。

// Conn is assumed to be stateful.
type Conn interface {
// Prepare returns a prepared statement, bound to this connection.
Prepare(query string) (Stmt, error) // Close invalidates and potentially stops any current
// prepared statements and transactions, marking this
// connection as no longer in use.
//
// Because the sql package maintains a free pool of
// connections and only calls Close when there's a surplus of
// idle connections, it shouldn't be necessary for drivers to
// do their own connection caching.
Close() error // Begin starts and returns a new transaction.
//
// Deprecated: Drivers should implement ConnBeginTx instead (or additionally).
Begin() (Tx, error)
}

五、流程梳理

5.1、先获取DB实例

在golang中,要想获取连接,一般我们都得通过下面这段代码获取到DB的封装结构体实例。

通过上面的三个结构体可以看出 DB 、driverConn、Conn的关系如下:

![未命名文件 (2)](/Users/dxm/Downloads/未命名文件 (2).png)

所以我们的代码一般长成下面这样,先获取一个DB结构体的实例,DB结果体中有维护连接池、以及和创建连接,关闭连接协程通信的channel,已经各种配置参数。

上图中浅蓝色部分的 freeConn就是空闲连接池,里面的driver包下的Conn interface就是具体的连接。

/**
* MySQL连接相关的逻辑
*/
type Conenctor struct {
BaseInfo BaseInfo
DB *sql.DB
} func (c *Conenctor) Open() {
// 读取配置
c.loadConfig()
dataSource := c.BaseInfo.RootUserName + ":" + c.BaseInfo.RootPassword + "@tcp(" + c.BaseInfo.Addr + ":" + c.BaseInfo.Port + ")/" + c.BaseInfo.DBName
db, Err := sql.Open("mysql", dataSource)
if Err != nil {
common.Error("Fail to opendb dataSource:[%v] Err:[%v]", dataSource, Err.Error())
return
}
db.SetMaxOpenConns(500)
db.SetMaxIdleConns(200)
c.DB = db
Err = db.Ping()
if Err != nil {
fmt.Printf("Fail to Ping DB Err :[%v]", Err.Error())
return
}
}

5.2、流程梳理入口:

比如我们自己写代码时,可能会搞这样一个方法做增删改

// 插入、更新、删除
func (c *Conenctor) Exec(ctx context.Context,
sqlText string,
params ...interface{}) (qr *QueryResults) {
qr = &QueryResults{}
result, err := c.DB.ExecContext(ctx, sqlText, params...)
defer HandleException()
if err != nil {
qr.EffectRow = 0
qr.Err = err
common.Error("Fail to exec qurey sqlText:[%v] params:[%v] err:[%v]", sqlText, params, err)
return
}
qr.EffectRow, _ = result.RowsAffected()
qr.LastInsertId, _ = result.LastInsertId()
return
}

主要是使用DB.ExecContext()执行SQL,获取返回值。

ctx是业务代码传入的上线文,通常是做超时限制使用。

其实这里并不是严格意义上的去执行sql,它其实是通过和MySQL-Server之间建立的连接将sql+params发往MySQL-Server去解析和执行。

进入DB.ExecContext()

主要逻辑如下:exec()方法的主要功能是:获取连接,发送sql和参数。

  • 如果获取一次失败一次,当失败的次数达到sql包预定义的常量maxBadConnRetries的情况下,将会创建新的连接使用
  • 未超过maxBadConnRetries,被打上cachedOrNewConn,优先从空闲池中获取连接
func (db *DB) ExecContext(ctx context.Context, query string, args ...interface{}) (Result, error) {
var res Result
var err error
for i := 0; i < maxBadConnRetries; i++ {
res, err = db.exec(ctx, query, args, cachedOrNewConn)
if err != driver.ErrBadConn {
break
}
}
if err == driver.ErrBadConn {
return db.exec(ctx, query, args, alwaysNewConn)
}
return res, err
}

跟进exec() --> db.conn(ctx, strategy)

func (db *DB) exec(ctx context.Context, query string, args []interface{}, strategy connReuseStrategy) (Result, error) {
// 这个strategy就是上一步我们告诉他是创建新连接,还是优先从缓存池中获取连接。
dc, err := db.conn(ctx, strategy)
..
}

5.3、获取连接

跟进conn()方法

conn方法的返回值是driverConn,也就是我们上面说的数据库连接,作用就是说,跟据传递进来的获取策略,获取数据库连接,如果正常就返回获取到的数据库连接,异常就返回错误err

这张图是conn获取连接的流程图,根据下面这段代码画出来的,注释有写在代码上

![未命名文件 (3)](/Users/dxm/Downloads/未命名文件 (4).png)

// conn returns a newly-opened or cached *driverConn.
func (db *DB) conn(ctx context.Context, strategy connReuseStrategy) (*driverConn, error) {
db.mu.Lock()
// 先监测db是否关闭了
if db.closed {
db.mu.Unlock()
// DB都关闭了,直接返回DBClosed错误,没必要再去获取连接。
return nil, errDBClosed
}
// 检查用户传递进来的Context是否过期了
select {
default:
// 如果用户那边使用了ctx.Done(),毫无疑问,会进入这个case中,返回Ctx错误
case <-ctx.Done():
db.mu.Unlock()
return nil, ctx.Err()
}
// 连接被重用的时间,如果为0,表示 理论上这个连接永不过期,一直可以被使用
lifetime := db.maxLifetime // 看一下空闲连接池(他是个slice)是否是还有空闲的连接
numFree := len(db.freeConn)
// 如果获取策略是优先从连接池中获取,并且连接池中确实存在空闲的连接,就从freeConn中取连接使用。
if strategy == cachedOrNewConn && numFree > 0 {
// 假设空闲池还剩下五条连接:【1,2,3,4,5】
// 取出第一条 conn == 1
conn := db.freeConn[0]
// 切片的拷贝,实现remove掉第一个连接的目的。
copy(db.freeConn, db.freeConn[1:])
// 如果db.freeConn[1:]会导致freeConn变小,所以这里是 db.freeConn = db.freeConn[:numFree-1]
db.freeConn = db.freeConn[:numFree-1]
// 这里获取的连接是driverConn,它其实是对真实连接,driver.Conn的封装。
// 在driver.Conn的基础上多一层封装可以实现在driver.Conn的基础上,加持上状态信息,如下
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未曾被标记为 lastErr状态。
// 一旦被标记为这个状态说明 ConnectionRestter协程在重置conn的状态时发生了错误。也就是这个连接其实已经坏掉了,不可使用。
conn.Lock()
err := conn.lastErr
conn.Unlock()
// 如果检测到这种错误,driver.ErrBadConn 表示连接不可用,关闭连接,返回错误。
if err == driver.ErrBadConn {
conn.Close()
return nil, driver.ErrBadConn
}
return conn, nil
} // Out of free connections or we were asked not to use one. If we're not
// allowed to open any more connections, make a request and wait.
// db.maxOpen > 0 表示当前DB实例允许打开连接
// db.numOpen >= db.maxOpen表示当前DB能打开的连接数,已经大于它能打开的最大连接数,就构建一个request,然后等待获取连接
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. // 构建connRequest这个channel,缓存大小是1
// 用于告诉connectionOpener协程,需要打开一个新的连接。
req := make(chan connRequest, 1) /**
nextRequestKeyLocked函数如下: func (db *DB) nextRequestKeyLocked() uint64 {
next := db.nextRequest
db.nextRequest++
return next
} 主要作用就是将nextRequest+1,
至于这个nextRequest的作用我们前面也说过了,它相当于binlog中的next_trx下一个事物的事物id。
言外之意是这个nextRequest递增的(因为这段代码被加了lock)。
看如下的代码中,将这个自增后的nextRequest当返回值返回出去。
然后紧接着将它作为map的key 至于这个map嘛:
在本文一开始的位置,我们介绍了DB结构体有这样一个属性,连接请求的map, key是自增的int64类型的数,
用于唯一标示这个请求分配的
connRequests map[uint64]chan connRequest
*/
reqKey := db.nextRequestKeyLocked()
// 将这个第n个请求对应channel缓存起来,开始等待有合适的机会分配给他连接
db.connRequests[reqKey] = req
// 等待数增加,解锁
db.waitCount++
db.mu.Unlock() waitStart := time.Now() // Timeout the connection request with the context.
// 进入下面的slice中
select {
// 如果客户端传入的上下文超时了,进入这个case
case <-ctx.Done():
// Remove the connection request and ensure no value has been sent
// on it after removing.
// 当上下文超时时,表示上层的客户端代码想断开,意味着在这个方法收到这个信号后需要退出了
// 这里将db的connRequests中的reqKey清除,防止还给他分配一个连接。
db.mu.Lock()
delete(db.connRequests, reqKey)
db.mu.Unlock() atomic.AddInt64(&db.waitDuration, int64(time.Since(waitStart)))
// 这里也会尝试从req channel中获取一下有没有可用的连接
// 如果有的话执行 db.putConn(ret.conn, ret.err, false) ,目的是释放掉这个连接
select {
default:
case ret, ok := <-req:
if ok && ret.conn != nil {
// 看到这里只需要知道他是用来释放连接的就ok,继续往下看,稍后再杀回来
db.putConn(ret.conn, ret.err, false)
}
}
//返回ctx异常。
return nil, ctx.Err()
// 尝试从 reqchannel 中取出连接
case ret, ok := <-req:
atomic.AddInt64(&db.waitDuration, int64(time.Since(waitStart)))
// 处理错误
if !ok {
return nil, errDBClosed
}
// 检测连接是否过期了,前面也提到过,DB实例有维护一个参数,maxLifeTime,0表示永不过期
if ret.err == nil && ret.conn.expired(lifetime) {
ret.conn.Close()
return nil, driver.ErrBadConn
}
// 健壮性检查
if ret.conn == nil {
return nil, ret.err
} // Lock around reading lastErr to ensure the session resetter finished.
// 检查连接是否可用
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
}
}
// 代码能运行到这里说明上面的if条件没有被命中。
// 换句话说,来到这里说明具备如下条件
// 1:当前DB实例的空闲连接池中已经没有空闲连接了,获取明确指定,不从空闲池中获取连接,就想新建连接。
// 2: 当前DB实例允许打开连接
// 3: DB实例目前打开的连接数还没有到达它能打开的最大连接数的上限。 // 记录当前DB已经打开的连接数+1
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
}

5.4、释放连接

连接被是过后是需要被释放的

释放连接的逻辑封装在DB实例中

db.putConn(ret.conn, ret.err, false)

释放连接的流程图如下:

![未命名文件 (5)](/Users/dxm/Downloads/未命名文件 (5).png)

流程图根据如下的代码画出。

方法详细信息如下:

func (db *DB) putConn(dc *driverConn, err error, resetSession bool) {
// 释放连接的操作加锁
db.mu.Lock()
// debug的信息
if !dc.inUse {
if debugGetPut {
fmt.Printf("putConn(%v) DUPLICATE was: %s\n\nPREVIOUS was: %s", dc, stack(), db.lastPut[dc])
}
panic("sql: connection returned that was never out")
}
if debugGetPut {
db.lastPut[dc] = stack()
}
// 标记driverConn处理不可用的状态
dc.inUse = false for _, fn := range dc.onPut {
fn()
}
dc.onPut = nil // 本方法的入参中有参数err
// 当会话获取出这个连接后,发现这个连接过期了、或者被标记上来lastErr时,再调用这个putConn方法时,同时会将这个错误传递进来,然后在这里判断,当出现坏掉的连接时就不直接把这个连接放回空闲连接池了。
if err == driver.ErrBadConn {
// Don't reuse bad connections.
// Since the conn is considered bad and is being discarded, treat it
// as closed. Don't decrement the open count here, finalClose will
// take care of that.
// 这个方法的作用如下:
// 他会去判断当前DB维护的map的容量,也就是前面提到的那种情况:当DB允许打开连接,但是现在的连接数已经达到当前DB允许打开的最大连接数上限了,那么针对接下来想要获取连接的请求的处理逻辑就是,构建一个req channel,放入connRequests这个map中,表示他们正在等待连接的建立。
// 换句话说,这时系统时繁忙的,业务处于高峰,那么问题来了,现在竟然出现了一个坏掉的连接,那为了把对业务线的影响降到最低,是不是得主动新建一个新的连接放到空闲连接池中呢?
// db.maybeOpenNewConnections() 函数主要干的就是这个事。
// 方法详情如下
/*
func (db *DB) maybeOpenNewConnections() {
numRequests := len(db.connRequests)
if db.maxOpen > 0 {
numCanOpen := db.maxOpen - db.numOpen
if numRequests > numCanOpen {
numRequests = numCanOpen
}
}
for numRequests > 0 {
db.numOpen++ // optimistically
numRequests--
if db.closed {
return
}
// 它只是往这个 openerCh channel中写入一个空的结构体,会有专门的协程负责创建连接
db.openerCh <- struct{}{}
}
}
*/
db.maybeOpenNewConnections()
//  解锁,关闭连接,返回
db.mu.Unlock()
dc.Close()
return
}
if putConnHook != nil {
putConnHook(db, dc)
}
// 如果DB已经关闭了,标记 resetSession为 false
if db.closed {
// Connections do not need to be reset if they will be closed.
// Prevents writing to resetterCh after the DB has closed.
// 当DB都已经关了,意味着DB里面的连接池都没有了,那当然不需要关闭连接池中的连接了~
resetSession = false
}
// 如果DB没有关闭的话,进入if代码块
if resetSession {
// 将dricerConn中的Conn验证转换为driver.SessionResetter
if _, resetSession = dc.ci.(driver.SessionResetter); resetSession {
// 在此处锁定driverConn,以便在连接重置之前不会释放。
// 必须在将连接放入池之前获取锁,以防止在重置之前将其取出
dc.Lock()
}
}
// 真正将连接放回空闲连接池中
// 满足connRequest或将driverConn放入空闲池并返回true或false
/*
func (db *DB) putConnDBLocked(dc *driverConn, err error) bool {
// 检测如果DB都关闭块,直接返回flase
if db.closed {
return false
}
// 如果DB当前打开的连接数大于DB能打开的最大的连接数,返回false
if db.maxOpen > 0 && db.numOpen > db.maxOpen {
return false
}
//如果等待获取连接的map中有存货
if c := len(db.connRequests); c > 0 { var req chan connRequest
var reqKey uint64
// 取出map中的第一个key
for reqKey, req = range db.connRequests {
break
}
// 将这个key,value再map中删除
delete(db.connRequests, reqKey) // Remove from pending requests.
// 重新标记这个连接是可用的状态
if err == nil {
dc.inUse = true
}
// 将这个连接放入到 req channel中,给等待连接到会话使用
req <- connRequest{
conn: dc,
err: err,
}
return true // 来到这个if,说明此时没有任何请求在等待获取连接,并且没有发生错误,DB也没有关闭
} else if err == nil && !db.closed {
// 比较当前空闲连接池的大小(默认是2) 和 freeConn空闲连接数的数量
// 意思是,如果空闲的连接超出了这个规定的阈值,空闲连接是需要被收回的。
if db.maxIdleConnsLocked() > len(db.freeConn) {
// 收回
db.freeConn = append(db.freeConn, dc)
db.startCleanerLocked()
return true
}
// 如果空闲连接还没到阈值,保留这个连接当作空闲连接
db.maxIdleClosed++
}
// 收回空闲连接返回false
return false
}
*/ // 如果将连接成功放入了空闲连接池,或者将连接成功给了等待连接到会话使用,此处返回true
// 收回空闲连接返回false
// 代码详情就是在上面的这段注释中
added := db.putConnDBLocked(dc, nil)
db.mu.Unlock() // 如果
if !added {
// 如果DB没有关闭,进入if
if resetSession {
dc.Unlock()
}
dc.Close()
return
}
// 重新校验,如果连接关闭了,进入if
if !resetSession {
return
} // 如果负责重置 conn状态的线程阻塞住了,那么标记这个driverConn为lastErr
select {
default:
// If the resetterCh is blocking then mark the connection
// as bad and continue on.
dc.lastErr = driver.ErrBadConn
dc.Unlock()
case db.resetterCh <- dc:
}
}

5.5、connectionOpener

5.5.1、是什么?

这个connectionOpener是一个工作协程,它会去尝试消费指定的channel,负责创建数据库连接,其实在前面阅读获取连接的逻辑时,有这样的两种情况会阻塞等待connectionOpener来新创建连接:

第一种:当获取连接的策略是优先从cache连接池中获取出来,但是空闲连接池已经没有空闲的连接了,首先这时DB允许打开连接,但是DB能打开的连接数已经达到了它能打开的连接数的上线,所以得等待有空闲连接出现,或者等有连接被释放后,DB能当前打开的连接数小于了它能打开的连接数的最大值,这时它会被阻塞等待去尝试创建连接。

第二种:获取连接的策略不再是优先从空闲缓冲池中获取连接,直接明了的想获取最一条新连接,同样的此时DB已经打开的连接数大于它能打开连接数的上线,它会被阻塞等待创建连接。

5.5.2、什么时候开启的?
func OpenDB(c driver.Connector) *DB {
ctx, cancel := context.WithCancel(context.Background())
db := &DB{
connector: c,
openerCh: make(chan struct{}, connectionRequestQueueSize),
resetterCh: make(chan *driverConn, 50),
lastPut: make(map[*driverConn]string),
connRequests: make(map[uint64]chan connRequest),
stop: cancel,
}
// 可以看到他是在DB被实例化时开启的。
go db.connectionOpener(ctx)
go db.connectionResetter(ctx) return db
}
5.5.3、代码详情

可以看到它一直尝试从db的openerCh中获取内容,而且只要获取到了内容,就会调用方法打开连接。

// Runs in a separate goroutine, opens new connections when requested.
func (db *DB) connectionOpener(ctx context.Context) {
for {
select {
case <-ctx.Done():
return
// here
case <-db.openerCh:
db.openNewConnection(ctx)
}
}
}
5.5.4、谁往openerCh中投放消息?

往channl中投放消息的逻辑在db的mayBeOpenNewConnections中

func (db *DB) maybeOpenNewConnections() {
// 通过检查这个map的长度来决定是否往opennerCh中投放消息
numRequests := len(db.connRequests)
if db.maxOpen > 0 {
numCanOpen := db.maxOpen - db.numOpen
if numRequests > numCanOpen {
numRequests = numCanOpen
}
}
for numRequests > 0 {
db.numOpen++ // optimistically
numRequests--
if db.closed {
return
}
// 一旦执行了这一步,connectionOpener 就会监听到去创建连接。
db.openerCh <- struct{}{}
}
}
5.5.5、注意点:

在DB结构体中有这样一个属性

  // 连接池的大小,0意味着使用默认的大小2, 小于0表示不使用连接池
maxIdle int // zero means defaultMaxIdleConns; negative means 0

表示空闲连接池默认的大小,如果它为0,表示都没有缓存池,也就意味着会为所有想获取连接的请求创建新的conn,这是也就不会有这个opnerCh,更不会有connectionOpener

5.6、connectionCleaner

5.6.1、是什么?有啥用?

它同样以一条协程的形式存在,用于定时清理数据库连接池中过期的连接

func (db *DB) startCleanerLocked() {
if db.maxLifetime > 0 && db.numOpen > 0 && db.cleanerCh == nil {
db.cleanerCh = make(chan struct{}, 1)
go db.connectionCleaner(db.maxLifetime)
}
}
5.6.2、注意点

同样的,DB中存在一个参数:maxLifetime

它表示数据库连接最大的生命时长,如果将它设置为0,表示这个连接永不过期,既然所有的连接永不过期,就不会存在connectionCleaner去定时根据maxLifetime 来定时清理连接。

它的调用时机是:需要将连接放回到连接池时调用。

5.7、connectionRestter

5.7.1、作用

我们使用获取的连接的封装结构体是driverConn,其实它是会driver包下的Conn连接的又一层封装,目的是增强

driver包下的Conn的,多出来了一些状态。当将使用完毕的连接放入连接池时,就得将这些状态清除掉。

使用谁去清除呢?就是这个go 协程:connectionRestter

当connectionRestter碰到错误时,会将这个conn标记为lastErr,连接使用者在使用连接时会先校验conn的诸多状态,比如出现lastErr,会返回给客户端 badConnErr

六、MySQL连接池所受的限制

数据库连接池大大小到底设置为多少,得根据业务流量已经数据库所在机器的性能综合考虑。

mysql连接数到配置在 my.cnf中,具体的参数是max_connections。

当业务流量异常猛烈时,很可能会出现这个问题:to many connections

对于操纵系统内核来说,当他接受到一个tcp请求就会在本地创建一个由文件系统管理的socket文件。在linux中我们将它叫做文件句柄。

linux为防止单一进程将系统资源全部耗费掉,会限制进程最大能打开的连接数为1024,这意味着,哪怕通过改配置文件,将mysql能打开的连接池设置为9999,事实上它能打开的文件数最多不会超过1024。

这个问题也好解决:

命令:设置单个进程能打开的最大连接数为65535

ulimit -HSn 65535

通过命令: 查看进程被限制的使用各种资源的量

ulimit -a 

core file size: 进程崩溃是转储文件大小限制
man loaded memort 最大锁定内存大小
open file 能打开的文件句柄数

这些变量定义在 /etc/security/limits.conf配置文件中。

七、关于失效的连接

情况1: 客户端主动断开

如果是客户端主动将连接close(), 那往合格连接中写数据时会得到ErrBadConn的错误,如果此时依然可以重试,将会获取新的连接。

代码如下:

func (db *DB) ExecContext(ctx context.Context, query string, args ...interface{}) (Result, error) {
var res Result
var err error
for i := 0; i < maxBadConnRetries; i++ {
res, err = db.exec(ctx, query, args, cachedOrNewConn)
if err != driver.ErrBadConn {
break
}
}
if err == driver.ErrBadConn {
return db.exec(ctx, query, args, alwaysNewConn)
}
return res, err
}

情况2: 服务端挂啦

因为这种数据库连接底层使用的是tcp实现。(tcp本身是支持全双工的,客户端和服务端支持同时往对方发送数据)依赖诸如:校验和、确认应答和序列号机制、超时重传、连接管理(3次握手,4次挥手)、以及滑动窗口、流量控制、拥赛避免去实现整个数据交互的可靠性,协调。

这时客户端拿着一条自认为是正常的连接,往连接里面写数据。然鹅,另一端端服务端已经挂了~,但是不幸的是,客户端的tcp连接根本感知不到~~~。

但是当它去读取服务端的返回数据时会遇到错误:unexceptBadConn EOF

八、连接的有效性

  • 思路1:

设置连接的属性: maxLifeTime

上面也说过了,当设置了这个属性后,DB会开启一条协程connectionCleaner,专门负责清理过期的连接。

这在一定程度上避免了服务端将连接断掉后,客户端无感知的情况。

maxLifeTime的值到底设置多大?参考值,比数据库的wait_timeout小一些就ok。

  • 思路2:

主动检查连接的有效性。

比如在连接放回到空闲连接池前ping测试。在使用连接发送数据前进行连通性测试。

Golang SQL连接池梳理的更多相关文章

  1. Golang 通用连接池库 Golang-Pool

    Golang 实现的连接池 功能: * 连接池中连接类型为interface{},使得更加通用 * 链接的最大空闲时间,超时的链接将关闭丢弃,可避免空闲时链接自动失效问题 * 使用channel处理池 ...

  2. golang redis连接池使用方法

    package main import ( "fmt" "github.com/garyburd/redigo/redis" ) var pool *redis ...

  3. golang的连接池例子

    github.com/jolestar/go-commons-pool 测试代码 package main import ( "github.com/jolestar/go-commons- ...

  4. Go组件学习——database/sql数据库连接池你用对了吗

    1.案例 case1: maxOpenConns > 1 func fewConns() { db, _ := db.Open("mysql", "root:roo ...

  5. [Go] golang实现mysql连接池

    golang中连接mysql数据库,需要使用一个第三方类库github.com/go-sql-driver/mysql,在这个类库中就实现了mysql的连接池,并且只需要设置两个参数就可以实现 一般连 ...

  6. golang mgo的mongo连接池设置:必须手动加上maxPoolSize

    本司礼物系统使用了golang的 mongo库 mgo,中间踩了一些坑,总结下避免大家再踩坑 golang的mgo库说明里是说明了开启连接复用的,但观察实验发现,这并没有根本实现连接的控制,连接复用仅 ...

  7. SQL Server的Execute As与连接池结合使用的测试

    简介     在SQL Server中,Execute As关键字允许当前账户在特定上下文中以另一个用户或登录名的身份执行SQL语句,比如用户张三有权限访问订单表,用户李四并没有权限访问订单表,那么给 ...

  8. JDBC连接池以及动态SQL处理

    复习一下: 1.先创建一个properties配置文件 ClasssName=oracle.jdbc.driver.OracleDriver url=jdbc:oracle:thin:@服务器IP:端 ...

  9. Sql Server 连接池及其用法

    其实我们一直在使用SqlServer的连接池.在连接字符串中,Pooling为是否启用连接池,默认值为true,表示启用. 与连接池相关的两个重要参数是 Min Pool Size和 Max Pool ...

  10. SQL Server 连接池 (ADO.NET) MSDN

    连接到数据库服务器通常由几个需要很长时间的步骤组成. 必须建立物理通道(例如套接字或命名管道),必须与服务器进行初次握手,必须分析连接字符串信息,必须由服务器对连接进行身份验证,必须运行检查以便在当前 ...

随机推荐

  1. 【bzoj1010】 HNOI2008—玩具装箱toy

    http://www.lydsy.com/JudgeOnline/problem.php?id=1010 (题目链接) 题意 给定N个物品,可以连续的划分为若干个组,每个组的代价是(物品数-1+每个物 ...

  2. Struts2(三):新建Struts2工程

    下载的struts2xx-all.zip包对搭建项目的作用 一般情况下,我们下载一个Struts2的full包时,并不知道新建过程中需要哪些包,那么这时我们可以从下载的包中解压出的目录\apps\st ...

  3. c#中using System.Runtime.Serialization.Json;不能引用

    原因:生成项目对于的.net版本是2.0的.需要手动切换成4.0 操作步骤:右键项目,属性页面,下拉框选择最高版本.net 截图:

  4. Learning Django Resources

    Learning Django Django makes it easier to build better Web apps more quickly and with less code. Web ...

  5. Excel大写和小写转换函数

    Excel中的大写和小写转换函数 (1).转换为所有小写字母:lower函数 (2).转换为所有大写字母:upper函数 (3).转换为首字母大写,其余小写字母:proper函数 三种函数的使用方式. ...

  6. Delaunay triangulation

    1,先花个圆: detail模式执行. #define XY 0x00 #define XZ 0x01 #define YZ 0x02 #define pi 3.1415926 #define clo ...

  7. Anaconda 虚拟环境的使用

    目录 前言 1. 创建虚拟环境 2. 虚拟环境管理 3. Conda虚拟环境的包管理 前言 今天把anaconda进行了滚动更新,实体环境python版本也相应从3.6跟新到了3.7.但是问题来了,之 ...

  8. (整理)在REHL6.5上部署ASP.NET MVC

    最近项目要使用Linux服务器(REHL6.5)+MySQL,因此特尝试操作. 1 Linux 安装Jexus 1.1 下载Jexus 因为服务器没有安装Xwindows,Jexus的下载又出现问题, ...

  9. ubuntu更改分辨率

    1. 输入:$cvt 1920 1080 2 输入: $xrandr 3 输入: $sudo xrandr --newmode "1920x1080_60.00" 173.00 1 ...

  10. 20155228 2017-5-10 课堂测试:Arrays和String单元测试

    20155228 2017-5-10 课堂测试:Arrays和String单元测试 题目和要求 在IDEA中以TDD的方式对String类和Arrays类进行学习 测试相关方法的正常,错误和边界情况 ...