本文共 17826 字，大约阅读时间需要 59 分钟。

前言

后端程序主要是数据驱动型的，因此大多数后端应用也就主要涉及几类和数据相关的操作：读数据，写数据，处理数据。其中，读写是处理的基础。读写操作必然涉及编程语言和各种各样数据库的交互，因此任何后端编程语言都必须提供良好的数据库交互接口方便程序读写数据。本文主要是介绍golang的数据库操作原理。

代码框架

现在生产中使用的数据库多种多样，因此在语言层面实现所有数据库的交互必然会导致编程语言变得无比臃肿。为了避免这个问题，golang通过接口的形式引入数据库驱动，使得我们可以从用户的角度自己编写数据库驱动，并在使用时嵌入到golang中。用go的机制来说，golang只需要提供一个Driver的interface，用户根据自己的数据库编写驱动实现Driver即可。当然，很多时候go官方已经以库的形式提供了常用的数据库操作驱动，我们只需要import即可。

通过以上对go操作数据库的简单阐述，我们大概可以知道知道它具有热插拔，灵活性等优点。为了深入阐述它的实现原理，我们必须对以下问题作出回答：

1. Driver是如何嵌入到go中的？
2. go是如何实现连接池管理的？
   

这里先回答第一个问题。

go语言为所有类型的数据库提供了一个统一的结构体：

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	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.	openerCh          chan struct{}	resetterCh        chan *driverConn	closed            bool	dep               map[finalCloser]depSet	lastPut           map[*driverConn]string // stacktrace of last conn's put; debug only	maxIdle           int                    // zero means defaultMaxIdleConns; negative means 0	maxOpen           int                    // <= 0 means unlimited	maxLifetime       time.Duration          // maximum amount of time a connection may be reused	cleanerCh         chan struct{}	waitCount         int64 // Total number of connections waited for.	maxIdleClosed     int64 // Total number of connections closed due to idle.	maxLifetimeClosed int64 // Total number of connections closed due to max free limit.	stop func() // stop cancels the connection opener and the session resetter.}

从这个DB结构体可以看到，它主要是包含连接池的控制信息。因为说白了，从客户端的角度来看，不同的sql数据库之间之所以不同主要是因为其指令不同，从C/S的角度来看的话，也就是网络传输格式的不同。所以说Driver本质上是实现不同的sql数据库中客户端与服务端的数据传输格式。所以我们可以看到，go语言自己实现不同数据库的连接池管理(因为这具有普遍性)，而各个Driver则简单地只负责按照不同数据库通信要求，实现客户端到服务端的通信编码。因此Driver的实现是比较轻量级的。

DB中的connector 成员就是Driver的插口：

type Connector interface {	// Connect returns a connection to the database.	// Connect 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 provided context.Context is for dialing purposes only	// (see net.DialContext) and should not be stored or used for	// other purposes.	//	// The returned connection is only used by one goroutine at a	// time.	Connect(context.Context) (Conn, error)	// Driver returns the underlying Driver of the Connector,	// mainly to maintain compatibility with the Driver method	// on sql.DB.	Driver() Driver}

它是一个interface类型。其中的connect函数就是connector和Driver的连接桥梁。一般connector都是一个包含Driver的结构体，该connector的connect函数通过调用Driver的Open函数获得连接，放入到DB结构体维护的连接池中

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)}

以dsnconnector为例：

type dsnConnector struct {	dsn    string	driver driver.Driver}func (t dsnConnector) Connect(_ context.Context) (driver.Conn, error) {	return t.driver.Open(t.dsn)}func (t dsnConnector) Driver() driver.Driver {	return t.driver}

我们在调用sql.Open函数创建DB结构体时默认就是DB.connector=dsnconnector。

到此，我们就知道了Driver是如何嵌入到go语言的DB中。

Driver接口

从上一节我们知道了Driver需要给用户程序返回数据库连接。连接反映了不同数据库的交互方式，因此Driver本身还应该提供一条连接具体可完成的操作：

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)}

该接口只显示了连接的三个操作，但是显然不会这么少，比如在连接上执行指令的操作就没有体现出来。

在DB的执行语句中，我们会看到这样的断言：

execerCtx, ok := dc.ci.(driver.ExecerContext)	var execer driver.Execer	if !ok {		execer, ok = dc.ci.(driver.Execer)	}

所以，我们可以简单地认为，Driver的连接上还实现了ExecerContext，Execer等这些接口。这样就可以在连接上执行数据库指令了。

Driver除了对连接抽象之外，还提供了指令操作结果的抽象：

