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

Can’t create more than max_prepared_stmt_count statements (current value: 16382)

• 打开mysql的general_log开关

show variables where Variable_name like "general_log%";set global general_log=on;

这个开关打开可以记录mysql执行过的所有的sql语句，便于观察。

• 测试用例

func main() {    var e error    DB, e := sql.Open("mysql", "[use]:[passwd]@tcp(127.0.0.1:3306)/[db]")    if e != nil {        panic(e)    }    DB.SetMaxOpenConns(5)    DB.SetMaxIdleConns(1)    DB.Ping()    stm, e := DB.Prepare("select * from test where id = ?")    if e != nil {        return    }    for i:=0; i<10000;i++ {        go func() {            result, err := stm.Exec(i)            if err != nil {            }            fmt.Println(result)        }()    }    <- time.After(time.Duration(5 * time.Second))    fmt.Printf("++++++++++++++++++++==")}

• 打开mysql日志文件

show variables where Variable_name like "general_log%";

上述语句可以看到文件位置。

tail -f /var/lib/mysql/xxx.log | grep Pre

这里只查看Prepare命令的语句。

• 执行上面的测试用例
  发现会执行最大连接数个Prepare。

2018-05-07T11:22:04.294438Z	 5529 Prepare	select * from test where id = ?2018-05-07T11:22:04.335158Z	 5530 Prepare	select * from test where id = ?2018-05-07T11:22:04.364443Z	 5531 Prepare	select * from test where id = ?2018-05-07T11:22:04.364444Z	 5533 Prepare	select * from test where id = ?2018-05-07T11:22:04.364489Z	 5532 Prepare	select * from test where id = ?

• 问题分析

  经过上面的测试发现问题，如果并发量特别大的时候，同一个sql语句会多次执行Prepare，这样就存在一个问题，mysql中记录Prepared语句的参数Prepared_stmt_count会一直增大直到达到最大值max_prepared_stmt_count，最后就会报本文第一行的错误。

• 源码分析

  这里贴几个比较重要的方法出来分析一下。

// Prepare creates a prepared statement for later queries or executions.// Multiple queries or executions may be run concurrently from the// returned statement.// The caller must call the statement's Close method// when the statement is no longer needed.// stm, e := DB.Prepare("select * from tbl_str_index_512 where dataid = ?")// 调用此方法预编译sqlfunc (db *DB) Prepare(query string) (*Stmt, error) {	var stmt *Stmt	var err error	for i := 0; i < maxBadConnRetries; i++ {		stmt, err = db.prepare(query, cachedOrNewConn)		if err != driver.ErrBadConn {			break		}	}	if err == driver.ErrBadConn {		return db.prepare(query, alwaysNewConn)	}	return stmt, err}func (db *DB) prepare(query string, strategy connReuseStrategy) (*Stmt, error) {	// TODO: check if db.driver supports an optional	// driver.Preparer interface and call that instead, if so,	// otherwise we make a prepared statement that's bound	// to a connection, and to execute this prepared statement	// we either need to use this connection (if it's free), else	// get a new connection + re-prepare + execute on that one.	dc, err := db.conn(strategy)  // 获取数据库连接（后面当成锁使用）	if err != nil {		return nil, err	}	dc.Lock()	si, err := dc.prepareLocked(query) // 这里是真正prepare的地方，并返回真正的stmt	dc.Unlock()	if err != nil {		db.putConn(dc, err)		return nil, err	}	stmt := &Stmt{		db:            db,		query:         query,		css:           []connStmt{
  {dc, si}}, // 把dc和si缓存到stmt中		lastNumClosed: atomic.LoadUint64(&db.numClosed),	}	db.addDep(stmt, stmt)	db.putConn(dc, nil)	return stmt, nil}func (dc *driverConn) prepareLocked(query string) (driver.Stmt, error) {	si, err := dc.ci.Prepare(query)  // 真正prepared的地方，返回预编译好的stmt	if err == nil {		// Track each driverConn's open statements, so we can close them		// before closing the conn.		//		// TODO(bradfitz): let drivers opt out of caring about		// stmt closes if the conn is about to close anyway? For now		// do the safe thing, in case stmts need to be closed.		//		// TODO(bradfitz): after Go 1.2, closing driver.Stmts		// should be moved to driverStmt, using unique		// *driverStmts everywhere (including from		// *Stmt.connStmt, instead of returning a		// driver.Stmt), using driverStmt as a pointer		// everywhere, and making it a finalCloser.		if dc.openStmt == nil {			dc.openStmt = make(map[driver.Stmt]bool)		}		dc.openStmt[si] = true	}	return si, err}// Exec executes a prepared statement with the given arguments and// returns a Result summarizing the effect of the statement.// result, err := stm.Exec(i)时调用的方法func (s *Stmt) Exec(args ...interface{}) (Result, error) {	s.closemu.RLock()	defer s.closemu.RUnlock()	var res Result	for i := 0; i < maxBadConnRetries; i++ {		dc, releaseConn, si, err := s.connStmt()  // 获取数据库连接		if err != nil {			if err == driver.ErrBadConn {				continue			}			return nil, err		}		res, err = resultFromStatement(driverStmt{dc, si}, args...)		releaseConn(err)		if err != driver.ErrBadConn {			return res, err		}	}	return nil, driver.ErrBadConn}// connStmt returns a free driver connection on which to execute the// statement, a function to call to release the connection, and a// statement bound to that connection.func (s *Stmt) connStmt() (ci *driverConn, releaseConn func(error), si driver.Stmt, 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, we always use the connection that the	// transaction was created on.	if s.tx != nil {		s.mu.Unlock()		ci, err = s.tx.grabConn() // blocks, waiting for the connection.		if err != nil {			return		}		releaseConn = func(error) {}		return ci, releaseConn, s.txsi.si, nil	}	s.removeClosedStmtLocked()	s.mu.Unlock()	// TODO(bradfitz): or always wait for one? make configurable later?	dc, err := s.db.conn(cachedOrNewConn)  // 获取数据库连接，注意参数	if err != nil {		return nil, nil, nil, err	}	s.mu.Lock()	for _, v := range s.css {  // 把获得的数据库连接跟缓存中的对比，找到了就直接返回（bug就在这里）		if v.dc == dc {  // 简单地比较数据库连接			s.mu.Unlock()			return dc, dc.releaseConn, v.si, nil		}	}	s.mu.Unlock()	// No luck; we need to prepare the statement on this connection	dc.Lock()	si, err = dc.prepareLocked(s.query)  // 如果缓存中没有，又执行了一次prepared（bug就在这里）	dc.Unlock()	if err != nil {		s.db.putConn(dc, err)		return nil, nil, nil, err	}	s.mu.Lock()	cs := connStmt{dc, si}	s.css = append(s.css, cs)  // 把新的连接也放到缓存中	s.mu.Unlock()	return dc, dc.releaseConn, si, nil}// conn returns a newly-opened or cached *driverConn.func (db *DB) conn(strategy connReuseStrategy) (*driverConn, error) {	db.mu.Lock()	if db.closed {		db.mu.Unlock()		return nil, errDBClosed	}	lifetime := db.maxLifetime	// Prefer a free connection, if possible.	numFree := len(db.freeConn)	 // 如果策略为cachedOrNewConn且有空闲连接，就从空闲连接中取一个（这段本身没有问题，但跟上面的逻辑结合起来就有问题了）	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	}	// 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)		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++ // optimistically	db.mu.Unlock()	ci, err := db.driver.Open(db.dsn)	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,	}	db.addDepLocked(dc, dc)	dc.inUse = true	db.mu.Unlock()	return dc, nil}

