Merge pull request #191 from funnyAnt/master

#189 连接池相关优化

backend/connection_pool.go

@@ -26,6 +26,7 @@ import (
 
 const (
 	getConnTimeout = 2 * time.Second
+	PING_PEROID    = 5 * time.Second
 )
 
 var (
@@ -114,7 +115,7 @@ func (cp *connectionPoolImpl) tryReuse(pc *pooledConnectImpl) error {
 }
 
 // Get return a connection, you should call PooledConnect's Recycle once done
-func (cp *connectionPoolImpl) Get(ctx context.Context) (PooledConnect, error) {
+func (cp *connectionPoolImpl) Get(ctx context.Context) (pc PooledConnect, err error) {
 	p := cp.pool()
 	if p == nil {
 		return nil, ErrConnectionPoolClosed
@@ -126,7 +127,16 @@ func (cp *connectionPoolImpl) Get(ctx context.Context) (PooledConnect, error) {
 	if err != nil {
 		return nil, err
 	}
-	return r.(*pooledConnectImpl), nil
+
+	pc = r.(*pooledConnectImpl)
+
+	//do ping when over the ping time. if error happen, create new one
+	if !pc.GetReturnTime().IsZero() && time.Until(pc.GetReturnTime().Add(PING_PEROID)) < 0 { 
+		if err = pc.Ping(); err != nil {
+			err = pc.Reconnect()
+		}
+	}
+	return pc, err
 }
 
 // Put recycle a connection into the pool
@@ -258,4 +268,4 @@ func (cp *connectionPoolImpl) IdleClosed() int64 {
 		return 0
 	}
 	return p.IdleClosed()
-}
+}

backend/interface.go

@@ -18,12 +18,14 @@ type PooledConnect interface {
 	Begin() error
 	Commit() error
 	Rollback() error
+	Ping() error
 	SetCharset(charset string, collation mysql.CollationID) (bool, error)
 	FieldList(table string, wildcard string) ([]*mysql.Field, error)
 	GetAddr() string
 	SetSessionVariables(frontend *mysql.SessionVariables) (bool, error)
 	WriteSetStatement() error
 	GetConnectionID() int64
+	GetReturnTime() time.Time
 }
 
 type ConnectionPool interface {

backend/pooled_connection.go

@@ -15,21 +15,30 @@
 package backend
 
 import (
+	"time"
+
 	"github.com/XiaoMi/Gaea/mysql"
 )
 
 // PooledConnect app use this object to exec sql
 type pooledConnectImpl struct {
 	directConnection *DirectConnection
 	pool             *connectionPoolImpl
+	returnTime       time.Time
 }
 
 // Recycle return PooledConnect to the pool
 func (pc *pooledConnectImpl) Recycle() {
+	//if has error,the connection can’t be recycled
+	if pc.directConnection.pkgErr != nil {
+		pc.Close()
+	}
+
 	if pc.IsClosed() {
 		pc.pool.Put(nil)
 	} else {
 		pc.pool.Put(pc)
+		pc.returnTime = time.Now()
 	}
 }
 
@@ -63,6 +72,9 @@ func (pc *pooledConnectImpl) UseDB(db string) error {
 	return pc.directConnection.UseDB(db)
 }
 
+func (pc *pooledConnectImpl) Ping() error {
+	return pc.directConnection.Ping()
+}
 // Execute wrapper of direct connection, execute sql
 func (pc *pooledConnectImpl) Execute(sql string, maxRows int) (*mysql.Result, error) {
 	return pc.directConnection.Execute(sql, maxRows)
@@ -116,3 +128,7 @@ func (pc *pooledConnectImpl) WriteSetStatement() error {
 func (pc *pooledConnectImpl) GetConnectionID() int64 {
 	return int64(pc.directConnection.conn.ConnectionID)
 }
+
+func (pc *pooledConnectImpl) GetReturnTime() time.Time {
+	return pc.returnTime
+}