database-sql包源码剖析

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29

//数据库配置
const (
    userName = "root"
    password = "123456"
    ip = "127.0.0.1"
    port = "3306"
    dbName = "loginserver"
)
//Db数据库连接池
var DB *sql.DB

//注意方法名大写，就是public
func InitDB()  {
    //构建连接："用户名:密码@tcp(IP:端口)/数据库?charset=utf8"
    path := strings.Join([]string{userName, ":", password, "@tcp(",ip, ":", port, ")/", dbName, "?charset=utf8"}, "")

    //打开数据库,前者是驱动名，所以要导入： _ "github.com/go-sql-driver/mysql"
    DB, _ = sql.Open("mysql", path)
    //设置数据库最大连接数
    DB.SetConnMaxLifetime(100)
    //设置上数据库最大闲置连接数
    DB.SetMaxIdleConns(10)
    //验证连接
    if err := DB.Ping(); err != nil{
        fmt.Println("opon database fail")
        return
    }
    fmt.Println("connnect success")
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

func main() {
    db, err := sql.Open("mysql", "root:@tcp(127.0.0.1:3306)/test?parseTime=true")
    if err != nil{
        log.Fatal(err)
    }
    defer db.Close()

    _, err = db.Exec("CREATE TABLE IF NOT EXISTS test.hello(world varchar(50))")
    if err != nil{
        log.Fatalln(err)
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15

func main() {
    db, err := sql.Open("mysql", "root:@tcp(127.0.0.1:3306)/test?parseTime=true")

    ...

    rs, err := db.Exec("INSERT INTO test.hello(world) VALUES ('hello world')")
    if err != nil{
        log.Fatalln(err)
    }
    rowCount, err := rs.RowsAffected()
    if err != nil{
        log.Fatalln(err)
    }
    log.Printf("inserted %d rows", rowCount)
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20

func main() {
    db, err := sql.Open("mysql", "root:@tcp(127.0.0.1:3306)/test?parseTime=true")

    ...

    rows, err := db.Query("SELECT world FROM test.hello")
    if err != nil{
        log.Fatalln(err)
    }

    for rows.Next(){
        var s string
        err = rows.Scan(&s)
        if err !=nil{
            log.Fatalln(err)
        }
        log.Printf("found row containing %q", s)
    }
    rows.Close()
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

db, err := sql.Open("driverName", "dataSourceName")
if err != nil{
    log.Fatalln(err)
}

defer db.Close()

err = db.Ping()
if err != nil{
   log.Fatalln(err)
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59

// 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 // Total time waited for new connections.
    //数据库实现驱动
	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
    // 空闲连接
    freeConn     []*driverConn
    // 阻塞请求队列，等连接数达到最大限制时，后续请求将插入此队列等待可用连接
    connRequests map[uint64]chan connRequest
    // 记录下一个key用于connRequests map的key
	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.
    // 用来接收创建新连接的信号
    // connectionOpener方法会读取该channel的信号，而当需要创建连接的时候，maybeOpenNewConnections就会往该channel发信号。
    // 当调用db.Close()的时候，该channel就会被关闭
    openerCh          chan struct{}
	closed            bool
	dep               map[finalCloser]depSet
	lastPut           map[*driverConn]string // stacktrace of last conn's put; debug only
    maxIdleCount      int                    // zero means defaultMaxIdleConns; negative means 0
    // 最大打开连接数
    maxOpen           int                    // <= 0 means unlimited
    // 连接最大存活时间
	maxLifetime       time.Duration          // maximum amount of time a connection may be reused
	maxIdleTime       time.Duration          // maximum amount of time a connection may be idle before being closed
	cleanerCh         chan struct{}
    waitCount         int64 // Total number of connections waited for.
	maxIdleClosed     int64 // Total number of connections closed due to idle count.
    maxIdleTimeClosed int64 // Total number of connections closed due to idle time.
    // 因为超过存活时间而被关闭的连接总数
	maxLifetimeClosed int64 // Total number of connections closed due to max connection lifetime limit.

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

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

// 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  // guards following  // 锁
	ci          driver.Conn// 对应具体的连接
	needReset   bool // The connection session should be reset before use if true.
	closed      bool    // 是否标记关闭
	finalClosed bool // ci.Close has been called    // 是否最终关闭
	openStmt    map[*driverStmt]bool    // 在这个连接上打开的状态

	// guarded by db.mu
	inUse      bool// 连接是否占用
    returnedAt time.Time // Time the connection was created or returned.
    // 连接归还时要运行的函数，在 noteUnusedDriverStatement 添加
    onPut      []func()  // code (with db.mu held) run when conn is next returned
     // 和 closed 状态一致，但是由锁保护，用于 removeClosedStmtLocked
	dbmuClosed bool      // same as closed, but guarded by db.mu, for removeClosedStmtLocked
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67

// Open opens a database specified by its database driver name and a
// driver-specific data source name, usually consisting of at least a
// database name and connection information.
//
// Most users will open a database via a driver-specific connection
// helper function that returns a *DB. No database drivers are included
// in the Go standard library. See https://golang.org/s/sqldrivers for
// a list of third-party drivers.
//
// Open may just validate its arguments without creating a connection
// to the database. To verify that the data source name is valid, call
// Ping.
//
// The returned DB is safe for concurrent use by multiple goroutines
// and maintains its own pool of idle connections. Thus, the Open
// function should be called just once. It is rarely necessary to
// close a DB.
func Open(driverName, dataSourceName string) (*DB, error) {
	driversMu.RLock()
	driveri, ok := drivers[driverName]
	driversMu.RUnlock()
	if !ok {
		return nil, fmt.Errorf("sql: unknown driver %q (forgotten import?)", driverName)
	}

	if driverCtx, ok := driveri.(driver.DriverContext); ok {
		connector, err := driverCtx.OpenConnector(dataSourceName)
		if err != nil {
			return nil, err
		}
		return OpenDB(connector), nil
	}

	return OpenDB(dsnConnector{dsn: dataSourceName, driver: driveri}), nil
}

// OpenDB opens a database using a Connector, allowing drivers to
// bypass a string based data source name.
//
// Most users will open a database via a driver-specific connection
// helper function that returns a *DB. No database drivers are included
// in the Go standard library. See https://golang.org/s/sqldrivers for
// a list of third-party drivers.
//
// OpenDB may just validate its arguments without creating a connection
// to the database. To verify that the data source name is valid, call
// Ping.
//
// The returned DB is safe for concurrent use by multiple goroutines
// and maintains its own pool of idle connections. Thus, the OpenDB
// function should be called just once. It is rarely necessary to
// close a DB.

func OpenDB(c driver.Connector) *DB {
	ctx, cancel := context.WithCancel(context.Background())
	db := &DB{
		connector:    c,
		openerCh:     make(chan struct{}, connectionRequestQueueSize),
		lastPut:      make(map[*driverConn]string),
		connRequests: make(map[uint64]chan connRequest),
		stop:         cancel,
	}

	go db.connectionOpener(ctx)

	return db
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29

func (db *DB) Close() error {
	db.mu.Lock()
	if db.closed { // Make DB.Close idempotent
		db.mu.Unlock()
		return nil
	}
	close(db.openerCh)
	if db.cleanerCh != nil {
		close(db.cleanerCh)
	}
	var err error
	fns := make([]func() error, 0, len(db.freeConn))
	for _, dc := range db.freeConn {
		fns = append(fns, dc.closeDBLocked())
	}
	db.freeConn = nil
	db.closed = true
	for _, req := range db.connRequests {
		close(req)
	}
	db.mu.Unlock()
	for _, fn := range fns {
		err1 := fn()
		if err1 != nil {
			err = err1
		}
	}
	return err
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31

func (db *DB) Query(query string, args ...interface{}) (*Rows, error) {
    return db.QueryContext(context.Background(), query, args...)
}
func (db *DB) QueryContext(ctx context.Context, query string, args ...interface{}) (*Rows, error) {
    var rows *Rows
    var err error
    // 优先用池子里的连接
    // maxBadConnRetries是一个常量，值为2
    for i := 0; i < maxBadConnRetries; i++ {
        // cacheOrNewConn: 是获取连接策略常量，优先从连接池中获取一个可用的连接，如果没有可用连接，此时如果连接数已经达到上限，则等待，否则则直接创建一个新连接。
        rows, err = db.query(ctx, query, args, cachedOrNewConn)
        // 如果连接异常，则重试
        if err != driver.ErrBadConn {
            break
        }
    }
    // 当然池子里的连接有可能会过期，如果重试两次还取不到可以用的连接
    // 那就从数据库请求一个新的
    if err == driver.ErrBadConn {
        // 再重试一遍，使用创建连接的策略
        return db.query(ctx, query, args, alwaysNewConn)
    }
    return rows, err
}
  func (db *DB) query(ctx context.Context, query string, args []interface{}, strategy connReuseStrategy) (*Rows, error) {
    dc, err := db.conn(ctx, strategy)
    if err != nil {
        return nil, err
    }
    return db.queryDC(ctx, nil, dc, dc.releaseConn, query, args)
}

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154

// conn returns a newly-opened or cached *driverConn.
// conn 获取一个缓存中的或者新打开的连接
func (db *DB) conn(ctx context.Context, strategy connReuseStrategy) (*driverConn, error) {
    //从连接池中获取连接时首先要对整个连接池加锁，如果连接池已经事先被关掉了，直接返回 errDBClosed 错误。如果连接池无恙，将会评估连接请求是否取消或过期。
	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
    //尽可能优先使用空闲的连接而不是新建一条连接（这也是连接池存在的意义）。看一下是否还剩下空闲连接，如果还有，就取第 0 条连接出来，然后左移所有连接填补空位。这里对连接本身操作都会上锁。
    // Prefer a free connection, if possible.
    // 查看是否有空闲的连接 如果有则直接使用空闲连接
    // 获取空闲的连接，判断是否超时，无超时返回调用方，超时返回err, 调用方会重试，默认是重试2次
	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]
        // 通过以上两步，成功从freeConn中删掉了开头的第一个连接
        conn.inUse = true
        // 检查连接是否已经过期
        // 注意如果过期并没有尝试取下一个，而是直接返回了错误
        // 由上层来继续发起连接，我理解这样会保持不同层之间逻辑的干净
		if conn.expired(lifetime) {
            // 如果超时了，则关闭该连接，返回该错误时，外部会重试
			db.maxLifetimeClosed++
			db.mu.Unlock()
			conn.Close()
			return nil, driver.ErrBadConn
		}
		db.mu.Unlock()

		// Reset the session if required.
		if err := conn.resetSession(ctx); 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
    //如果没有空闲连接了，而且已打开的 + 即将打开的连接数超过了限定的最大打开的连接数，就要发送一条连接请求然后排队（不会新建连接）。等待排队期间同时监听连接请求是否取消或过期，如果此时连接被取消很不巧正好有连接来了，就将连接放回连接池中；如果等着等着连接来了，会先检查这个连接的上一次会话是否被重置，确认没问题就用这条连接。
	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.
        // 创建一个chan
        req := make(chan connRequest, 1)
        // 获取下一个request 作为map 中的key
		reqKey := db.nextRequestKeyLocked()
		db.connRequests[reqKey] = req
		db.waitCount++
		db.mu.Unlock()

		waitStart := nowFunc()

		// Timeout the connection request with the context.
		select {
        //下面就是select+ctx判断请求有没有超时，还有等待队列是否接收到数据
        // 注意这个select没有default分支，所以会一直阻塞在这两个分支上
        // 直到任何一个分支的条件满足，即：1. query被取消了 2. 有free连接可用了
		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()
        // 如果没有取消则从req chan中获取数据 阻塞主一直等待有conn数据传入
		case ret, ok := <-req:
			atomic.AddInt64(&db.waitDuration, int64(time.Since(waitStart)))

			if !ok {// 连接被关闭
				return nil, errDBClosed// 说明 DB 已经关闭了
			}
			// Only check if the connection is expired if the strategy is cachedOrNewConns.
			// If we require a new connection, just re-use the connection without looking
			// at the expiry time. If it is expired, it will be checked when it is placed
			// back into the connection pool.
			// This prioritizes giving a valid connection to a client over the exact connection
            // lifetime, which could expire exactly after this point anyway.
            // 判断超时
			if strategy == cachedOrNewConn && ret.err == nil && ret.conn.expired(lifetime) {
				db.mu.Lock()
				db.maxLifetimeClosed++
				db.mu.Unlock()
				ret.conn.Close()
				return nil, driver.ErrBadConn
			}
			if ret.conn == nil {
				return nil, ret.err
			}

			// Reset the session if required.
			if err := ret.conn.resetSession(ctx); err == driver.ErrBadConn {
				ret.conn.Close()
				return nil, driver.ErrBadConn
			}
			return ret.conn, ret.err
		}
    }
    // 如果还没到限定的最大打开的连接数，新建一条
    // 如果没有空闲连接且尚未达到最大连接数的限制
    // 那就开一个新连接
	db.numOpen++ // optimistically
    db.mu.Unlock()
    // 调用connector的Connect方法建立连接
	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(),
		returnedAt: nowFunc(),
		ci:         ci,
		inUse:      true,
	}
	db.addDepLocked(dc, dc)
	db.mu.Unlock()
	return dc, nil
}

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121

func (dc *driverConn) releaseConn(err error) {
	dc.db.putConn(dc, err, true)
}

// 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) {
	if err != driver.ErrBadConn {
		if !dc.validateConnection(resetSession) {
			err = driver.ErrBadConn
		}
	}
    db.mu.Lock()
    // 检查连接是否在使用中
	if !dc.inUse {
		db.mu.Unlock()
		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 err != driver.ErrBadConn && dc.expired(db.maxLifetime) {// 连接有问题
		db.maxLifetimeClosed++
		err = driver.ErrBadConn
	}
	if debugGetPut {
		db.lastPut[dc] = stack()
    }
    // 设置已经在使用中
	dc.inUse = false
	dc.returnedAt = nowFunc()

	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)
    }
    // 调用方法 释放连接
    // 函数名带 Locked 表示已经加过锁
	added := db.putConnDBLocked(dc, nil)
	db.mu.Unlock()
    // 判断如果没有加到了空闲列表中 dc关闭
	if !added {
        // 回池出现问题，强行关闭连接
		dc.Close()
		return
	}
}

// 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.
func (db *DB) putConnDBLocked(dc *driverConn, err error) bool {
	if db.closed {
		return false
    }
    // 即将打开 + 已打开的连接数超过了最大打开的连接数
	if db.maxOpen > 0 && db.numOpen > db.maxOpen {
		return false
    }
    // 如果等待chan列表大于0
    // 判断有没有等待队列
    // 如果有connRequests（连接需求），优先满足它
	if c := len(db.connRequests); c > 0 {
		var req chan connRequest
        var reqKey uint64
        // 获取map 中chan和key
		for reqKey, req = range db.connRequests {
            // 从表中随便拿一个
			break
        }
        // 从列表中删除chan
		delete(db.connRequests, reqKey) // Remove from pending requests.
		if err == nil {
			dc.inUse = true
        }
        // 把连接传入chan中 让之前获取连接被阻塞的获取函数继续
        // 将连接发送给连接请求
		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
        }
        // 为啥直接增加了maxIdleClosed而没有做什么呢？
        // 因为这种情况下本函数会返回false，进而导致
        // 连接被关闭
		db.maxIdleClosed++
	}
	return false
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54

// This is the size of the connectionOpener request chan (DB.openerCh).
// This value should be larger than the maximum typical value
// used for db.maxOpen. If maxOpen is significantly larger than
// connectionRequestQueueSize then it is possible for ALL calls into the *DB
// to block until the connectionOpener can satisfy the backlog of requests.
var connectionRequestQueueSize = 1000000

// Runs in a separate goroutine, opens new connections when requested.
func (db *DB) connectionOpener(ctx context.Context) {
	for {
		select {
		case <-ctx.Done():
			return
		case <-db.openerCh:
			db.openNewConnection(ctx)
		}
	}
}

// Open one new connection

func (db *DB) openNewConnection() {
    // 在发送到db.openerCh之前，maybeOpenNewConnections之前已经将numOpen++
    // 如果新连接创建失败，或者已经被关闭，则在返回之前必须减1
	ci, err := db.driver.Open(db.dsn)

    // 创建连接的过程加了互斥锁
	db.mu.Lock()
	defer db.mu.Unlock()
	if db.closed {
		if err == nil {
			ci.Close()
		}
		db.numOpen--
		return
	}
	if err != nil {
		db.numOpen--
		db.putConnDBLocked(nil, err)
		db.maybeOpenNewConnections()
		return
	}
	dc := &driverConn{
		db:        db,
		createdAt: nowFunc(),
		ci:        ci,
	}
	if db.putConnDBLocked(dc, err) {
		db.addDepLocked(dc, dc)
	} else {
		db.numOpen--
		ci.Close()
	}
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21

// 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.

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 执行打开连接的操作
    }
}

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116

// SetConnMaxIdleTime sets the maximum amount of time a connection may be idle.
//
// Expired connections may be closed lazily before reuse.
//
// If d <= 0, connections are not closed due to a connection's idle time.
func (db *DB) SetConnMaxIdleTime(d time.Duration) {
	if d < 0 {
		d = 0
	}
	db.mu.Lock()
	defer db.mu.Unlock()

	// Wake cleaner up when idle time is shortened.
	if d > 0 && d < db.maxIdleTime && db.cleanerCh != nil {
		select {
		case db.cleanerCh <- struct{}{}:
		default:
		}
	}
	db.maxIdleTime = d
	db.startCleanerLocked()
}

// startCleanerLocked starts connectionCleaner if needed.
func (db *DB) startCleanerLocked() {
    // 尽管每次putConnDBLocked的时候都会被调用，但注意它的条件db.cleanerCh == nil
    // 这意味着其实db.connectionCleaner只会被启动一次
    if db.maxLifetime > 0 && db.numOpen > 0 && db.cleanerCh == nil {
        db.cleanerCh = make(chan struct{}, 1)
        go db.connectionCleaner(db.maxLifetime)
    }
}
 // 清理过期连接的逻辑
 func (db *DB) connectionCleaner(d time.Duration) {
    const minInterval = time.Second
    // 精度最高为秒级
    if d < minInterval {
        d = minInterval
    }
    t := time.NewTimer(d)
    for {
        // 这个写法很有意思
        select {
        case <-t.C:
        case <-db.cleanerCh: // maxLifetime was changed or db was closed.
        }
        db.mu.Lock()
        d = db.maxLifetime
        if db.closed || db.numOpen == 0 || d <= 0 {
            db.cleanerCh = nil
            db.mu.Unlock()
            return
        }
        expiredSince := nowFunc().Add(-d)
        var closing []*driverConn
        for i := 0; i < len(db.freeConn); i++ {
            c := db.freeConn[i]
            if c.createdAt.Before(expiredSince) {
                closing = append(closing, c)
                last := len(db.freeConn) - 1
                // 数组中经典的删除元素的方法
                db.freeConn[i] = db.freeConn[last]
                db.freeConn[last] = nil
                db.freeConn = db.freeConn[:last]
                // 后退一步是为了可以检查刚刚置换过来的元素
                i--
            }
        }
        db.maxLifetimeClosed += int64(len(closing))
        db.mu.Unlock()
        for _, c := range closing {
            c.Close()
        }
        if d < minInterval {
            d = minInterval
        }
        t.Reset(d)
      }
}

func (db *DB) connectionCleanerRunLocked() (closing []*driverConn) {
	if db.maxLifetime > 0 {
		expiredSince := nowFunc().Add(-db.maxLifetime)
		for i := 0; i < len(db.freeConn); i++ {
			c := db.freeConn[i]
			if c.createdAt.Before(expiredSince) {
				closing = append(closing, c)
				last := len(db.freeConn) - 1
				db.freeConn[i] = db.freeConn[last]
				db.freeConn[last] = nil
				db.freeConn = db.freeConn[:last]
				i--
			}
		}
		db.maxLifetimeClosed += int64(len(closing))
	}

	if db.maxIdleTime > 0 {
		expiredSince := nowFunc().Add(-db.maxIdleTime)
		var expiredCount int64
		for i := 0; i < len(db.freeConn); i++ {
			c := db.freeConn[i]
			if db.maxIdleTime > 0 && c.returnedAt.Before(expiredSince) {
				closing = append(closing, c)
				expiredCount++
				last := len(db.freeConn) - 1
				db.freeConn[i] = db.freeConn[last]
				db.freeConn[last] = nil
				db.freeConn = db.freeConn[:last]
				i--
			}
		}
		db.maxIdleTimeClosed += expiredCount
	}
	return
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20

type Rows struct {
      dc          *driverConn // owned; must call releaseConn when closed to release
      releaseConn func(error)
      rowsi       driver.Rows
      cancel      func()      // called when Rows is closed, may be nil.
      closeStmt   *driverStmt // if non-nil, statement to Close on close

      // closemu prevents Rows from closing while there
      // is an active streaming result. It is held for read during non-close operations
      // and exclusively during close.
      //
      // closemu guards lasterr and closed.
      closemu sync.RWMutex
      closed  bool
      lasterr error // non-nil only if closed is true

      // lastcols is only used in Scan, Next, and NextResultSet which are expected
      // not to be called concurrently.
      lastcols []driver.Value
  }

1
2
3
4
5
6
7
8
9

func (db *DB) query(ctx context.Context, query string, args []interface{}, strategy connReuseStrategy) (*Rows, error) {
	ci, err := db.conn(ctx, strategy)
	if err != nil {
		return nil, err
	}

    // 执行查询
	return db.queryConn(ctx, ci, ci.releaseConn, query, args)
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69

// queryConn 执行查询
// The connection gets released by the releaseConn function.
func (db *DB) queryConn(ctx context.Context, dc *driverConn, releaseConn func(error), query string, args []interface{}) (*Rows, error) {
    // 判断驱动是否实现了Queryer
	if queryer, ok := dc.ci.(driver.Queryer); ok {
		dargs, err := driverArgs(nil, args)
		if err != nil {
			releaseConn(err)
			return nil, err
		}
        var rowsi driver.Rows
        // 调用驱动的查询方法
		withLock(dc, func() {
			rowsi, err = ctxDriverQuery(ctx, queryer, query, dargs)
        })
        // 不带参数的返回
		if err != driver.ErrSkip {
            // 说明用户没有要求用Prepared Statement
			if err != nil {
				releaseConn(err)
				return nil, err
			}
			// Note: ownership of dc passes to the *Rows, to be freed
            // with releaseConn.
            // 注意：这里将驱动返回的行信息rowsi在这里封装成了
            // sql.Rows
			rows := &Rows{
				dc:          dc,
				releaseConn: releaseConn,
				rowsi:       rowsi,
			}
			rows.initContextClose(ctx)
			return rows, nil
		}
	}
    // 如果驱动没有实现Queryer接口或者用户指定要使用Prepared Statement(返回了driver.ErrSkip），则使用Prepared Statement来做查询
	var si driver.Stmt
    var err error
    // 带参数的返回，创建prepare对象
	withLock(dc, func() {
		si, err = ctxDriverPrepare(ctx, dc.ci, query)
	})
	if err != nil {
		releaseConn(err)
		return nil, err
	}

    ds := driverStmt{dc, si}
    // 执行语句
	rowsi, err := rowsiFromStatement(ctx, ds, args...)
	if err != nil {
		withLock(dc, func() {
			si.Close()
		})
		releaseConn(err)
		return nil, err
	}

	// Note: ownership of ci passes to the *Rows, to be freed
	// with releaseConn.
	rows := &Rows{
		dc:          dc,
		releaseConn: releaseConn,
		rowsi:       rowsi,
		closeStmt:   si,     // close的时候，这个也需要关闭
	}
	rows.initContextClose(ctx)
	return rows, nil
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17

func ctxDriverQuery(ctx context.Context, queryerCtx driver.QueryerContext, queryer driver.Queryer, query string, nvdargs []driver.NamedValue) (driver.Rows, error) {
	if queryerCtx != nil {
		return queryerCtx.QueryContext(ctx, query, nvdargs)
	}
	dargs, err := namedValueToValue(nvdargs)
	if err != nil {
		return nil, err
	}
    // 注意这里ctx的用法，根据ctx是否已被取消，提早返回
	select {
	default:
	case <-ctx.Done():
		return nil, ctx.Err()
    }
    // 从这里以后就会到驱动的代码了
	return queryer.Query(query, dargs)
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50

  func (mc *mysqlConn) Query(query string, args []driver.Value) (driver.Rows, error) {
      return mc.query(query, args)
  }

    func (mc *mysqlConn) query(query string, args []driver.Value) (*textRows, error) {
      if mc.closed.IsSet() {
          errLog.Print(ErrInvalidConn)
          return nil, driver.ErrBadConn
      }
      if len(args) != 0 {
          // 是否转义sql中的参数来防止sql注入
          // 如果指定不转义的话会使用 prepared statement
          if !mc.cfg.InterpolateParams {
              return nil, driver.ErrSkip
          }
          // try client-side prepare to reduce roundtrip
          prepared, err := mc.interpolateParams(query, args)
          if err != nil {
              return nil, err
          }
          query = prepared
      }
      // Send command
      err := mc.writeCommandPacketStr(comQuery, query)
      if err == nil {
          // Read Result
          var resLen int
          resLen, err = mc.readResultSetHeaderPacket()
          if err == nil {
              rows := new(textRows)
              rows.mc = mc

              if resLen == 0 {
                  rows.rs.done = true

                  switch err := rows.NextResultSet(); err {
                  case nil, io.EOF:
                      return rows, nil
                  default:
                      return nil, err
                  }
              }

              // Columns
              rows.rs.columns, err = mc.readColumns(resLen)
              return rows, err
          }
      }
      return nil, mc.markBadConn(err)
  }

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15

func ctxDriverPrepare(ctx context.Context, ci driver.Conn, query string) (driver.Stmt, error) {
	if ciCtx, is := ci.(driver.ConnPrepareContext); is {
		return ciCtx.PrepareContext(ctx, query)
	}
	si, err := ci.Prepare(query)
	if err == nil {
		select {
		default:
		case <-ctx.Done():
			si.Close()
			return nil, ctx.Err()
		}
	}
	return si, err
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19

func (mc *mysqlConn) PrepareContext(ctx context.Context, query string) (driver.Stmt, error) {
	if err := mc.watchCancel(ctx); err != nil {
		return nil, err
	}

	stmt, err := mc.Prepare(query)
	mc.finish()
	if err != nil {
		return nil, err
	}

	select {
	default:
	case <-ctx.Done():
		stmt.Close()
		return nil, ctx.Err()
	}
	return stmt, nil
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

func rowsiFromStatement(ds driverStmt, args ...interface{}) (driver.Rows, error) {
 ds.Lock()
 want := ds.si.NumInput()
 ds.Unlock()

 // -1 means the driver doesn't know how to count the number of
 // placeholders, so we won't sanity check input here and instead let the
 // driver deal with errors.
 if want != -1 && len(args) != want {
  return nil, fmt.Errorf("sql: statement expects %d inputs; got %d", want, len(args))
 }

 dargs, err := driverArgs(&ds, args)
 if err != nil {
  return nil, err
 }

 ds.Lock()
 rowsi, err := ds.si.Query(dargs)
 ds.Unlock()
 if err != nil {
  return nil, err
 }
 return rowsi, nil
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37

func (stmt *mysqlStmt) Query(args []driver.Value) (driver.Rows, error) {
 if stmt.mc.netConn == nil {
  errLog.Print(ErrInvalidConn)
  return nil, driver.ErrBadConn
 }

 // Send command
 err := stmt.writeExecutePacket(args)
 if err != nil {
  return nil, err
 }

 mc := stmt.mc

 // Read Result
 resLen, err := mc.readResultSetHeaderPacket()
 if err != nil {
  return nil, err
 }

 rows := new(binaryRows)

 if resLen > 0 {
  rows.mc = mc
  // Columns
  // If not cached, read them and cache them
  if stmt.columns == nil {
   rows.columns, err = mc.readColumns(resLen)
   stmt.columns = rows.columns
  } else {
   rows.columns = stmt.columns
   err = mc.readUntilEOF()
  }
 }

 return rows, err
}