Golang SQL 连接池

$ tree $GOROOT/src/database/sql
├── convert.go # scan row
├── convert_test.go
├── ctxutil.go # 判断 ctx，然后执行 prepare/exec/query/close 等操作
├── doc.txt
├── driver
│   ├── driver.go # 定义了实现数据库驱动所需的接口，由 sql 包和具体的驱动包来实现
│   ├── types.go # 数据类型的别名和转换
│   └── types_test.go
├── example_cli_test.go
├── example_service_test.go
├── example_test.go
├── fakedb_test.go
├── sql.go # 关于 SQL 数据库的一些通用接口和类型，包括：连接池、数据类型、连接、事务、状态
└── sql_test.go

// DB is a database handle representing a pool of zero or more
// underlying connections. It's safe for concurrent use by multiple
// goroutines.
//
// The sql package creates and frees connections automatically; it
// also maintains a free pool of idle connections. If the database has
// a concept of per-connection state, such state can be reliably observed
// within a transaction (Tx) or connection (Conn). Once DB.Begin is called, the
// returned Tx is bound to a single connection. Once Commit or
// Rollback is called on the transaction, that transaction's
// connection is returned to DB's idle connection pool. The pool size
// can be controlled with 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 // 等待新连接的总时间，用于统计

    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 // 锁
    freeConn     []*driverConn // 可用连接池
    connRequests map[uint64]chan connRequest // 连接请求表，key 是分配的自增键
    nextRequest  uint64 // 连接请求的自增键
    numOpen      int    // 已经打开 + 即将打开的连接数
    // 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.
    openerCh          chan struct{} // 告知 connectionOpener 需要新的连接
    resetterCh        chan *driverConn // connectionResetter 函数，连接放回连接池的时候会用到
    closed            bool
    dep               map[finalCloser]depSet
    lastPut           map[*driverConn]string // debug 时使用，记录上一个放回的连接
    maxIdle           int                    // 连接池大小，默认大小为 2，<= 0 时不使用连接池
    maxOpen           int                    // 最大打开的连接数，<= 0 不限制
    maxLifetime       time.Duration          // 一个连接可以被重用的最大时限，也就是它在连接池中的最大存活时间，0 表示可以一直重用
    cleanerCh         chan struct{} // 告知 connectionCleaner 清理连接
    waitCount         int64 // 等待的连接总数
    maxIdleClosed     int64 // 释放连接时，因为连接池已满而被关闭的连接总数
    maxLifetimeClosed int64 // 因为超过存活时间而被关闭的连接总数

    stop func() // stop cancels the connection opener and the session resetter.
}

// 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)
type driverConn struct {
    db        *DB // 数据库句柄
    createdAt time.Time

    sync.Mutex  // 锁
    ci          driver.Conn // 对应具体的连接
    closed      bool // 是否标记关闭
    finalClosed bool // 是否最终关闭
    openStmt    map[*driverStmt]bool // 在这个连接上打开的状态
    lastErr     error // connectionResetter 的返回结果

    // guarded by db.mu
    inUse      bool // 连接是否占用
    onPut      []func() // 连接归还时要运行的函数，在 noteUnusedDriverStatement 添加
    dbmuClosed bool     // 和 closed 状态一致，但是由锁保护，用于 removeClosedStmtLocked
}

// conn returns a newly-opened or cached *driverConn.
func (db *DB) conn(ctx context.Context, strategy connReuseStrategy) (*driverConn, error) {
    db.mu.Lock() // 获取连接要对整个连接池加锁
    if db.closed {
        // 连接池已经被关闭
        db.mu.Unlock()
        return nil, errDBClosed
    }
    // Check if the context is expired.
    // 检查连接请求是被否取消或过期
    select {
    default:
    case <-ctx.Done(): // 连接请求被取消
        db.mu.Unlock()
        return nil, ctx.Err()
    }
    lifetime := db.maxLifetime

    // Prefer a free connection, if possible.
    // 优先使用空闲连接
    numFree := len(db.freeConn)
    if strategy == cachedOrNewConn && numFree > 0 {
        conn := db.freeConn[0] // 取走第一个空闲连接
        copy(db.freeConn, db.freeConn[1:]) // 将 1 ~ n-1 的连接拷贝到 0 ~ n-2，也就是左移
        db.freeConn = db.freeConn[:numFree-1] // 删除 n-1，表示从 n 个空闲连接中取走了一个
        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.Lock()
        err := conn.lastErr // 检查这个连接会话重置是否完成
        conn.Unlock()
        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.
    // 设置了最大连接数
    // 如果已打开的连接数 >= 最大连接数，就发一个连接请求
    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)
        reqKey := db.nextRequestKeyLocked()
        db.connRequests[reqKey] = req // 新增一个连接请求
        db.waitCount++ // 等待连接数 +1
        db.mu.Unlock()

        waitStart := time.Now() // 开始计算连接等待时间

        // Timeout the connection request with the context.
        select {
        case <-ctx.Done(): // 超时
            // Remove the connection request and ensure no value has been sent
            // on it after removing.
            db.mu.Lock()
            delete(db.connRequests, reqKey) // 删掉连接请求
            db.mu.Unlock()

            atomic.AddInt64(&db.waitDuration, int64(time.Since(waitStart)))

            select {
            default:
            case ret, ok := <-req: // 连接来了
                if ok && ret.conn != nil {
                    db.putConn(ret.conn, ret.err, false) // 把连接放回连接池，而且不用重置会话
                }
            }
            return nil, ctx.Err()
        case ret, ok := <-req: // 连接来了
            atomic.AddInt64(&db.waitDuration, int64(time.Since(waitStart)))

            if !ok { // 连接被关闭
                return nil, errDBClosed // 说明 DB 已经关闭了
            }
            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
        }
    }

    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
}