// Result is the result of a query execution.type Result interface {	// LastInsertId returns the database's auto-generated ID	// after, for example, an INSERT into a table with primary	// key.	LastInsertId() (int64, error)	// RowsAffected returns the number of rows affected by the	// query.	RowsAffected() (int64, error)}// Stmt is a prepared statement. It is bound to a Conn and not// used by multiple goroutines concurrently.type Stmt interface {	// Close closes the statement.	//	// As of Go 1.1, a Stmt will not be closed if it's in use	// by any queries.	Close() error	// NumInput returns the number of placeholder parameters.	//	// If NumInput returns >= 0, the sql package will sanity check	// argument counts from callers and return errors to the caller	// before the statement's Exec or Query methods are called.	//	// NumInput may also return -1, if the driver doesn't know	// its number of placeholders. In that case, the sql package	// will not sanity check Exec or Query argument counts.	NumInput() int	// Exec executes a query that doesn't return rows, such	// as an INSERT or UPDATE.	//	// Deprecated: Drivers should implement StmtExecContext instead (or additionally).	Exec(args []Value) (Result, error)	// Query executes a query that may return rows, such as a	// SELECT.	//	// Deprecated: Drivers should implement StmtQueryContext instead (or additionally).	Query(args []Value) (Rows, error)}type Rows interface {	// Columns returns the names of the columns. The number of	// columns of the result is inferred from the length of the	// slice. If a particular column name isn't known, an empty	// string should be returned for that entry.	Columns() []string	// Close closes the rows iterator.	Close() error	// Next is called to populate the next row of data into	// the provided slice. The provided slice will be the same	// size as the Columns() are wide.	//	// Next should return io.EOF when there are no more rows.	//	// The dest should not be written to outside of Next. Care	// should be taken when closing Rows not to modify	// a buffer held in dest.	Next(dest []Value) error}

这些都是Conn执行接口函数的返回值接口。这些接口提供了对返回值的抽象，比如Rows接口提供迭代器抽象，方便对数据库查询结果的遍历。

总结一下Driver的实现中的两个重要的点：

1. 实现数据库的连接接口(Conn)，该接口抽象出了驱动使用者可以对数据库进行的操作
2. 抽象了数据库连接接口的返回值(Result,Rows)，并根据返回值属性提供对返回值的操作接口。

后面我们在解析mysql驱动实现时再详细深入了解这两点。

连接池管理策略

database/sql的实现逻辑很简单，通过Driver获取连接，然后调用连接实现的接口执行数据库操作指令。其中第二部完全由驱动实现，第一步涉及优化问题，显然不能每次执行一次数据库操作都去获取一条新的连接，因此需要在database/sql层设计一个连接缓冲池。连接缓冲池的管理也是database/sql的核心。

我们从func (db *DB) conn(ctx context.Context, strategy connReuseStrategy) (*driverConn, error) 函数入手。conn函数会从连接池获取或者新建一条连接。DB每次执行指令时都会调用这个函数获取连接。

该函数在获取连接时会有三个逻辑：

1. 首先尝试从连接池中获取空闲连接。如果连接池中没有空闲连接，走2
2. 查看当前连接缓冲中开的连接是否已经超过最大允许的连接数。如果不超过，则新建连接，否则，走3
3. 如果当前连接数已经超过最大限制数，则等待其他操作处理完返回连接再利用。

下面分别根据代码看一下这三个获取连接的步骤。

1.

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