重点关注下connStmt()这个方法中的这几段：

// TODO(bradfitz): or always wait for one? make configurable later?	dc, err := s.db.conn(cachedOrNewConn)  // 获取数据库连接，注意参数	if err != nil {		return nil, nil, nil, err	}	s.mu.Lock()	for _, v := range s.css {  // 把获得的数据库连接跟缓存中的对比，找到了就直接返回（bug就在这里）		if v.dc == dc {  // 简单地比较数据库连接			s.mu.Unlock()			return dc, dc.releaseConn, v.si, nil		}	}	s.mu.Unlock()	// No luck; we need to prepare the statement on this connection	dc.Lock()	si, err = dc.prepareLocked(s.query)  // 如果缓存中没有，又执行了一次prepared（bug就在这里）	dc.Unlock()	if err != nil {		s.db.putConn(dc, err)		return nil, nil, nil, err	}

前面Prepare()的时候已经执行了一次prepareLocked()（真正prepare的地方）了并缓存了数据库连接，后面Exec()的时候会先取一个数据库连接与缓存中的比较，如果有则直接执行，这段逻辑没有问题。如果获取的数据库连接跟缓存中的不一致，又会执行一下prepareLocked()方法，相当于又Prepared了一次，这样在高并发的时候同一个sql语句多次执行的时候就会出现多次Prepared，最终导致mysql参数prepared_stmt_count不断增加最后达到最大值max_prepared_stmt_count，然后再有新的Prepared进来就报本文第一行的错误了。

• mysql中语句执行过程分析
  mysql中一条语句的执行是要先Prepared再Execute的，而且这个Prepare只在会话期间有效，也就是一条sql语句的执行在每个数据库连接都要先Prepare一下，我们可以通过msql的客户端来做下实验：
  打开一个mysql客户端：

mysql> prepare stmt from 'select ? * ?';Query OK, 0 rows affected (0.04 sec)Statement preparedmysql> set @a = 2;Query OK, 0 rows affected (0.00 sec)mysql> set @b = 3;Query OK, 0 rows affected (0.00 sec)mysql> execute stmt using @a, @b;+-------+| ? * ? |+-------+|     6 |+-------+1 row in set (0.00 sec)

再打开另一个mysql客户端：

mysql> set @a=2;Query OK, 0 rows affected (0.00 sec)mysql> set @b=6;Query OK, 0 rows affected (0.00 sec)mysql> execute stmt using @a, @b;ERROR 1243 (HY000): Unknown prepared statement handler (stmt) given to EXECUTE

可以看到在前面的客户端中Prepare的语句不能在后面的客户端中执行，所以每条sql语句在每个数据库连接（会话）中都要先Prepare。

• 解决方案
  目前来看，解决方案只能增大mysql的max_prepared_stmt_count参数。

set global max_prepared_stmt_count=65536;

欢迎关注我的公众号“彤哥读源码”，查看更多“源码&架构&算法”系列文章, 与彤哥一起畅游源码的海洋。

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