// Assumes db.mu is locked.
// If there are connRequests and the connection limit hasn't been reached,
// then tell the connectionOpener to open new connections.
// 如果有连接请求，并且没有达到连接数的限制，告知 connectionOpener 打开新的连接
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
        }
        db.openerCh <- struct{}{} // 告知 connectionOpener 打开新的连接，由 connectionOpener 执行打开连接的操作
    }
}

// putConn adds a connection to the db's free pool.
// err is optionally the last error that occurred on this connection.
// 业务完成，将连接放回连接池
func (db *DB) putConn(dc *driverConn, err error, resetSession bool) {
    db.mu.Lock()
    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()
    }
    dc.inUse = false

    for _, fn := range dc.onPut {
        fn()
    }
    dc.onPut = nil

    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.maybeOpenNewConnections()
        db.mu.Unlock()
        dc.Close()
        return
    }
    if putConnHook != nil {
        putConnHook(db, dc)
    }
    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 关闭时，resetterCh 也关闭，如果还执行会话重置操作会导致写已关闭的 channel
        resetSession = false
    }
    if resetSession {
        if _, resetSession = dc.ci.(driver.SessionResetter); resetSession { // 检查驱动是否实现了会话重置接口
            // Lock the driverConn here so it isn't released until
            // the connection is reset.
            // The lock must be taken before the connection is put into
            // the pool to prevent it from being taken out before it is reset.
            dc.Lock()
        }
    }
    added := db.putConnDBLocked(dc, nil) // 函数名带 Locked 表示已经加过锁
    db.mu.Unlock()

    if !added {
        if resetSession {
            dc.Unlock()
        }
        dc.Close() // 回池出现问题，强行关闭连接
        return
    }
    if !resetSession {
        return
    }
    select {
    default:
        // If the resetterCh is blocking then mark the connection
        // as bad and continue on.
        // 如果 resetterCh 阻塞，将连接标记为 BadConn
        dc.lastErr = driver.ErrBadConn
        dc.Unlock()
    case db.resetterCh <- dc:
    }
}

// Satisfy a connRequest or put the driverConn in the idle pool and return true
// or return false.
// putConnDBLocked will satisfy a connRequest if there is one, or it will
// return the *driverConn to the freeConn list if err == nil and the idle
// connection limit will not be exceeded.
// If err != nil, the value of dc is ignored.
// If err == nil, then dc must not equal nil.
// If a connRequest was fulfilled or the *driverConn was placed in the
// freeConn list, then true is returned, otherwise false is returned.
// 尝试将 driverConn 放到连接池或者满足一个连接请求
func (db *DB) putConnDBLocked(dc *driverConn, err error) bool {
    if db.closed {
        return false
    }
    // 即将打开 + 已打开的连接数超过了最大打开的连接数
    if db.maxOpen > 0 && db.numOpen > db.maxOpen {
        return false
    }
    if c := len(db.connRequests); c > 0 {
        var req chan connRequest
        var reqKey uint64
        for reqKey, req = range db.connRequests { // 从表中随便拿一个
            break
        }
        delete(db.connRequests, reqKey) // Remove from pending requests.
        if err == nil {
            dc.inUse = true // 连接即将被复用
        }
        // 将连接发送给连接请求
        req <- connRequest{
            conn: dc,
            err:  err,
        }
        return true
    } else if err == nil && !db.closed {
        // 如果连接池 < 最大闲散连接数，放回连接池
        if db.maxIdleConnsLocked() > len(db.freeConn) {
            db.freeConn = append(db.freeConn, dc) // 放回连接池中
            db.startCleanerLocked()
            return true
        }
        db.maxIdleClosed++
    }
    return false
}