if db.maxOpen > 0 && db.numOpen >= db.maxOpen { //正常情况下不会出现 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的请求列表中		db.connRequests[reqKey] = req		db.waitCount++		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			}			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		}	}

针对第二个分支，我们对应看一下释放连接的函数是如何响应这个connRequest的：

//func (dc *driverConn) releaseConn(err error) => func (db *DB) putConn(dc *driverConn, err error, resetSession bool)  =>  func (db *DB) putConnDBLocked(dc *driverConn, err error) boolfunc (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 && db.maxIdleConnsLocked() > len(db.freeConn) { //否则放到free poo里		db.freeConn = append(db.freeConn, dc)		db.maxIdleClosed++		db.startCleanerLocked()		return true	}	return false}

可以看到，释放一条连接，要么将连接返回给请求列表中的请求，要么放在空闲连接池中。

在DB结构体中还有一个openerCh成员：

openerCh          chan struct{}

该成员是为了单独起一个协程开连接做通信用的：

func (db *DB) connectionOpener(ctx context.Context) {	for {		select {		case <-ctx.Done():			return		case <-db.openerCh:			db.openNewConnection(ctx)		}	}}

相关用户接口实现

本小节介绍一下database/sql提供给用户使用的而一些接口的实现细节，这些接口包括：

func (db *DB) Prepare(query string) (*Stmt, error)func (db *DB) Exec(query string, args ...interface{}) (Result, error)func (db *DB) Query(query string, args ...interface{}) (*Rows, error)func (db *DB) QueryRow(query string, args ...interface{}) *Row

这几个接口真正的实现者是：

func (db *DB) PrepareContext(ctx context.Context, query string) (*Stmt, error)func (db *DB) ExecContext(ctx context.Context, query string, args ...interface{}) (Result, error)func (db *DB) QueryContext(ctx context.Context, query string, args ...interface{}) (*Rows, error)

核心功能实现者：

func (db *DB) prepare(ctx context.Context, query string, strategy connReuseStrategy) (*Stmt, error)func (db *DB) exec(ctx context.Context, query string, args []interface{}, strategy connReuseStrategy) (Result, error)func (db *DB) query(ctx context.Context, query string, args []interface{}, strategy connReuseStrategy) (*Rows, error)

然后最终落实到：

func (db *DB) prepareDC(ctx context.Context, dc *driverConn, release func(error), cg stmtConnGrabber, query string) (*Stmt, error)func (db *DB) execDC(ctx context.Context, dc *driverConn, release func(error), query string, args []interface{}) (res Result, err errorfunc (db *DB) queryDC(ctx, txctx context.Context, dc *driverConn, releaseConn func(error), query string, args []interface{}) (*Rows, error)

*DC函数主要是通过调用driverConn实现的函数来执行数据库操作指令，当然driverConn也是继续调用Driver.Conn的实现的函数，这就和具体的驱动实现有关了，这里我们只关心到driverConn。事实上，这里会对应调用Driver.Conn的下列函数：

type ConnPrepareContext interface {	// PrepareContext returns a prepared statement, bound to this connection.	// context is for the preparation of the statement,	// it must not store the context within the statement itself.	PrepareContext(ctx context.Context, query string) (Stmt, error)}type ExecerContext interface {	ExecContext(ctx context.Context, query string, args []NamedValue) (Result, error)}type QueryerContext interface {	QueryContext(ctx context.Context, query string, args []NamedValue) (Rows, error)}

Driver.Conn会实现上述的接口。(prepare比较特殊，因为Driver.Conn本身就有一个Prepare函数，所以可以选择不实现ConnPrepareContext)：

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)	}	//如果Driver.Conn没有实现ConnPrepareContext，则调用Prepare	si, err := ci.Prepare(query) //Driver.Conn已经实现了Preapare接口	if err == nil {		select {		default:		case <-ctx.Done():			si.Close()			return nil, ctx.Err()		}	}	return si, err}

下面简单看一下这几个*DC函数的实现。

DC函数的逻辑主要是两个：

1. 在连接上执行对应指令
2. 释放连接

不过queryDC有些特殊。queryDC在执行查询语句之后并不会立刻释放连接，而是只有在rows.close之后才会释放。这主要是因为rows是个迭代器结构，后面还会依赖这个连接遍历数据库数据，所以不能释放连接。

rows

rows是Query接口的返回，它是一个迭代器抽象，可以通过rows.Next遍历查询操作返回的所有结果行。

rows依赖于它的子成员rowsi，它是driver.Rows类型。rows的Next函数依赖于rowsi的Next函数来完成：

//func (rs *Rows) nextLocked() (doClose, ok bool)rs.lasterr = rs.rowsi.Next(rs.lastcols)

在使用rows时，我们并不一定都需要close它：

func (rs *Rows) Next() bool {	var doClose, ok bool	withLock(rs.closemu.RLocker(), func() {		doClose, ok = rs.nextLocked()	})	if doClose {		rs.Close()	}	return ok}

当nextLocked函数返回EOF或其他类型的错误时，就会自动close调这个rows，而rows.close会release调他所占用的连接。

//func (rs *Rows) close(err error) error....rs.releaseConn(err)

Stmt

Prepare会返回一个stmt，表示一个被prepared的语句。然后就可以提供具体参数，调用这个stmt的Exec执行。需要注意的是，Prepare完成之后该Prepare使用的连接机会被回收，而不会等到它返回的stmt执行close才被回收。也就是说，即使prepare了stmt，但是不用，也不会使连接空转，它会被返回到连接池中重用。

当stmt要执行时，它会去连接池中找连接。 这里需要注意的是，每个prepare statement的id只会被执行该prepare语句的连接识别。也就是说每个prepare statement只和唯一一个连接绑定，不能在一条连接上prepare，而在另一条连接上Exec()。

golang的stmt结构体提供了一个css结构来记录prepare stmt和connect之间的绑定关系：

type Stmt struct {    ...	// css is a list of underlying driver statement interfaces	// that are valid on particular connections. This is only	// used if cg == nil and one is found that has idle	// connections. If cg != nil, cgds is always used.	css []connStmt	....}

如上图，css记录了query语句prepare得到的statement和connection关系。当stmt执行Exec时，就会尝试寻找该stmt执行prepare使用的连接，如果该连接正忙，则会重新找一条连接再prepare一遍，然后再Exec，并把对应的新连接和新prepare的statement记录到css中。

func (s *Stmt) connStmt(ctx context.Context, strategy connReuseStrategy) (dc *driverConn, releaseConn func(error), ds *driverStmt, err error) {	if err = s.stickyErr; err != nil {		return	}	s.mu.Lock()	if s.closed {		s.mu.Unlock()		err = errors.New("sql: statement is closed")		return	}	// In a transaction or connection, we always use the connection that the	// stmt was created on.	if s.cg != nil {		s.mu.Unlock()		dc, releaseConn, err = s.cg.grabConn(ctx) // blocks, waiting for the connection.		if err != nil {			return		}		return dc, releaseConn, s.cgds, nil	}	s.removeClosedStmtLocked() //这里把所有close的连接从css中删除掉，主要是为了优化遍历s.css的速度	s.mu.Unlock()	//为什么不直接先在css里面尝试寻找，没有的话，再从全局的空闲连接池中找。(css中的连接无法判断是否是空闲？？不是有一个inUse字段的吗.从复杂度方面考虑一下)	dc, err = s.db.conn(ctx, strategy)//并不会优先获得执行过prepare语句的connection	if err != nil {		return nil, nil, nil, err	}	s.mu.Lock()	for _, v := range s.css { //		if v.dc == dc { //检查从空闲连接池获得的是不是该statement执行prepare指令的连接			s.mu.Unlock()			return dc, dc.releaseConn, v.ds, nil		}	}	s.mu.Unlock()	//如果不是，则需要重新在这条新的连接上prepare这个statement	// No luck; we need to prepare the statement on this connection	withLock(dc, func() {		ds, err = s.prepareOnConnLocked(ctx, dc)	})	if err != nil {		dc.releaseConn(err)		return nil, nil, nil, err	}	return dc, dc.releaseConn, ds, nil}

因此可以看到在高并发的情况下，由于连接繁忙，会出现大量的reprepare，导致性能低下。而且会导致相同的语句在多条连接上prepare，而连接上的statement只有在连接释放时才会调用statement.close()，这样对于M条prepare语句(可能相同)，freeConn有N条连接的情况，最坏会有N*M个prepared statement在数据库服务器端，如果这个值大于max_prepared_stmt_count，就会引发数据库操作错误了。所以在使用时一定要记得对stmt进行close操作。

statement.close操作：

func (s *Stmt) finalClose() error {	s.mu.Lock()	defer s.mu.Unlock()	if s.css != nil {		for _, v := range s.css {			s.db.noteUnusedDriverStatement(v.dc, v.ds)			v.dc.removeOpenStmt(v.ds)		}		s.css = nil	}	return nil}func (db *DB) noteUnusedDriverStatement(c *driverConn, ds *driverStmt) {	db.mu.Lock()	defer db.mu.Unlock()	if c.inUse { //只有在当前连接不在使用时才会close掉这个statement		c.onPut = append(c.onPut, func() {			ds.Close()		})	} else {		c.Lock()		fc := c.finalClosed		c.Unlock()		if !fc {			ds.Close()		}	}}

mysql驱动实现细节

驱动主要是实现和数据库服务端的通信部分功能，包括收发包的缓冲区管理，通信编码等内容。和实际数据库相关性比较大，比如不同数据库服务器对指令和执行结果的传输编解码比较不一样，需要阅读开发文档。

对比前几节的内容，简单来说，驱动主要是实现了driver.Conn，rows，stmt，当然还有transaction，但是本文没有涉及事务，所以这里略过。所有这些核心就是通过将上层传过来的指令字符串编码传输到服务端，以及通过封装通信细节使得返回的rows支持迭代器功能，以及将conn封装在stmt中，使stmt支持执行指令的功能。

总结

本文介绍了golang中database/sql的实现细节。通过自上而下的方式，首先从DB给用户暴露的常用基础接口：Prepare，Query，Exec等的执行逻辑开始，然后介绍这些接口中使用到的一些基础组件，主要包括：Conn，Rows，Stmt，以及对应的Driver层面的Driver.Conn，Driver.Stmt，Driver.Rows的实现，通过逐层剖析，我们看到了driver和database/sql是如何协同工作的。除此之外，我们还解析了database/sql的连接池的实现细节。最后我们简单阐述了一下driver的实现逻辑。

在底层来说，用户层面所谓的执行sql语句其实是将用户提供的sql语句串编码传输到数据库服务器端。所以本质上还是一个C/S的程序架构，只不过由于数据库多种多样，因此抽离出编码细节，将数据库操作分成database/sql和driver两层逻辑。database/sql负责提供用户接口，以及一些不涉及具体数据库的实现，比如连接池管理等；Driver层负责数据库通信。

转载地址：http://opqxi.baihongyu.com/